oxrlib/tests/test_LoaderChain.py

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"