oxrlib/tests/test_OXRAPIRequest.py

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()