Skip to content
Browse files

Rebase on Tornado's IOLoop, remove type checking

Collapse _v2 classes into core Etsy API
rename files
fix this commit message
  • Loading branch information...
1 parent b371343 commit 26bcef1cabd31262ca67fc950ea78169cd56ffa8 @alekstorm committed Nov 7, 2011
Showing with 268 additions and 823 deletions.
  1. +186 −2 etsy/__init__.py
  2. +0 −348 etsy/_core.py
  3. +0 −54 etsy/_v2.py
  4. +2 −2 etsy/env.py
  5. +1 −1 setup.py
  6. +75 −0 test/core.py
  7. +0 −389 test/test_core.py
  8. +4 −27 test/util.py
View
188 etsy/__init__.py
@@ -1,9 +1,193 @@
-from _v2 import Association, EtsyV2 as Etsy
-from env import EtsyEnvSandbox, EtsyEnvProduction
+from __future__ import with_statement
+try:
+ import simplejson as json
+except ImportError:
+ import json
+import logging
+import re
+from tornado import gen
+from tornado.httpclient import AsyncHTTPClient
+from tornado.ioloop import IOLoop
+from tornado.template import Template
+from urllib import urlencode
+from _util import encode_multipart_formdata
+from env import SandboxEnv
__version__ = '0.3.1'
__author__ = 'Dan McKinley'
__copyright__ = 'Copyright 2010, Etsy Inc.'
__license__ = 'GPL v3'
__email__ = 'dan@etsy.com'
+
+logger = logging.getLogger('etsy')
+
+# TODO required non-url args
+# TODO declarative method table assertions
+
+class EtsyV2(object):
+ def __init__(self, api_key, callback, env=SandboxEnv, oauth_client=None, io_loop=None):
+ """
+ Parameters:
+ io_loop - An IO loop to use for method calls.
+
+ Example method specification:
+ {
+ 'name': 'createListing',
+ 'uri': '/listings',
+ 'visibility': 'private',
+ 'http_method': 'POST',
+ 'params': {
+ 'tags': 'array(string)',
+ 'price': 'float',
+ 'title': 'string',
+ 'quantity': 'int',
+ },
+ 'defaults': {
+ 'materials': None,
+ 'shop_section_id': None
+ },
+ 'type': 'Listing',
+ 'description': 'Creates a new Listing'
+ }
+ """
+
+ self.http_client = AsyncHTTPClient(io_loop=io_loop)
+ self.api_key = api_key
+ self.env = env
+ self.oauth_client = oauth_client
+
+ logger.info('Creating Etsy API, base url: %s', self.env.api_url)
+
+ self._compiled_methods = {}
+ self._methods = {}
+ def methods_callback(methods):
+ logger.info('Loaded method table: %r', methods)
+ self._methods = dict([(method['name'], method) for method in methods])
+ callback()
+ self._get_method_table(methods_callback)
+
+ def __getattr__(self, name):
+ method = self._compiled_methods.get(name, None)
+ if method:
+ return method
+
+ spec = self._methods[name]
+ uri = spec['uri']
+ positionals = set(re.findall(':([^/]+)', uri))
+
+ for p in positionals:
+ uri = re.sub(':%s(?=/|$)' % p, '%%(%s)s' % p, uri)
+ keywords = set(spec['params'].keys()) - positionals
+
+ code = Template("""def compiled({%for arg in positionals%}{{arg}},{%end%} callback, {%for arg in keywords%}{{arg}}=None,{%end%} fields=None, includes=None, api_key=None):
+ {% for arg in positionals %}
+ {% if spec['params'][arg].startswith('array') %}
+ if {{arg}} is not None and len({{arg}}) == 0:
+ raise ValueError("Positional argument '{{arg}}' must not be an empty array")
+ {% end %}
+ {% end %}
+ {% for arg in positionals|keywords %}
+ {% if spec['params'][arg].startswith('array') %}
+ if {{arg}} is not None:
+ {{arg}} = ','.join([str(arg) for arg in {{arg}}])
+ {% end %}
+ {% end %}
+
+ kwargs = { {{param_dict(keywords)}} }
+ kwargs['api_key'] = api_key or api.api_key
+ if fields:
+ kwargs['fields'] = ','.join(fields)
+ if includes:
+ kwargs['includes'] = ','.join([str(include) for include in includes])
+ return api._fetch_resource(
+ '{{uri}}' % { {{param_dict(positionals)}} },
+ '{{spec['http_method']}}',
+ dict([(name,value) for name,value in kwargs.iteritems() if value is not None]),
+ callback=callback)"""
+ ).generate(
+ positionals=positionals,
+ keywords=keywords,
+ spec=spec,
+ uri=uri,
+ param_dict=lambda params: ','.join(["'%s':%s" % (arg,arg) for arg in params])
+ )
+
+ namespace = {'api': self}
+ exec code in namespace
+ compiled = namespace['compiled']
+ compiled.__doc__ = spec['description']
+ self._compiled_methods[name] = compiled
+ return compiled
+
+
+ def _get_method_table(self, callback):
+ self._fetch_resource('/', 'GET', {'api_key': self.api_key}, callback)
+
+
+ def _fetch_url(self, url, http_method, content_type, body, callback):
+ logger.info("Fetching url: %r, method: %r, body: %r", url, http_method, body)
+ headers = {'Content-Type': content_type} if content_type else {}
+ if self.oauth_client is not None:
+ self.oauth_client.do_oauth_request(url, http_method, content_type, body, callback)
+ else:
+ self.http_client.fetch(self.env.api_url+url, callback,
+ method=http_method,
+ headers=headers,
+ body=body)
+
+
+ @gen.engine
+ def _fetch_resource(self, url, http_method, params, callback):
+ body = None
+ content_type = None
+ if http_method == 'GET':
+ url = '%s?%s' % (url, urlencode(params))
+ elif http_method == 'POST':
+ fields = []
+ files = []
+
+ for name, value in params.iteritems():
+ if hasattr(value, 'read'):
+ files.append((name, value.name, value.read()))
+ else:
+ fields.append((name, str(value)))
+
+ content_type, body = encode_multipart_formdata(fields, files)
+
+ data = (yield gen.Task(self._fetch_url, url, http_method, content_type, body)).body
+
+ logger.info('Data received: %r' % data)
+
+ try:
+ callback(json.loads(data)['results'])
+ except json.JSONDecodeError:
+ raise ValueError('Could not decode response from Etsy as JSON: %r' % data)
+
+
+class Association(object):
+ class Bounds(object):
+ def __init__(self, limit, offset=None):
+ self.limit = limit
+ self.offset = offset
+
+ def __init__(self, name, fields=None, scope=None, bounds=None, child=None):
+ self.name = name
+ self.fields = fields
+ self.scope = scope
+ self.bounds = bounds
+ self.child = child
+
+ def __str__(self):
+ elems = [self.name]
+ if self.fields is not None:
+ elems.extend(['(', ','.join(self.fields), ')'])
+ if self.scope is not None:
+ elems.extend([':', self.scope])
+ if self.bounds is not None:
+ elems.extend([':', str(self.bounds.limit)])
+ if self.bounds.offset is not None:
+ elems.extend([':', str(self.bounds.offset)])
+ if self.child is not None:
+ elems.extend(['/', str(self.child)])
+ return ''.join(elems)
View
348 etsy/_core.py
@@ -1,348 +0,0 @@
-from __future__ import with_statement
-from contextlib import closing
-import simplejson as json
-import urllib2
-from urllib import urlencode
-import os
-import re
-import tempfile
-import time
-from _util import encode_multipart_formdata
-
-
-missing = object()
-
-
-
-
-class TypeChecker(object):
- def __init__(self):
- self.checkers = {
- 'int': self.check_int,
- 'float': self.check_float,
- 'string': self.check_string,
- }
-
-
- def __call__(self, method, kwargs):
- params = method['params']
- for k, v in kwargs.items():
- if k == 'includes': continue
-
- if k not in params:
- raise ValueError('Unexpected argument: %s=%s' % (k, v))
-
- t = params[k]
- checker = self.compile(t)
- ok, converted = checker(v)
- if not ok:
- raise ValueError(
- "Bad value for parameter %s of type '%s' - %s" % (k, t, v))
- kwargs[k] = converted
-
-
- def compile(self, t):
- if t in self.checkers:
- return self.checkers[t]
- if t.startswith('enum'):
- return self.compile_enum(t)
- if t.startswith('array'):
- sub_t = self.compile(t[6:-1])
- def check_array(array):
- if not isinstance(array, list) or len(array) == 0:
- return False, array
- ok_elems, serialized_elems = zip(*map(sub_t, array)) if len(array) > 0 else ([], [])
- return reduce(lambda a,b: a and b, ok_elems, True), ','.join(serialized_elems)
- return check_array
- return self.always_ok
-
-
- def compile_enum(self, t):
- terms = [x.strip() for x in t[5:-1].split(',')]
- def check_enum(value):
- return (value in terms), value
- return check_enum
-
-
- def always_ok(self, value):
- return True, value
-
-
- def check_int(self, value):
- if isinstance(value, long):
- return True, value
- return isinstance(value, int), str(value)
-
-
- def check_float(self, value):
- if isinstance(value, int):
- return True, value
- return isinstance(value, float), str(value)
-
-
- def check_string(self, value):
- return isinstance(value, basestring), value
-
-
-
-
-class APIMethod(object):
- def __init__(self, api, spec):
- """
- Parameters:
- api - API object that this method is associated with.
- spec - dict with the method specification; e.g.:
-
- {'name': 'createListing', 'uri': '/listings', 'visibility':
- 'private', 'http_method': 'POST', 'params': {'description':
- 'text', 'tags': 'array(string)', 'price': 'float', 'title':
- 'string', 'materials': 'array(string)', 'shipping_template_id':
- 'int', 'quantity': 'int', 'shop_section_id': 'int'}, 'defaults':
- {'materials': None, 'shop_section_id': None}, 'type': 'Listing',
- 'description': 'Creates a new Listing'}
- """
-
- self.api = api
- self.spec = spec
- self.type_checker = self.api.type_checker
- self.__doc__ = self.spec['description']
- self.compiled = False
-
-
- def __call__(self, *args, **kwargs):
- if not self.compiled:
- self.compile()
- return self.invoke(*args, **kwargs)
-
-
- def compile(self):
- uri = self.spec['uri']
- self.positionals = re.findall(':([^/]+)', uri)
-
- for p in self.positionals:
- uri = re.sub(':%s(?=/|$)' % p, '%%(%s)s' % p, uri)
- self.uri_format = uri
-
- self.compiled = True
-
-
- def invoke(self, *args, **kwargs):
- if len(args) > len(self.positionals):
- raise ValueError('Too many positional arguments.')
-
- for k, v in zip(self.positionals, args):
- if k in kwargs:
- raise ValueError(
- 'Positional argument duplicated in kwargs: %s' % k)
- kwargs[k] = v
-
- ps = {}
- for p in self.positionals:
- if p not in kwargs:
- raise ValueError("Required argument '%s' not provided." % p)
- ps[p] = kwargs[p]
- del kwargs[p]
-
- fields = kwargs.pop('fields', None)
- includes = kwargs.pop('includes', None)
-
- self.type_checker(self.spec, ps)
- self.type_checker(self.spec, **kwargs)
-
- if fields:
- kwargs['fields'] = ','.join(fields)
- if includes:
- kwargs['includes'] = ','.join([str(include) for include in includes])
- return self.api._get(self.spec['http_method'], self.uri_format % ps, **kwargs)
-
-
-class MethodTableCache(object):
- max_age = 60*60*24
-
- def __init__(self, api, method_cache):
- self.api = api
- self.filename = self.resolve_file(method_cache)
- self.used_cache = False
- self.wrote_cache = False
-
-
- def resolve_file(self, method_cache):
- if method_cache is missing:
- return self.default_file()
- return method_cache
-
-
- def etsy_home(self):
- return self.api.etsy_home()
-
-
- def default_file(self):
- etsy_home = self.etsy_home()
- d = etsy_home if os.path.isdir(etsy_home) else tempfile.gettempdir()
- return os.path.join(d, 'methods.%s.json' % self.api.api_version)
-
-
- def get(self):
- ms = self.get_cached()
- if not ms:
- ms = self.api.get_method_table()
- self.cache(ms)
- return ms
-
-
- def get_cached(self):
- if self.filename is None or not os.path.isfile(self.filename):
- self.api.log('Not using cached method table.')
- return None
- if time.time() - os.stat(self.filename).st_mtime > self.max_age:
- self.api.log('Method table too old.')
- return None
- with open(self.filename, 'r') as f:
- self.used_cache = True
- self.api.log('Reading method table cache: %s' % self.filename)
- return json.loads(f.read())
-
-
- def cache(self, methods):
- if self.filename is None:
- self.api.log('Method table caching disabled, not writing new cache.')
- return
- with open(self.filename, 'w') as f:
- json.dump(methods, f)
- self.wrote_cache = True
- self.api.log('Wrote method table cache: %s' % self.filename)
-
-
-
-
-class API(object):
- def __init__(self, api_key='', key_file=None, method_cache=missing,
- log=None):
- """
- Creates a new API instance. When called with no arguments,
- reads the appropriate API key from the default ($HOME/.etsy/keys)
- file.
-
- Parameters:
- api_key - An explicit API key to use.
- key_file - A file to read the API keys from.
- method_cache - A file to save the API method table in for
- 24 hours. This speeds up the creation of API
- objects.
- log - An callable that accepts a string parameter.
- Receives log messages. No logging is done if
- this is None.
-
- Only one of api_key and key_file may be passed.
-
- If method_cache is explicitly set to None, no method table
- caching is performed. If the parameter is not passed, a file in
- $HOME/.etsy is used if that directory exists. Otherwise, a
- temp file is used.
- """
- if not getattr(self, 'api_url', None):
- raise AssertionError('No api_url configured.')
-
- if self.api_url.endswith('/'):
- raise AssertionError('api_url should not end with a slash.')
-
- if not getattr(self, 'api_version', None):
- raise AssertionError('API object should define api_version')
-
- if api_key and key_file:
- raise AssertionError('Keys can be read from a file or passed, '
- 'but not both.')
-
- if api_key:
- self.api_key = api_key
- else:
- self.api_key = self._read_key(key_file)
-
- self.log = log or self._ignore
- if not callable(self.log):
- raise ValueError('log must be a callable.')
-
- self.type_checker = TypeChecker()
-
- self.decode = json.loads
-
- self.log('Creating %s Etsy API, base url=%s.' % (
- self.api_version, self.api_url))
- self._get_methods(method_cache)
-
-
-
- def _ignore(self, _):
- pass
-
-
- def _get_methods(self, method_cache):
- self.method_cache = MethodTableCache(self, method_cache)
- ms = self.method_cache.get()
- self._methods = dict([(m['name'], m) for m in ms])
-
- for method in ms:
- setattr(self, method['name'], APIMethod(self, method))
-
- # self.log('API._get_methods: self._methods = %r' % self._methods)
-
-
- def etsy_home(self):
- return os.path.expanduser('~/.etsy')
-
-
- def get_method_table(self):
- return self._get('GET', '/')
-
-
- def _read_key(self, key_file):
- key_file = key_file or os.path.join(self.etsy_home(), 'keys')
- if not os.path.isfile(key_file):
- raise AssertionError(
- "The key file '%s' does not exist. Create a key file or "
- 'pass an API key explicitly.' % key_file)
-
- gs = {}
- execfile(key_file, gs)
- return gs[self.api_version]
-
-
- def _get_url(self, url, http_method, content_type, body):
- self.log("API._get_url: url = %r" % url)
- with closing(urllib2.urlopen(url)) as f:
- return f.read()
-
-
- def _get(self, http_method, url, **kwargs):
- kwargs.update(dict(api_key=self.api_key))
-
- if http_method == 'GET':
- url = '%s%s?%s' % (self.api_url, url, urlencode(kwargs))
- body = None
- content_type = None
- elif http_method == 'POST':
- url = '%s%s' % (self.api_url, url)
- fields = []
- files = []
-
- for name, value in kwargs.items():
- if hasattr(value, 'read'):
- files.append((name, value.name, value.read()))
- else:
- fields.append((name, str(value)))
-
- content_type, body = encode_multipart_formdata(fields, files)
-
- self.last_url = url
- data = self._get_url(url, http_method, content_type, body)
-
- self.log('API._get: http_method = %r, url = %r, data = %r' % (http_method, url, data))
-
- try:
- self.data = self.decode(data)
- except json.JSONDecodeError:
- raise ValueError('Could not decode response from Etsy as JSON: %r' % data)
-
- self.count = self.data['count']
- return self.data['results']
-
View
54 etsy/_v2.py
@@ -1,54 +0,0 @@
-import urllib
-from _core import API, missing
-from env import EtsyEnvSandbox, EtsyEnvProduction
-
-try:
- from urlparse import parse_qsl
-except ImportError:
- from cgi import parse_qsl
-
-class EtsyV2(API):
- api_version = 'v2'
-
- def __init__(self, api_key='', key_file=None, method_cache=missing,
- etsy_env=EtsyEnvSandbox, log=None, etsy_oauth_client=None):
- self.api_url = etsy_env.api_url
- self.etsy_oauth_client = None
-
- if etsy_oauth_client:
- self.etsy_oauth_client = etsy_oauth_client
-
- super(EtsyV2, self).__init__(api_key, key_file, method_cache, log)
-
- def _get_url(self, url, http_method, content_type, body):
- if self.etsy_oauth_client is not None:
- return self.etsy_oauth_client.do_oauth_request(url, http_method, content_type, body)
- return API._get_url(self, url, http_method, content_type, body)
-
-
-class Association(object):
- class Bounds(object):
- def __init__(self, limit, offset=None):
- self.limit = limit
- self.offset = offset
-
- def __init__(self, name, fields=None, scope=None, bounds=None, child=None):
- self.name = name
- self.fields = fields
- self.scope = scope
- self.bounds = bounds
- self.child = child
-
- def __str__(self):
- elems = [self.name]
- if self.fields is not None:
- elems.extend(['(', ','.join(self.fields), ')'])
- if self.scope is not None:
- elems.extend([':', self.scope])
- if self.bounds is not None:
- elems.extend([':', str(self.bounds.limit)])
- if self.bounds.offset is not None:
- elems.extend([':', str(self.bounds.offset)])
- if self.child is not None:
- elems.extend(['/', str(self.child)])
- return ''.join(elems)
View
4 etsy/env.py
@@ -1,10 +1,10 @@
-class EtsyEnvSandbox(object):
+class SandboxEnv(object):
request_token_url = 'http://sandbox.openapi.etsy.com/v2/oauth/request_token'
access_token_url = 'http://sandbox.openapi.etsy.com/v2/oauth/access_token'
signin_url = 'https://www.etsy.com/oauth/signin'
api_url = 'http://sandbox.openapi.etsy.com/v2'
-class EtsyEnvProduction(object):
+class ProductionEnv(object):
request_token_url = 'http://openapi.etsy.com/v2/oauth/request_token'
access_token_url = 'http://openapi.etsy.com/v2/oauth/access_token'
signin_url = 'https://www.etsy.com/oauth/signin'
View
2 setup.py
@@ -15,7 +15,7 @@
keywords = 'etsy api handmade',
packages = ['etsy'],
long_description = long_description,
- test_suite = 'test.test_core',
+ test_suite = 'test.core',
install_requires=['simplejson >= 2.0'],
extras_require = {
'OAuth': ["oauth2>=1.2.0"],
View
75 test/core.py
@@ -0,0 +1,75 @@
+from __future__ import with_statement
+from StringIO import StringIO
+from tornado import gen
+from tornado.httpclient import HTTPRequest, HTTPResponse
+
+from etsy import EtsyV2
+from util import EtsyTestCase, ignore
+
+# TODO associations tests
+
+class MockAPI(EtsyV2):
+ def __init__(self, *args, **kwargs):
+ EtsyV2.__init__(self, callback=ignore, *args, **kwargs)
+
+ def _get_method_table(self, callback):
+ callback([{
+ 'name': 'testMethod',
+ 'uri': '/test/:ps_arr_str',
+ 'http_method': 'GET',
+ 'params': {
+ 'ps_arr_str': 'array(string)',
+ 'kw_string': 'string',
+ 'kw_int': 'int',
+ 'kw_float': 'float',
+ 'kw_array_int': 'array(int)',
+ 'kw_enum': 'enum(foo, bar, baz)',
+ 'kw_unknown': 'unknown type',
+ },
+ 'type': 'echo',
+ 'description': 'test method'
+ }])
+
+ def _fetch_resource(self, url, http_method, params, callback):
+ callback([{
+ 'url': url,
+ 'http_method': http_method,
+ 'params': params,
+ }])
+
+
+class CoreTests(EtsyTestCase):
+ def setUp(self):
+ super(CoreTests, self).setUp()
+ self.app = MockAPI(api_key='apikey')
+
+ def test_method_created(self):
+ self.assertTrue(getattr(self.app, 'testMethod', None) is not None)
+
+ @gen.engine
+ def test_positional_argument_in_url(self):
+ result = (yield gen.Task(self.app.testMethod, ps_arr_str=['a']))[0]
+ self.assertEquals(result['url'], '/test/a')
+
+ @gen.engine
+ def test_positional_argument_array_commas_in_url(self):
+ result = (yield gen.Task(self.app.testMethod, ps_arr_str=['a','b']))[0]
+ self.assertEquals(result['url'], '/test/a,b')
+
+ def test_invalid_empty_positional_argument_array(self):
+ msg = self.assertRaises(ValueError, self.app.testMethod, ps_arr_str=[], callback=ignore)
+ self.assertEqual(msg, "Positional argument 'ps_arr_str' must not be an empty array")
+
+ @gen.engine
+ def test_http_method_match(self):
+ result = (yield gen.Task(self.app.testMethod, ps_arr_str=['a']))[0]
+ self.assertEquals(result['http_method'], 'GET')
+
+ @gen.engine
+ def test_keyword_argument_in_params(self):
+ result = (yield gen.Task(self.app.testMethod, ps_arr_str=['a'], kw_int=5))[0]
+ self.assertTrue('kw_int' in result['params'])
+ self.assertEquals(result['params']['kw_int'], 5)
+
+ def test_docstring_set(self):
+ self.assertEquals(self.app.testMethod.__doc__, 'test method')
View
389 test/test_core.py
@@ -1,389 +0,0 @@
-from __future__ import with_statement
-from etsy._core import API, MethodTableCache, missing
-from cgi import parse_qs
-from urlparse import urlparse
-import os
-from util import Test
-import tempfile
-
-
-
-class MockAPI(API):
- api_url = 'http://host'
- api_version = 'v1'
-
-
- def etsy_home(self):
- return Test.scratch_dir
-
-
- def get_method_table(self, *args):
- return [{'name': 'testMethod',
- 'uri': '/test/:ps_arr_str',
- 'http_method': 'GET',
- 'params': {
- 'ps_arr_str': 'array(string)',
- 'kw_string': 'string',
- 'kw_int': 'int',
- 'kw_float': 'float',
- 'kw_array_int': 'array(int)',
- 'kw_enum': 'enum(foo, bar, baz)',
- 'kw_unknown': 'unknown type',
- },
- 'type': 'int',
- 'description': 'test method.'}]
-
-
- def _get_url(self, url, http_method, content_type, body):
- return '{ "count": 1, "results": [3] }'
-
-
-
-class MockLog(object):
- def __init__(self, test):
- self.lines = []
- self.test = test
-
- def __call__(self, msg):
- self.lines.append(msg)
-
-
- def assertLine(self, msg):
- failmsg = 'Could not find "%s" in the log. The log was:\n\n%s' % (
- msg, '\n'.join([' %s' % x for x in self.lines]))
- self.test.assertTrue(msg in self.lines, failmsg)
-
-
-
-class CoreTests(Test):
- def setUp(self):
- self.api = MockAPI('apikey', log=MockLog(self))
-
-
- def last_query(self):
- qs = urlparse(self.api.last_url).query
- return parse_qs(qs)
-
-
- def test_method_created(self):
- self.assertTrue('testMethod' in dir(self.api))
-
-
- def test_url_params(self):
- self.api.testMethod(ps_arr_str=['a'])
- self.assertEqual(self.api.last_url,
- 'http://host/test/a?api_key=apikey')
-
-
- def test_count_saved(self):
- self.api.testMethod(ps_arr_str=['a'])
- self.assertTrue(self.api.count)
-
-
- def test_results_returned(self):
- x = self.api.testMethod(ps_arr_str=['a'])
- self.assertEquals(x, [3])
-
-
- def test_query_params(self):
- self.api.testMethod(ps_arr_str=['a'], kw_int=1)
- self.assertEqual(self.last_query(), {
- 'api_key': ['apikey'],
- 'kw_int': ['1'],
- })
-
-
- def test_docstring_set(self):
- self.assertEquals(self.api.testMethod.__doc__,
- 'test method.')
-
-
-
- def test_api_url_required(self):
- msg = self.assertRaises(AssertionError, API, '')
- self.assertEqual('No api_url configured.', msg)
-
-
- def test_api_url_cannot_end_with_slash(self):
- class Foo(API):
- api_url = 'http://host/'
-
- msg = self.assertRaises(AssertionError, Foo, '')
- self.assertEqual('api_url should not end with a slash.', msg)
-
-
- def test_api_should_define_version(self):
- class Foo(API):
- api_url = 'http://host'
-
- msg = self.assertRaises(AssertionError, Foo)
- self.assertEqual(msg, 'API object should define api_version')
-
-
- def test_key_file_does_not_exist(self):
- msg = self.assertRaises(AssertionError, MockAPI,
- key_file='this does not exist')
- self.assertTrue("'this does not exist' does not exist" in msg)
-
-
- def test_reading_api_key(self):
- with open('testkeys', 'w') as f:
- f.write("v1 = 'abcdef'")
- try:
- self.assertEqual(MockAPI(key_file='testkeys').api_key, 'abcdef')
- finally:
- os.unlink('testkeys')
-
-
- def test_unrecognized_kwarg(self):
- msg = self.assertRaises(ValueError, self.api.testMethod,
- ps_arr_str=['a'], not_an_arg=1)
- self.assertEqual(msg, 'Unexpected argument: not_an_arg=1')
-
-
- def test_unknown_parameter_type_is_passed(self):
- self.api.testMethod(ps_arr_str=['a'], kw_unknown=1)
- self.assertEqual(self.last_query()['kw_unknown'], ['1'])
-
-
- def test_parameter_type_int(self):
- self.api.testMethod(ps_arr_str=['a'], kw_int=5)
- self.assertEqual(self.last_query()['kw_int'], ['5'])
-
-
- def test_parameter_type_long(self):
- self.api.testMethod(ps_arr_str=['a'], kw_int=5L)
- self.assertEqual(self.last_query()['kw_int'], ['5'])
-
-
- def bad_value_msg(self, name, t, v):
- return "Bad value for parameter %s of type '%s' - %s" % (name, t, v)
-
- def test_invalid_parameter_type_int(self):
- msg = self.assertRaises(ValueError, self.api.testMethod,
- ps_arr_str=['a'], kw_int=5.6)
- self.assertEqual(msg, self.bad_value_msg('kw_int', 'int', 5.6))
-
-
- def test_parameter_type_float(self):
- self.api.testMethod(ps_arr_str=['a'], kw_float=42.1)
- self.assertEqual(self.last_query()['kw_float'], ['42.1'])
-
-
- def test_invalid_parameter_type_float(self):
- msg = self.assertRaises(ValueError, self.api.testMethod,
- ps_arr_str=['a'], kw_float='x')
- self.assertEqual(msg, self.bad_value_msg('kw_float', 'float', 'x'))
-
-
- def test_int_accepted_as_float(self):
- self.api.testMethod(ps_arr_str=['a'], kw_int=3)
- self.assertEqual(self.last_query()['kw_int'], ['3'])
-
-
- def test_parameter_type_enum(self):
- self.api.testMethod(ps_arr_str=['a'], kw_enum='bar')
- self.assertEqual(self.last_query()['kw_enum'], ['bar'])
-
-
- def test_invalid_parameter_type_enum(self):
- msg = self.assertRaises(ValueError, self.api.testMethod,
- ps_arr_str=['a'], kw_enum='goo')
- self.assertEqual(msg, self.bad_value_msg(
- 'kw_enum', 'enum(foo, bar, baz)', 'goo'))
-
-
- def test_parameter_type_string(self):
- self.api.testMethod(ps_arr_str=['a'], kw_string='blah')
- self.assertEqual(self.last_query()['kw_string'], ['blah'])
-
-
- def test_invalid_parameter_type_string(self):
- msg = self.assertRaises(ValueError, self.api.testMethod,
- ps_arr_str=['a'], kw_string=5)
- self.assertEqual(msg, self.bad_value_msg('kw_string', 'string', 5))
-
-
- def test_url_arguments_work_positionally(self):
- self.api.testMethod(['foo'])
- self.assertEqual(self.api.last_url,
- 'http://host/test/foo?api_key=apikey')
-
- def test_invalid_positional_argument_type_arr_string(self):
- msg = self.assertRaises(ValueError, self.api.testMethod,
- ps_arr_str=5)
- self.assertEqual(msg, self.bad_value_msg('ps_arr_str', 'array(string)', 5))
-
- def test_invalid_empty_positional_argument_array(self):
- msg = self.assertRaises(ValueError, self.api.testMethod,
- ps_arr_str=[])
- self.assertEqual(msg, self.bad_value_msg('ps_arr_str', 'array(string)', []))
-
- def test_too_many_positionals(self):
- msg = self.assertRaises(ValueError, self.api.testMethod, ['a'], ['a'])
- self.assertEqual('Too many positional arguments.', msg)
-
-
- def test_positional_argument_not_provided(self):
- msg = self.assertRaises(ValueError, self.api.testMethod)
- self.assertEqual("Required argument 'ps_arr_str' not provided.", msg)
-
-
- def test_positional_argument_duplicated_in_kwargs(self):
- msg = self.assertRaises(ValueError, self.api.testMethod, ['a'], ps_arr_str=['a'])
- self.assertEqual('Positional argument duplicated in kwargs: ps_arr_str',
- msg)
-
-
- def test_api_key_and_key_file_both_passed(self):
- msg = self.assertRaises(AssertionError, MockAPI,
- api_key='x', key_file='y')
- self.assertEqual('Keys can be read from a file or passed, but not both.',
- msg)
-
-
- def test_logging_works(self):
- self.api.log('foo')
- self.api.log.assertLine('foo')
-
-
- def test_log_at_startup(self):
- self.api.log.assertLine('Creating v1 Etsy API, base url=http://host.')
-
-
-
-
-
-
-class MockAPI_NoMethods(MockAPI):
- def _get_methods(self, method_cache):
- pass
-
-
-
-class MethodTableCacheTests(Test):
-
-
- def cache(self, method_cache=missing):
- self.api = MockAPI_NoMethods('apikey')
- self._cache = MethodTableCache(self.api, method_cache)
- return self._cache
-
-
- def test_uses_etsy_home_if_exists(self):
- c = self.cache()
- self.assertEqual(os.path.dirname(c.filename), self.scratch_dir)
-
-
- def test_uses_temp_dir_if_no_etsy_home(self):
- self.delete_scratch()
- c = self.cache()
- self.assertEqual(os.path.dirname(c.filename), tempfile.gettempdir())
-
-
- def test_uses_provided_file(self):
- fn = os.path.join(self.scratch_dir, 'foo.json')
- self.assertEqual(self.cache(method_cache=fn).filename, fn)
-
-
- def test_multiple_versions(self):
- c = self.cache()
-
- class MockAPI2(MockAPI):
- api_version = 'v3'
-
- self.assertNotEqual(MockAPI2('key').method_cache.filename, c.filename)
-
-
- def get_uncached(self):
- c = self.cache()
- return c.get()
-
-
- def test_no_cache_file_returns_results(self):
- self.assertEqual(1, len(self.get_uncached()))
-
-
- def test_no_cache_file_writes_cache(self):
- self.get_uncached()
- self.assertTrue(self._cache.wrote_cache)
-
-
- def test_no_cache_file(self):
- self.get_uncached()
- self.assertFalse(self._cache.used_cache)
-
-
- def get_cached(self):
- c = self.cache()
- c.get()
- c = self.cache()
- return c.get()
-
-
- def test_caching(self):
- self.get_cached()
- self.assertTrue(self._cache.used_cache)
-
-
- def test_caching_returns_results(self):
- self.assertEqual(1, len(self.get_cached()))
-
-
- def test_caching_doesnt_overwrite_cache(self):
- self.get_cached()
- self.assertFalse(self._cache.wrote_cache)
-
-
- def make_old_cache(self):
- self.get_cached()
- fn = self._cache.filename
- s = os.stat(fn)
- os.utime(fn, (s.st_atime, s.st_mtime - 48*60*60))
-
-
- def test_expired(self):
- self.make_old_cache()
- c = self.cache()
- c.get()
- self.assertFalse(c.used_cache)
-
-
- def test_none_passed_does_not_cache(self):
- self.get_cached()
- c = self.cache(method_cache=None)
- c.get()
- self.assertFalse(c.used_cache)
-
-
- def log_tester(self, method_cache=missing):
- return MockAPI('key', method_cache=method_cache, log=MockLog(self))
-
-
- def test_logs_when_not_using_cache(self):
- api = self.log_tester(None)
- api.log.assertLine('Not using cached method table.')
-
-
- def test_logs_when_method_table_too_old(self):
- self.make_old_cache()
- self.log_tester().log.assertLine('Method table too old.')
-
-
- def test_logs_when_reading_cache(self):
- api = MockAPI('key')
- self.log_tester().log.assertLine('Reading method table cache: %s' %
- api.method_cache.filename)
-
-
- def test_logs_when_not_writing_new_cache(self):
- api = self.log_tester(None)
- api.log.assertLine(
- 'Method table caching disabled, not writing new cache.')
-
-
- def test_logs_when_writing_new_cache(self):
- t = self.log_tester()
- t.log.assertLine('Wrote method table cache: %s' %
- t.method_cache.filename)
-
View
31 test/util.py
@@ -1,30 +1,9 @@
-from unittest import TestCase
-import os
-import shutil
-
-
-this_dir = os.path.realpath(os.path.dirname(__file__))
-
-
-class Test(TestCase):
-
- scratch_dir = os.path.join(this_dir, 'scratch')
-
-
- def setUp(self):
- if not os.path.isdir(self.scratch_dir):
- os.mkdir(self.scratch_dir)
-
-
- def tearDown(self):
- self.delete_scratch()
-
-
- def delete_scratch(self):
- if os.path.isdir(self.scratch_dir):
- shutil.rmtree(self.scratch_dir)
+from tornado.testing import AsyncTestCase
+def ignore(*args, **kwargs):
+ pass
+class EtsyTestCase(AsyncTestCase):
def assertRaises(self, cls, f, *args, **kwargs):
try:
f(*args, **kwargs)
@@ -33,5 +12,3 @@ def assertRaises(self, cls, f, *args, **kwargs):
else:
name = cls.__name__ if hasattr(cls, '__name__') else str(cls)
raise self.failureException, "%s not raised" % name
-
-

0 comments on commit 26bcef1

Please sign in to comment.
Something went wrong with that request. Please try again.