Files @ 71893ace4dc0
Branch filter:

Location: NPO-Accounting/oxrlib/tests/test_LoaderChain.py

Brett Smith
tests: Historical tests use more flexible rate matching.

This lets the tests be more flexible about how much precision is used in
rates when appropriate, and makes them ready to parametrize for
Beancount.
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"