Changeset - 4cbd89099230
[Not reviewed]
0 4 1
Brett Smith - 7 years ago 2017-05-12 12:12:21
brettcsmith@brettcsmith.org
errors: Start module.
5 files changed with 23 insertions and 33 deletions:
0 comments (0 inline, 0 general)
oxrlib/errors.py
Show inline comments
 
new file 100644
 
class LoaderError(Exception): pass
 
class LoaderNoDataError(LoaderError): pass
 
class LoaderBadRequestError(LoaderError): pass
 
class LoaderSourceError(LoaderError): pass
 
class NoLoadersError(Exception): pass
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
 

	
 
from . import errors
 

	
 
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
 
            raise errors.LoaderNoDataError(path) from error
 

	
 

	
 
class OXRAPIRequest:
 
    DEFAULT_API_ROOT = 'https://openexchangerates.org/api/'
 
    DEFAULT_RESPONSE_ENCODING = 'utf-8'
 

	
...
 
@@ -64,17 +46,17 @@ class OXRAPIRequest:
 
        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
 
            exc_class = errors.LoaderNoDataError
 
        elif status_code >= 500:
 
            exc_class = LoaderSourceError
 
            exc_class = errors.LoaderSourceError
 
        else:
 
            exc_class = LoaderBadRequestError
 
            exc_class = errors.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()),
...
 
@@ -94,18 +76,18 @@ class LoaderChain:
 
        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:
 
                except errors.LoaderError as this_error:
 
                    error = this_error
 
                else:
 
                    self.used_loader = loader
 
                    return response
 
            else:
 
                raise NoLoadersError() if error is None else error
 
                raise errors.NoLoadersError() if error is None else error
 
        return load_wrapper
 

	
 
    @_wrap_load_method
 
    def historical(self, date, base):
 
        pass
tests/test_FileCache.py
Show inline comments
 
import datetime
 
import pathlib
 

	
 
import pytest
 

	
 
from . import relpath
 
import oxrlib.errors
 
import oxrlib.loaders
 

	
 
CACHE_PATH = relpath('filecache')
 
CACHE_PATTERN = '{date}_{base}_cache.json'
 

	
 
@pytest.fixture
...
 
@@ -26,11 +27,11 @@ def test_cache_success(dummycache, date, base):
 
    (datetime.date(1999, 2, 1), 'EUR'),
 
    (datetime.date(1999, 3, 1), 'USD'),
 
])
 
def test_cache_not_found(dummycache, date, base):
 
    try:
 
        cache_file = dummycache.historical(date, base)
 
    except oxrlib.loaders.LoaderNoDataError:
 
    except oxrlib.errors.LoaderNoDataError:
 
        pass
 
    else:
 
        cache_file.close()
 
        assert False, "cache file found when unexpected"
tests/test_LoaderChain.py
Show inline comments
 
import io
 

	
 
import pytest
 

	
 
import oxrlib.errors
 
import oxrlib.loaders
 

	
 
from . import any_date
 

	
 
SUCCESS_S = '"success"\n'
 
ERROR = oxrlib.loaders.LoaderNoDataError("test")
 
ERROR = oxrlib.errors.LoaderNoDataError("test")
 

	
 
class FakeLoader:
 
    def __init__(self, result):
 
        self.result = result
 

	
 
    def _respond(self, *args, **kwargs):
...
 
@@ -40,13 +41,13 @@ def good_loader():
 
def error_loader():
 
    return FakeErrorLoader(ERROR)
 

	
 
def test_no_loaders(lchain, any_date):
 
    try:
 
        lchain.historical(any_date, 'USD')
 
    except oxrlib.loaders.NoLoadersError:
 
    except oxrlib.errors.NoLoadersError:
 
        pass
 
    else:
 
        assert False, "expected NoLoadersError not raised"
 

	
 
def test_one_with_success(lchain, any_date, good_loader):
 
    lchain.add_loader(good_loader)
tests/test_OXRAPIRequest.py
Show inline comments
...
 
@@ -4,12 +4,13 @@ import json
 
import os
 
import random
 
import string
 
import urllib.parse
 

	
 
import pytest
 
import oxrlib.errors
 
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))
...
 
@@ -79,17 +80,17 @@ def test_success(api_client, any_date, base):
 
    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),
 
    (400, oxrlib.errors.LoaderBadRequestError),
 
    (403, oxrlib.errors.LoaderBadRequestError),
 
    (404, oxrlib.errors.LoaderNoDataError),
 
    (410, oxrlib.errors.LoaderNoDataError),
 
    (500, oxrlib.errors.LoaderSourceError),
 
])
 
def test_failure(api_client, any_date, status_code, expect_exctype):
 
    opener = FakeOpener(FakeResponse(status_code))
 
    api_client.open_url = opener
 
    try:
 
        response = api_client.historical(any_date, 'USD')
0 comments (0 inline, 0 general)