91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
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, *, 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 _respond(self, *args, **kwargs):
|
|
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():
|
|
return FakeErrorLoader(ERROR)
|
|
|
|
def test_no_loaders(lchain, any_date):
|
|
try:
|
|
lchain.historical(any_date, 'USD')
|
|
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)
|
|
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__)
|
|
|
|
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"
|