261 lines
10 KiB
Python
261 lines
10 KiB
Python
import argparse
|
|
import decimal
|
|
import io
|
|
import itertools
|
|
import json
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from . import any_date, relpath
|
|
|
|
import oxrlib.commands.historical as oxrhist
|
|
|
|
class FakeResponder:
|
|
def __init__(self, *response_paths):
|
|
self.paths = itertools.cycle(response_paths)
|
|
|
|
def _respond(self, *args, **kwargs):
|
|
return next(self.paths).open()
|
|
|
|
def __getattr__(self, name):
|
|
return self._respond
|
|
|
|
def should_cache(self):
|
|
return False
|
|
|
|
|
|
class FakeConfig:
|
|
def __init__(self, responder, argvars=None):
|
|
self.responder = responder
|
|
self.args = argparse.Namespace()
|
|
if argvars is not None:
|
|
for key in argvars:
|
|
setattr(self.args, key, argvars[key])
|
|
|
|
def get_loaders(self):
|
|
return self.responder
|
|
|
|
|
|
output = pytest.fixture(lambda: io.StringIO())
|
|
parametrize_format = pytest.mark.parametrize('output_format', [
|
|
oxrhist.Formats.LEDGER,
|
|
oxrhist.Formats.BEANCOUNT,
|
|
])
|
|
|
|
@pytest.fixture(scope='module')
|
|
def single_responder():
|
|
return FakeResponder(relpath('historical1.json'))
|
|
|
|
@pytest.fixture
|
|
def alternate_responder():
|
|
return FakeResponder(
|
|
relpath('historical1.json'),
|
|
relpath('historical2.json'),
|
|
)
|
|
|
|
def build_config(
|
|
responder,
|
|
date,
|
|
amount=None,
|
|
from_currency=None,
|
|
to_currency=None,
|
|
from_date=None,
|
|
output_format=oxrhist.Formats.RAW,
|
|
signed_currencies=None,
|
|
denomination=None,
|
|
base='USD',
|
|
):
|
|
return FakeConfig(responder, {
|
|
'date': date,
|
|
'base': base,
|
|
'amount': None if amount is None else decimal.Decimal(amount),
|
|
'from_currency': from_currency,
|
|
'to_currency': base if to_currency is None else to_currency,
|
|
'from_date': from_date,
|
|
'output_format': output_format,
|
|
'signed_currencies': [base] if signed_currencies is None else signed_currencies,
|
|
'denomination': denomination,
|
|
})
|
|
|
|
def lines_from_run(config, output):
|
|
oxrhist.run(config, output, output)
|
|
output.seek(0)
|
|
return iter(output)
|
|
|
|
def check_fx_amount(config, lines, amount, cost, fx_code, fx_sign=None, price=None):
|
|
if price is None:
|
|
price = cost
|
|
rate_fmt = f'{{}} {re.escape(fx_code)}'
|
|
cost = re.escape(cost) + r'\d*'
|
|
price = re.escape(price) + r'\d*'
|
|
if config.args.output_format is oxrhist.Formats.LEDGER:
|
|
if fx_sign is not None and fx_code in config.args.signed_currencies:
|
|
rate_fmt = f'{re.escape(fx_sign)}{{}}'
|
|
cost_re = '{{={}}}'.format(rate_fmt.format(cost))
|
|
price_re = ' @ {}'.format(rate_fmt.format(price))
|
|
else:
|
|
cost_re = '{{{}}}'.format(rate_fmt.format(cost))
|
|
if config.args.from_date is None:
|
|
price_re = ''
|
|
else:
|
|
price_re = ' @ {}'.format(rate_fmt.format(price))
|
|
pattern = r'^{} {}{}$'.format(re.escape(amount), cost_re, price_re)
|
|
line = next(lines, "<EOF>")
|
|
assert re.match(pattern, line)
|
|
|
|
def check_nonfx_amount(config, lines, amount, code=None, sign=None):
|
|
if config.args.output_format is oxrhist.Formats.LEDGER:
|
|
if code is None:
|
|
code = 'USD'
|
|
sign = '$'
|
|
if code in config.args.signed_currencies and sign is not None:
|
|
expected = f'{sign}{amount}\n'
|
|
else:
|
|
expected = f'{amount} {code}\n'
|
|
else:
|
|
expected = f'{amount} {code or "USD"}\n'
|
|
assert next(lines, "<EOF>") == expected
|
|
|
|
def test_rate_list(single_responder, output, any_date):
|
|
config = build_config(single_responder, any_date)
|
|
lines = lines_from_run(config, output)
|
|
assert next(lines).startswith('1 AED = 0.27229')
|
|
assert next(lines) == '1 USD = 3.67246 AED\n'
|
|
assert next(lines).startswith('1 ALL = 0.0069189')
|
|
assert next(lines) == '1 USD = 144.529793 ALL\n'
|
|
assert next(lines).startswith('1 ANG = 0.55865')
|
|
assert next(lines) == '1 USD = 1.79 ANG\n'
|
|
|
|
def test_one_rate(single_responder, output, any_date):
|
|
config = build_config(single_responder, any_date, from_currency='ANG')
|
|
lines = lines_from_run(config, output)
|
|
assert next(lines).startswith('1 ANG = 0.55865')
|
|
assert next(lines) == '1 USD = 1.79 ANG\n'
|
|
assert next(lines, None) is None
|
|
|
|
def test_conversion(single_responder, output, any_date):
|
|
config = build_config(single_responder, any_date, amount=10, from_currency='AED')
|
|
lines = lines_from_run(config, output)
|
|
assert next(lines) == '10.00 AED = 2.72 USD\n'
|
|
assert next(lines, None) is None
|
|
|
|
def test_back_conversion(single_responder, output, any_date):
|
|
config = build_config(single_responder, any_date,
|
|
amount=2, from_currency='USD', to_currency='ALL')
|
|
lines = lines_from_run(config, output)
|
|
assert next(lines) == '2.00 USD = 289 ALL\n'
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_ledger_rate(single_responder, output, any_date, output_format):
|
|
config = build_config(single_responder, any_date,
|
|
from_currency='ANG', output_format=output_format)
|
|
lines = lines_from_run(config, output)
|
|
check_fx_amount(config, lines, '1 ANG', '0.5586', 'USD', '$')
|
|
check_fx_amount(config, lines, '1 USD', '1.79', 'ANG')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_ledger_conversion(single_responder, output, any_date, output_format):
|
|
config = build_config(single_responder, any_date, from_currency='ALL',
|
|
amount=300, output_format=output_format)
|
|
lines = lines_from_run(config, output)
|
|
check_fx_amount(config, lines, '300 ALL', '0.00691', 'USD', '$')
|
|
check_nonfx_amount(config, lines, '2.08')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_signed_currencies(single_responder, output, any_date, output_format):
|
|
config = build_config(single_responder, any_date, from_currency='AED',
|
|
output_format=output_format, signed_currencies=['EUR'])
|
|
lines = lines_from_run(config, output)
|
|
check_fx_amount(config, lines, '1 AED', '0.272', 'USD', '$')
|
|
check_fx_amount(config, lines, '1 USD', '3.672', 'AED')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_denomination(single_responder, output, any_date, output_format):
|
|
config = build_config(single_responder, any_date,
|
|
from_currency='ANG', to_currency='AED', amount=10,
|
|
output_format=output_format, denomination='USD')
|
|
lines = lines_from_run(config, output)
|
|
check_fx_amount(config, lines, '10.00 ANG', '0.558', 'USD', '$')
|
|
check_fx_amount(config, lines, '20.52 AED', '0.272', 'USD', '$')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_redundant_denomination(single_responder, output, any_date, output_format):
|
|
config = build_config(single_responder, any_date,
|
|
from_currency='ANG', to_currency='USD', amount=10,
|
|
output_format=output_format, denomination='USD')
|
|
lines = lines_from_run(config, output)
|
|
check_fx_amount(config, lines, '10.00 ANG', '0.558', 'USD', '$')
|
|
check_nonfx_amount(config, lines, '5.59')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_from_denomination(single_responder, output, any_date, output_format):
|
|
config = build_config(single_responder, any_date,
|
|
from_currency='USD', to_currency='ALL', amount=10,
|
|
output_format=output_format, denomination='USD')
|
|
lines = lines_from_run(config, output)
|
|
check_nonfx_amount(config, lines, '10.00')
|
|
check_fx_amount(config, lines, '1,445 ALL', '0.00691', 'USD', '$')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_rate_precision_added_as_needed(single_responder, output, any_date, output_format):
|
|
config = build_config(single_responder, any_date,
|
|
from_currency='RUB', to_currency='USD', amount=63805,
|
|
output_format=output_format, denomination='USD')
|
|
lines = lines_from_run(config, output)
|
|
# 63,805 / 57.0763 (the RUB rate) == $1,117.89
|
|
# But using the truncated rate: 63,805 * .01752 == $1,117.86
|
|
# Make sure the rate is specified with enough precision to get the
|
|
# correct conversion amount.
|
|
check_fx_amount(config, lines, '63,805.00 RUB', '0.0175204', 'USD', '$')
|
|
check_nonfx_amount(config, lines, '1,117.89')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_from_date_rates(alternate_responder, output, any_date, output_format):
|
|
config = build_config(alternate_responder, any_date,
|
|
from_currency='ANG', to_currency='AED',
|
|
from_date=any_date, output_format=output_format,
|
|
denomination='USD')
|
|
lines = lines_from_run(config, output)
|
|
check_fx_amount(config, lines, '1 ANG', '1.909', 'AED', None, '2.051')
|
|
check_fx_amount(config, lines, '1 AED', '0.523', 'ANG', None, '0.487')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_from_date_conversion(alternate_responder, output, any_date, output_format):
|
|
config = build_config(alternate_responder, any_date,
|
|
from_currency='ANG', to_currency='AED', amount=10,
|
|
from_date=any_date, output_format=output_format,
|
|
denomination='USD')
|
|
lines = lines_from_run(config, output)
|
|
check_fx_amount(config, lines, '10.00 ANG', '0.507', 'USD', '$', '0.558')
|
|
check_fx_amount(config, lines, '19.10 AED', '0.265', 'USD', '$', '0.272')
|
|
assert next(lines, None) is None
|
|
|
|
@parametrize_format
|
|
def test_rate_consistent_as_cost_and_price(alternate_responder, any_date, output_format):
|
|
config_kwargs = {
|
|
'responder': alternate_responder,
|
|
'amount': 65000,
|
|
'from_currency': 'RUB',
|
|
'output_format': output_format,
|
|
'signed_currencies': (),
|
|
}
|
|
config = build_config(date=any_date, **config_kwargs)
|
|
with io.StringIO() as output:
|
|
lines = lines_from_run(config, output)
|
|
amount, _, _ = next(lines).partition('@')
|
|
expected = amount.replace('\n', ' ')
|
|
future_date = any_date.replace(year=any_date.year + 1)
|
|
config = build_config(date=future_date, from_date=any_date, **config_kwargs)
|
|
with io.StringIO() as output:
|
|
lines = lines_from_run(config, output)
|
|
assert next(lines, "<EOF>").startswith(expected)
|