NPO-Accounting/oxrlib Changeset - 351811bb30b1

Changeset - 351811bb30b1

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Brett Smith - 7 years ago 2017-05-17 16:27:20
brettcsmith@brettcsmith.org

LoaderChain: Add should_cache() method.

2 files changed with 35 insertions and 5 deletions:

oxrlib/loaders.py

tests/test_LoaderChain.py

0 comments (0 inline, 0 general)

oxrlib/loaders.py

➞

Show inline comments

@@ ... / @@ -11,34 +11,40 @@ class ReadCacheFile(cache.CacheFileBase): @@
         (FileNotFoundError, errors.LoaderNoDataError),
         (OSError, errors.LoaderSourceError),
+    ]
 class FileCache(cache.CacheBase):
     CacheFile = ReadCacheFile
     ConfigurationError = errors.CacheLoaderConfigurationError
     def open(self, path):
         return self.CacheFile(path)
     def is_cache(self):
         return True
 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 is_cache(self):
         return False
     def _get_response_encoding(self, response, default=None):
         try:
             content_type = response.getheader('Content-Type', 'application/json')
             _, ct_options = cgi.parse_header(content_type)
             encoding = ct_options['charset']
         except (KeyError, ValueError):
             encoding = self.DEFAULT_RESPONSE_ENCODING if default is None else default
         return encoding
     def _raw_query(self, url_tail, params):
         url = '{}?{}'.format(
             urllib.parse.urljoin(self.api_root, url_tail),
@@ ... / @@ -60,36 +66,41 @@ class OXRAPIRequest: @@
             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 = []
         self.can_cache = False
     def add_loader(self, loader):
         self.loaders.append(loader)
         self.can_cache = self.can_cache or loader.is_cache()
     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 errors.LoaderError as this_error:
                     error = this_error
                 else:
                     self.used_loader = loader
                     return response
             else:
                 raise errors.NoLoadersError() if error is None else error
         return load_wrapper
     @_wrap_load_method
     def historical(self, date, base):
         pass
     def should_cache(self):
         return self.can_cache and self.used_loader and not self.used_loader.is_cache()

tests/test_LoaderChain.py

➞

Show inline comments

@@ ... / @@ -2,40 +2,41 @@ import io @@
 import pytest
 import oxrlib.errors
 import oxrlib.loaders
 from . import any_date
 SUCCESS_S = '"success"\n'
 ERROR = oxrlib.errors.LoaderNoDataError("test")
 class FakeLoader:
     def __init__(self, result):
+    def __init__(self, result, *, is_cache=False):
         self.result = result
         self._is_cache = is_cache
     def _respond(self, *args, **kwargs):
         return io.StringIO(self.result)
     def __getattr__(self, name):
         return self._respond
     def is_cache(self):
         return self._is_cache
 class FakeErrorLoader(FakeLoader):
     def __init__(self, error):
         self.error = error
 class FakeErrorLoader(FakeLoader):
     def _respond(self, *args, **kwargs):
-        raise self.error
+        raise self.result
 @pytest.fixture
 def lchain():
     return oxrlib.loaders.LoaderChain()
 @pytest.fixture
 def good_loader():
     return FakeLoader(SUCCESS_S)
 @pytest.fixture
 def error_loader():
@@ ... @@
     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__)
 def test_should_cache(lchain, any_date, good_loader):
     cache_loader = FakeErrorLoader(ERROR, is_cache=True)
     lchain.add_loader(cache_loader)
     lchain.add_loader(good_loader)
     lchain.historical(any_date, 'USD')
     assert lchain.should_cache()
 def test_should_cache_unable(lchain, any_date, good_loader):
     lchain.add_loader(good_loader)
     lchain.historical(any_date, 'USD')
     assert not lchain.should_cache(), "suggested using unavailable cache"
 def test_should_cache_unneeded(lchain, any_date):
     loader = FakeLoader(SUCCESS_S, is_cache=True)
     lchain.add_loader(loader)
     lchain.historical(any_date, 'USD')
     assert not lchain.should_cache(), "suggested rewriting cache"

0 comments (0 inline, 0 general)