NPO-Accounting/oxrlib Changeset - 1927a1812033

Changeset - 1927a1812033

Parent rev.

Child rev.

[Not reviewed]

0 3 1

Brett Smith - 7 years ago 2017-05-12 11:40:27
brettcsmith@brettcsmith.org

loaders: Add LoaderChain.

4 files changed with 115 insertions and 5 deletions:

oxrlib/loaders.py

tests/__init__.py

tests/test_LoaderChain.py

tests/test_OXRAPIRequest.py

0 comments (0 inline, 0 general)

oxrlib/loaders.py

➞

Show inline comments

 import cgi
 import functools
 import io
 import urllib.request
 import urllib.parse
 class LoaderError(Exception):
     pass
 class LoaderNoDataError(LoaderError):
     pass
 class LoaderBadRequestError(LoaderError):
     pass
 class LoaderSourceError(LoaderError):
     pass
 class NoLoadersError(Exception):
     pass
 class FileCache:
     def __init__(self, dir_path, filename_pattern):
         self.dir_path = dir_path
         self.pattern = filename_pattern
     def historical(self, date, base):
         path = self.dir_path / self.pattern.format(date=date.isoformat(), base=base)
         try:
             return path.open()
         except FileNotFoundError as error:
             raise LoaderNoDataError(path) from error
 class OXRAPIRequest:
     DEFAULT_API_ROOT = 'https://openexchangerates.org/api/'
     DEFAULT_RESPONSE_ENCODING = 'utf-8'
     def __init__(self, app_id, api_root=None, *, open_func=urllib.request.urlopen):
         self.api_root = self.DEFAULT_API_ROOT if api_root is None else api_root
         self.app_id = app_id
         self.open_url = open_func
     def _get_response_encoding(self, response, default=None):
         try:
@@ ... / @@ -54,24 +59,53 @@ class OXRAPIRequest: @@
         url = '{}?{}'.format(
             urllib.parse.urljoin(self.api_root, url_tail),
             urllib.parse.urlencode(params),
+        )
         response = self.open_url(url)
         status_code = response.status
         encoding = self._get_response_encoding(response)
         response_body = io.TextIOWrapper(response, encoding=encoding)
         if 200 <= status_code < 203:
             return response_body
         elif status_code == 404 or status_code == 410:
             exc_class = LoaderNoDataError
         elif status_code >= 500:
             exc_class = LoaderSourceError
         else:
             exc_class = LoaderBadRequestError
         with response_body:
             raise exc_class(url, response_body.read(64 * 1024))
     def historical(self, date, base):
         return self._raw_query(
             'historical/{}.json'.format(date.isoformat()),
             {'app_id': self.app_id, 'base': base},
+        )
 class LoaderChain:
     def __init__(self):
         self.loaders = []
     def add_loader(self, loader):
         self.loaders.append(loader)
     def _wrap_load_method(orig_func):
         @functools.wraps(orig_func)
         def load_wrapper(self, *args, **kwargs):
             self.used_loader = None
             error = None
             for loader in self.loaders:
                 try:
                     response = getattr(loader, orig_func.__name__)(*args, **kwargs)
                 except LoaderError as this_error:
                     error = this_error
                 else:
                     self.used_loader = loader
                     return response
             else:
                 raise NoLoadersError() if error is None else error
         return load_wrapper
     @_wrap_load_method
     def historical(self, date, base):
         pass

tests/__init__.py

➞

Show inline comments

 import datetime
 import pathlib
 import random
 import pytest
 TEST_DIR = pathlib.Path(__file__).parent
 def relpath(*parts):
     return TEST_DIR / pathlib.Path(*parts)
 @pytest.fixture
 def any_date():
     return datetime.date.today() - datetime.timedelta(days=730 - random.randint(0, 365))

tests/test_LoaderChain.py

➞

Show inline comments

 new file 100644
 import io
 import pytest
 import oxrlib.loaders
 from . import any_date
 SUCCESS_S = '"success"\n'
 ERROR = oxrlib.loaders.LoaderNoDataError("test")
 class FakeLoader:
     def __init__(self, result):
         self.result = result
     def _respond(self, *args, **kwargs):
         return io.StringIO(self.result)
     def __getattr__(self, name):
         return self._respond
 class FakeErrorLoader(FakeLoader):
     def __init__(self, error):
         self.error = error
     def _respond(self, *args, **kwargs):
         raise self.error
 @pytest.fixture
 def lchain():
     return oxrlib.loaders.LoaderChain()
 @pytest.fixture
 def good_loader():
     return FakeLoader(SUCCESS_S)
 @pytest.fixture
 def error_loader():
     return FakeErrorLoader(ERROR)
 def test_no_loaders(lchain, any_date):
     try:
         lchain.historical(any_date, 'USD')
     except oxrlib.loaders.NoLoadersError:
         pass
     else:
         assert False, "expected NoLoadersError not raised"
 def test_one_with_success(lchain, any_date, good_loader):
     lchain.add_loader(good_loader)
     response = lchain.historical(any_date, 'USD')
     assert response.read(32) == SUCCESS_S
 def test_two_with_success(lchain, any_date, good_loader, error_loader):
     lchain.add_loader(error_loader)
     lchain.add_loader(good_loader)
     response = lchain.historical(any_date, 'USD')
     assert response.read(32) == SUCCESS_S
 @pytest.mark.parametrize('count', [1, 2])
 def test_no_success(lchain, any_date, error_loader, count):
     for _ in range(count):
         lchain.add_loader(error_loader)
     try:
         lchain.historical(any_date, 'USD')
     except type(ERROR) as error:
         assert error is ERROR
     else:
         assert False, "{} not raised".format(type(ERROR).__name__)

tests/test_OXRAPIRequest.py

➞

Show inline comments

 import datetime
 import http.client
 import io
 import json
 import os
 import random
 import string
 import urllib.parse
 import pytest
 import oxrlib.loaders
 from . import any_date
 APPID_CHARS = string.ascii_letters + string.digits
 RANDOM_APPID = ''.join(random.choice(APPID_CHARS) for _ in range(32))
 API_ROOT = 'http://[100::]/oxrlibtest/'
 API_ROOT_PATH = urllib.parse.urlsplit(API_ROOT).path
 class FakeResponse:
     debuglevel = 0
     version = 11
     def __init__(self, status_code, reason=None, body=None, headers=None, encoding='utf-16'):
         if reason is None:
             reason = http.client.responses[status_code]
         if body is None:
             body = json.dumps(reason)
         if headers is None:
             headers = {
                 'Content-Type': 'application/json; charset={}'.format(encoding),
                 'Content-Length': str(len(body)),
+            }
         self.status = status_code
         self.reason = reason
         read_fd, write_fd = os.pipe()
         with open(write_fd, 'w', encoding=encoding) as body_file:
             print('\ufeff', body, sep='', file=body_file)
@@ ... / @@ -46,52 +47,48 @@ class FakeResponse: @@
     def getheaders(self):
         return list(self.headers.itervalues())
 class FakeOpener:
     def __init__(self, response):
         self.response = response
         self.call_list = []
     def __call__(self, url, **kwargs):
         self.call_list.append((url, kwargs))
         return self.response
     def call_count(self):
         return len(self.call_list)
     def last_called_url(self):
         return self.call_list[-1][0]
 @pytest.fixture
 def api_client():
     return oxrlib.loaders.OXRAPIRequest(RANDOM_APPID, API_ROOT)
 @pytest.fixture
 def any_date():
     return datetime.date.today() - datetime.timedelta(days=730 - random.randint(0, 365))
 @pytest.mark.parametrize('base', ['USD', 'JPY'])
 def test_success(api_client, any_date, base):
     body = "Good Test"
     opener = FakeOpener(FakeResponse(200, body))
     api_client.open_url = opener
     response = api_client.historical(any_date, base)
     assert opener.call_count() == 1
     urlparts = urllib.parse.urlsplit(opener.last_called_url())
     assert urlparts.path == '{}historical/{}.json'.format(API_ROOT_PATH, any_date.isoformat())
     params = urllib.parse.parse_qs(urlparts.query)
     assert params['base'] == [base]
     assert response.read() == (json.dumps(body) + "\n")
 @pytest.mark.parametrize('status_code,expect_exctype', [
     (400, oxrlib.loaders.LoaderBadRequestError),
     (403, oxrlib.loaders.LoaderBadRequestError),
     (404, oxrlib.loaders.LoaderNoDataError),
     (410, oxrlib.loaders.LoaderNoDataError),
     (500, oxrlib.loaders.LoaderSourceError),
 ])
 def test_failure(api_client, any_date, status_code, expect_exctype):
     opener = FakeOpener(FakeResponse(status_code))
     api_client.open_url = opener
     try:

0 comments (0 inline, 0 general)