100 lines
3 KiB
Python
100 lines
3 KiB
Python
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()
|