Files @ ae3e4617d31e
Branch filter:

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

Brett Smith
historical: Always format rates with the same precision.

When we format a rate as a price, we don't know how much precision
is "enough" to do the conversion, because we don't know what's
being converted to. As a result, we may (=will almost certainly)
end up formatting the rate with different precision on the cost
date vs. the price date, and that causes Beancount/Ledger to fail
to make the connection between them.

Using a constant of 6 is enough to make the current test for
"enough" precision pass, so just do that for now. This might need
further refinement in the future.
import http.client
import io
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))
API_ROOT = 'http://[100::]/oxrlibtest/'
API_ROOT_PATH = urllib.parse.urlsplit(API_ROOT).path

class FakeResponse:
    debuglevel = 0
    version = 11

    def __init__(self, status_code, reason=None, body=None, headers=None, encoding='utf-16'):
        if reason is None:
            reason = http.client.responses[status_code]
        if body is None:
            body = json.dumps(reason)
        if headers is None:
            headers = {
                'Content-Type': 'application/json; charset={}'.format(encoding),
                'Content-Length': str(len(body)),
            }
        self.status = status_code
        self.reason = reason
        read_fd, write_fd = os.pipe()
        with open(write_fd, 'w', encoding=encoding) as body_file:
            print('\ufeff', body, sep='', file=body_file)
        self.fp = open(read_fd, 'rb')
        self.headers = headers

    def __getattr__(self, name):
        return getattr(self.fp, name)

    def getheader(self, name, default=None):
        return self.headers.get(name, default)

    def getheaders(self):
        return list(self.headers.itervalues())


class FakeOpener:
    def __init__(self, response):
        self.response = response
        self.call_list = []

    def __call__(self, url, **kwargs):
        self.call_list.append((url, kwargs))
        return self.response

    def call_count(self):
        return len(self.call_list)

    def last_called_url(self):
        return self.call_list[-1][0]


@pytest.fixture
def api_client():
    return oxrlib.loaders.OXRAPIRequest(RANDOM_APPID, API_ROOT)

@pytest.mark.parametrize('base', ['USD', 'JPY'])
def test_success(api_client, any_date, base):
    body = "Good Test"
    opener = FakeOpener(FakeResponse(200, body))
    api_client.open_url = opener
    response = api_client.historical(any_date, base)
    assert opener.call_count() == 1
    urlparts = urllib.parse.urlsplit(opener.last_called_url())
    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.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')
    except expect_exctype:
        pass
    else:
        assert False, "got response: " + response.read()