Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

pylint fixes

  • Loading branch information...
commit ea3da86749c8bdb6d51cdcb766375ae0919cf4c7 1 parent 9cc4049
Piotr Zalewa zalun authored
34 example/commands.py
View
@@ -1,11 +1,7 @@
-import time
-import requests
+"""Commands to run on the Marketplace API
+"""
import sys
-
import json
-import oauth2 as oauth
-
-import config
def validate_manifest(auth, manifest_url):
@@ -62,24 +58,24 @@ def update(auth, app_id):
# obtaining current data
data = json.loads(auth.status(app_id).content)
data['payment_type'] = data['premium_type']
- for k in data.keys():
- if k not in editable_keys:
- del data[k]
+ for key in data.keys():
+ if key not in editable_keys:
+ del data[key]
sys.stderr.write('Please provide data, hit Enter for no change\n')
- def get_value(k, v):
- variable = raw_input('%s (%s): ' % (k, v))
- if k in truthy_keys and not variable and not v:
+ def get_value(key, val):
+ variable = raw_input('%s (%s): ' % (key, val))
+ if key in truthy_keys and not variable and not val:
sys.stdout.write('This parameter is required.\n')
- variable = get_value(k, v)
+ variable = get_value(key, val)
return variable
- for k, v in data.items():
- if k in editable_keys:
- variable = get_value(k, v)
+ for key, val in data.items():
+ if key in editable_keys:
+ variable = get_value(key, val)
if variable != '':
- if isinstance(v, list):
- data[k] = variable.split(',')
+ if isinstance(val, list):
+ data[key] = variable.split(',')
else:
- data[k] = variable
+ data[key] = variable
response = auth.update(app_id, data)
if response.status_code != 202:
return {'success': False,
4 example/config.py
View
@@ -1,5 +1,5 @@
-import oauth2 as oauth
-
+"""Read config from environment variables
+"""
from os import environ
CONSUMER_KEY = (environ['CONSUMER_KEY']
57 example/main.py
View
@@ -2,11 +2,11 @@
import sys
import commands
-import config as config
+import config
import marketplace
-commands = {'validate_manifest': commands.validate_manifest,
+COMMANDS = {'validate_manifest': commands.validate_manifest,
'is_manifest_valid': commands.is_manifest_valid,
'create': commands.create,
'status': commands.status,
@@ -16,27 +16,32 @@
'del_screenshot': commands.del_screenshot,
'get_categories': commands.get_categories}
-parser = argparse.ArgumentParser(description='Command line Marketplace client')
-parser.add_argument('method', type=str, help='command to be run on arguments',
- choices=commands.keys())
-parser.add_argument('attrs', metavar='attr', type=str, nargs='*',
- help='command arguments')
-args = parser.parse_args()
-
-client = marketplace.Client(
- domain=config.MARKETPLACE_DOMAIN,
- protocol=config.MARKETPLACE_PROTOCOL,
- port=config.MARKETPLACE_PORT,
- consumer_key=config.CONSUMER_KEY,
- consumer_secret=config.CONSUMER_SECRET)
-
-if args.attrs:
- result = commands[args.method](client, *args.attrs)
-else:
- result = commands[args.method](client)
-
-if result['success']:
- sys.stdout.write('%s\n' % result['message'])
-else:
- sys.stderr.write('%s\n' % result['message'])
- sys.exit(1)
+def main():
+ parser = argparse.ArgumentParser(
+ description='Command line Marketplace client')
+ parser.add_argument('method', type=str,
+ help='command to be run on arguments', choices=COMMANDS.keys())
+ parser.add_argument('attrs', metavar='attr', type=str, nargs='*',
+ help='command arguments')
+ args = parser.parse_args()
+
+ client = marketplace.Client(
+ domain=config.MARKETPLACE_DOMAIN,
+ protocol=config.MARKETPLACE_PROTOCOL,
+ port=config.MARKETPLACE_PORT,
+ consumer_key=config.CONSUMER_KEY,
+ consumer_secret=config.CONSUMER_SECRET)
+
+ if args.attrs:
+ result = COMMANDS[args.method](client, *args.attrs)
+ else:
+ result = COMMANDS[args.method](client)
+
+ if result['success']:
+ sys.stdout.write('%s\n' % result['message'])
+ else:
+ sys.stderr.write('%s\n' % result['message'])
+ sys.exit(1)
+
+if __name__ == "__main__":
+ main()
3  marketplace/client.py
View
@@ -28,8 +28,7 @@
'app': '/apps/app/%s/',
'create_screenshot': '/apps/preview/?app=%s',
'screenshot': '/apps/preview/%s/',
- 'categories': '/apps/category/',
-}
+ 'categories': '/apps/category/'}
class Client:
217 marketplace/tests/testClient.py
View
@@ -1,217 +0,0 @@
-"""
-tests.testClient
-----------------
-"""
-import json
-import logging
-import os
-import unittest
-
-from base64 import b64encode
-
-import httplib2
-import requests
-
-from mock import Mock
-from nose import SkipTest
-from nose.tools import eq_
-
-import marketplace
-
-log = logging.getLogger('test.%s' % __name__)
-
-# Preparing to mock the requests
-OLD_POST = requests.post
-OLD_PUT = requests.put
-OLD_GET = requests.get
-OLD_DELETE = requests.delete
-
-MARKETPLACE_PORT = 443
-MARKETPLACE_DOMAIN = 'marketplace-dev.allizom.org'
-MARKETPLACE_PROTOCOL = 'https'
-
-class Response:
- def __init__(self, status_code, content=None):
- self.status_code = status_code
- self.content = content
-
-class TestClient(unittest.TestCase):
-
- def setUp(self):
- self.marketplace = marketplace.Client(
- domain=MARKETPLACE_DOMAIN,
- port=MARKETPLACE_PORT,
- protocol=MARKETPLACE_PROTOCOL,
- consumer_key='consumer_key',
- consumer_secret='consumer_secret')
-
- def tearDown(self):
- requests.post = OLD_POST
- requests.put = OLD_PUT
- requests.get = OLD_GET
- requests.delete = OLD_DELETE
-
- def test_init(self):
- eq_(self.marketplace.domain, MARKETPLACE_DOMAIN)
- eq_(self.marketplace.port, MARKETPLACE_PORT)
- eq_(self.marketplace.protocol, MARKETPLACE_PROTOCOL)
-
- def test_order_validation(self):
- manifest_url = 'http://example.com/'
- resp = {'id': 'abcd',
- 'manifest': manifest_url,
- 'processed': False,
- 'resource_uri': '/en-US/api/apps/validation/123/',
- 'valid': False,
- 'validation': ''}
-
- requests.post = Mock(return_value=Response(201, resp))
-
- response = self.marketplace.validate_manifest(manifest_url)
- log.debug(response.content)
- log.debug(response.status_code)
- eq_(response.content['id'], 'abcd')
-
- def test_get_validation_result(self):
- resp = {'id': 'abcd',
- 'processed': False,
- 'resource_uri': '/en-US/api/apps/validation/123/',
- 'valid': False,
- 'validation': ''}
-
- requests.get = Mock(return_value=Response(201, json.dumps(resp)))
-
- # providing id by string
- response = self.marketplace.get_manifest_validation_result('abcd')
- content = json.loads(response.content)
- eq_(content['id'], 'abcd')
-
- def test_is_manifest_valid(self):
- # not processed
- resp = {'id': '123',
- 'processed': False,
- 'resource_uri': '/en-US/api/apps/validation/123/',
- 'valid': False,
- 'validation': ''}
-
- requests.get = Mock(return_value=Response(200, json.dumps(resp)))
- eq_(self.marketplace.is_manifest_valid(123), None)
-
- # valid
- resp = {'id': '123',
- 'processed': True,
- 'resource_uri': '/en-US/api/apps/validation/123/',
- 'valid': True,
- 'validation': ''}
-
- requests.get = Mock(return_value=Response(200, json.dumps(resp)))
- eq_(self.marketplace.is_manifest_valid(123), True)
-
- # invalid
- resp = {'id': 'abcd',
- 'processed': True,
- 'resource_uri': '/en-US/api/apps/validation/123/',
- 'valid': False,
- 'validation': {'validation': 'object'}}
-
- requests.get = Mock(return_value=Response(200, json.dumps(resp)))
- eq_(self.marketplace.is_manifest_valid(123), resp['validation'])
-
- def test_create(self):
- resp = {'categories': [],
- 'description': None,
- 'device_types': [],
- 'homepage': None,
- 'id': 1,
- 'manifest': '0a650e5e4c434b5cb60c5495c0d88a89',
- 'name': 'MozillaBall',
- 'premium_type': 'free',
- 'privacy_policy': None,
- 'resource_uri': '/en-US/api/apps/app/1/',
- 'slug': 'mozillaball',
- 'status': 0,
- 'summary': 'Exciting Open Web development action!',
- 'support_email': None,
- 'support_url': None}
-
- requests.post = Mock(return_value=Response(201, json.dumps(resp)))
- response = self.marketplace.create('abcd')
- content = json.loads(response.content)
- eq_(response.status_code, 201)
- eq_(content['manifest'], '0a650e5e4c434b5cb60c5495c0d88a89')
-
- def test_update(self):
- requests.put = Mock(return_value=Response(202, ''))
- response = self.marketplace.update(123, {
- 'name': 'MozillaBall',
- 'summary': 'Changed Summary',
- 'categories': ['Business', 'Game'],
- 'support_email': 'john@doe.com',
- 'device_types': ['phone',],
- 'privacy_policy': 'any',
- 'payment_type': 'free'})
- eq_(response.status_code, 202)
- assert not response.content
-
- def test_status(self):
- resp = {'categories': ['Business', 'Game'],
- 'description': None,
- 'device_types': ['phone',],
- 'homepage': None,
- 'id': 1,
- 'manifest': '0a650e5e4c434b5cb60c5495c0d88a89',
- 'name': 'MozillaBall',
- 'premium_type': 'free',
- 'privacy_policy': None,
- 'resource_uri': '/en-US/api/apps/app/1/',
- 'slug': 'mozillaball',
- 'status': 0,
- 'summary': 'Exciting Open Web development action!',
- 'support_email': 'john@doe.com',
- 'support_url': None}
-
- requests.get = Mock(return_value=Response(200, json.dumps(resp)))
- response = self.marketplace.status(1)
- content = json.loads(response.content)
- eq_(response.status_code, 200)
- eq_(content['manifest'], '0a650e5e4c434b5cb60c5495c0d88a89')
-
- def test_delete(self):
- raise SkipTest()
- requests.delete = Mock(return_value=Response(204, ''))
- response = self.marketplace.delete(1)
- eq_(response.status_code, 204)
- assert not response.content
-
- def test_delete_not_implemented(self):
- self.assertRaises(NotImplementedError, self.marketplace.delete, 1)
-
- def test_add_screenshot(self):
- path = lambda *a: os.path.join(
- os.path.dirname(os.path.abspath(__file__)), *a)
- resp = {'filetype': 'image/png',
- 'thumbnail_url': 'https://marketplace-dev-cdn.allizom.org/img/'
- 'uploads/previews/thumbs/71/71761.png?'
- 'modified=1340899716',
- 'image_url': 'https://marketplace-dev-cdn.allizom.org/img/'
- 'uploads/previews/full/71/71761.png?'
- 'modified=1340899716',
- 'position': 1,
- 'id': 71761,
- 'resource_uri': '/en-US/api/apps/preview/71761'}
- requests.post = Mock(return_value=Response(201, resp))
- response = self.marketplace.create_screenshot(123, path('mozilla.jpg'))
- eq_(response.status_code, 201)
- eq_(response.content['position'], 1)
- with open(path('mozilla.jpg')) as moz_file:
- b64_file = b64encode(moz_file.read())
- data = json.loads(requests.post.call_args[1]['data'])
- eq_(data['position'], 1)
- eq_(data['file']['data'], b64_file)
- eq_(data['file']['type'], 'image/jpeg')
- # create a screenshot with a jpeg image and not default position
- self.marketplace.create_screenshot(123,
- path('mozilla.jpg'), position=2)
- data = json.loads(requests.post.call_args[1]['data'])
- eq_(data['position'], 2)
- eq_(data['file']['type'], 'image/jpeg')
107 marketplace/tests/testConnection.py
View
@@ -1,107 +0,0 @@
-"""
-tests.testClient
-----------------
-"""
-import json
-import logging
-import unittest
-import urllib
-
-import requests
-import oauth2 as oauth
-
-from mock import Mock
-from nose import SkipTest
-from nose.tools import eq_
-
-from marketplace.connection import Connection
-
-log = logging.getLogger('test.%s' % __name__)
-
-# Preparing to mock the requests
-OLD_POST = requests.post
-OLD_PUT = requests.put
-OLD_GET = requests.get
-OLD_DELETE = requests.delete
-
-
-class Response(requests.Response):
- def __init__(self, status_code, content=None):
- super(Response, self).__init__()
- self.status_code = status_code
- self._content = content
-
-class TestClient(unittest.TestCase):
-
- def setUp(self):
- self.conn = Connection(consumer_key='key', consumer_secret='secret')
-
- def tearDown(self):
- requests.post = OLD_POST
- requests.put = OLD_PUT
- requests.get = OLD_GET
- requests.delete = OLD_DELETE
-
- def test_raising_on_httperror(self):
- resp = {"reason": "Error with OAuth headers"}
- requests.post = Mock(return_value=Response(401, json.dumps(resp)))
- self.assertRaises(requests.exceptions.HTTPError, self.conn.fetch,
- 'POST', 'http://example.com/', {})
-
- resp = "<html><title>404</title><body><p>Error 404</p></body></html>"
- requests.post = Mock(return_value=Response(404, resp))
- self.assertRaises(requests.exceptions.HTTPError, self.conn.fetch, 'POST',
- 'http://example.com/', {})
-
- def test_raising_on_unexpected(self):
- resp = {"reason": "Error with OAuth headers"}
- requests.post = Mock(return_value=Response(204, json.dumps(resp)))
- self.assertRaises(requests.exceptions.HTTPError, self.conn.fetch, 'POST',
- 'http://example.com/', {}, 201)
-
- def test_error_reason_json(self):
- resp = {"reason": "message"}
- eq_(Connection._get_error_reason(Response(204, json.dumps(resp))),
- resp['reason'])
-
- def test_error_reason_text(self):
- resp = "<html><title>404</title><body><p>Error 404</p></body></html>"
- eq_(Connection._get_error_reason(Response(204, resp)), resp)
-
- def test_set_consumer(self):
- assert isinstance(self.conn.consumer, oauth.Consumer)
-
- def test_prepare_request(self):
- prepared = self.conn.prepare_request('GET', 'http://example.com')
- assert 'headers' in prepared
- assert 'data' in prepared
- assert not prepared['data']
-
- data = {"some": "data"}
- prepared = self.conn.prepare_request('POST', 'http://ex.com', data)
- eq_(prepared['data'], json.dumps(data))
-
- prepared = self.conn.prepare_request('GET', 'http://ex.com', data)
- eq_(prepared['data'], urllib.urlencode(data))
-
- def test_get(self):
- requests.get = Mock(return_value=Response(200, '{}'))
- self.conn.fetch('GET', 'http://ex.com')
- assert requests.get.called
-
- def test_post(self):
- requests.post = Mock(return_value=Response(201, '{}'))
- self.conn.fetch('POST', 'http://ex.com')
- assert requests.post.called
-
- def test_put(self):
- requests.put = Mock(return_value=Response(202, '{}'))
- self.conn.fetch('PUT', 'http://ex.com')
- assert requests.put.called
-
- def test_delete(self):
- requests.delete = Mock(return_value=Response(204, '{}'))
- self.conn.fetch('DELETE', 'http://ex.com')
- assert requests.delete.called
-
-
Please sign in to comment.
Something went wrong with that request. Please try again.