diff --git a/tests/test_OXRAPIRequest.py b/tests/test_OXRAPIRequest.py new file mode 100644 index 0000000000000000000000000000000000000000..147af8a11f327b3e9e038117397ee30feb0bea48 --- /dev/null +++ b/tests/test_OXRAPIRequest.py @@ -0,0 +1,102 @@ +import datetime +import http.client +import io +import json +import os +import random +import string +import urllib.parse + +import pytest +import oxrlib.loaders + +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.fixture +def any_date(): + return datetime.date.today() - datetime.timedelta(days=730 - random.randint(0, 365)) + +@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.loaders.LoaderBadRequestError), + (403, oxrlib.loaders.LoaderBadRequestError), + (404, oxrlib.loaders.LoaderNoDataError), + (410, oxrlib.loaders.LoaderNoDataError), + (500, oxrlib.loaders.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()