Changeset - 351811bb30b1
[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:
0 comments (0 inline, 0 general)
oxrlib/loaders.py
Show inline comments
...
 
@@ -17,22 +17,28 @@ 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):
...
 
@@ -66,15 +72,17 @@ class OXRAPIRequest:
 
        )
 

	
 

	
 
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
...
 
@@ -90,6 +98,9 @@ class LoaderChain:
 
                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
...
 
@@ -8,28 +8,29 @@ 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()
 

	
...
 
@@ -67,6 +68,24 @@ def test_no_success(lchain, any_date, error_loader, count):
 
    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)