261 lines
8.5 KiB
Python
261 lines
8.5 KiB
Python
"""Tests for foreign exchange functionality."""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from unittest.mock import patch
|
|
|
|
from agenda.fx import get_rates, read_cached_rates
|
|
|
|
|
|
class TestReadCachedRates:
|
|
"""Test the read_cached_rates function."""
|
|
|
|
def test_read_cached_rates_none_filename(self) -> None:
|
|
"""Test with None filename returns empty dict."""
|
|
result = read_cached_rates(None, ["USD", "EUR"])
|
|
assert result == {}
|
|
|
|
def test_read_cached_rates_valid_file(self) -> None:
|
|
"""Test reading valid exchangerate.host cached rates file."""
|
|
currencies = ["USD", "EUR", "JPY"]
|
|
data = {
|
|
"quotes": {
|
|
"GBPUSD": 1.25,
|
|
"GBPEUR": 1.15,
|
|
"GBPJPY": 150.0,
|
|
"GBPCAD": 1.70, # Not requested
|
|
}
|
|
}
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
|
json.dump(data, f)
|
|
filepath = f.name
|
|
|
|
try:
|
|
result = read_cached_rates(filepath, currencies)
|
|
|
|
assert len(result) == 3
|
|
assert result["USD"] == Decimal("1.25")
|
|
assert result["EUR"] == Decimal("1.15")
|
|
assert result["JPY"] == Decimal("150.0")
|
|
assert "CAD" not in result # Not requested
|
|
|
|
finally:
|
|
os.unlink(filepath)
|
|
|
|
def test_read_cached_rates_frankfurter_file(self) -> None:
|
|
"""Test reading valid Frankfurter cached rates file."""
|
|
currencies = ["USD", "EUR", "JPY"]
|
|
data = [
|
|
{"date": "2026-06-20", "base": "GBP", "quote": "USD", "rate": 1.25},
|
|
{"date": "2026-06-20", "base": "GBP", "quote": "EUR", "rate": 1.15},
|
|
{"date": "2026-06-20", "base": "GBP", "quote": "JPY", "rate": 150.0},
|
|
{
|
|
"date": "2026-06-20",
|
|
"base": "GBP",
|
|
"quote": "CAD",
|
|
"rate": 1.70,
|
|
}, # Not requested
|
|
]
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
|
json.dump(data, f)
|
|
filepath = f.name
|
|
|
|
try:
|
|
result = read_cached_rates(filepath, currencies)
|
|
|
|
assert len(result) == 3
|
|
assert result["USD"] == Decimal("1.25")
|
|
assert result["EUR"] == Decimal("1.15")
|
|
assert result["JPY"] == Decimal("150.0")
|
|
assert "CAD" not in result # Not requested
|
|
|
|
finally:
|
|
os.unlink(filepath)
|
|
|
|
def test_read_cached_rates_missing_currencies(self) -> None:
|
|
"""Test with some currencies missing from data."""
|
|
currencies = ["USD", "EUR", "CHF"] # CHF not in data
|
|
data = {
|
|
"quotes": {
|
|
"GBPUSD": 1.25,
|
|
"GBPEUR": 1.15,
|
|
# GBPCHF missing
|
|
}
|
|
}
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
|
json.dump(data, f)
|
|
filepath = f.name
|
|
|
|
try:
|
|
result = read_cached_rates(filepath, currencies)
|
|
|
|
assert len(result) == 2
|
|
assert result["USD"] == Decimal("1.25")
|
|
assert result["EUR"] == Decimal("1.15")
|
|
assert "CHF" not in result # Missing from data
|
|
|
|
finally:
|
|
os.unlink(filepath)
|
|
|
|
def test_read_cached_rates_empty_currencies_list(self) -> None:
|
|
"""Test with empty currencies list."""
|
|
data = {
|
|
"quotes": {
|
|
"GBPUSD": 1.25,
|
|
"GBPEUR": 1.15,
|
|
}
|
|
}
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
|
json.dump(data, f)
|
|
filepath = f.name
|
|
|
|
try:
|
|
result = read_cached_rates(filepath, [])
|
|
assert result == {}
|
|
|
|
finally:
|
|
os.unlink(filepath)
|
|
|
|
def test_read_cached_rates_no_quotes_key(self) -> None:
|
|
"""Test with data missing quotes key."""
|
|
currencies = ["USD", "EUR"]
|
|
data = {"other_key": "value"} # No quotes key
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
|
json.dump(data, f)
|
|
filepath = f.name
|
|
|
|
try:
|
|
assert read_cached_rates(filepath, currencies) == {}
|
|
|
|
finally:
|
|
os.unlink(filepath)
|
|
|
|
def test_read_cached_rates_decimal_precision(self) -> None:
|
|
"""Test that rates are returned as Decimal with proper precision."""
|
|
currencies = ["USD"]
|
|
data = {"quotes": {"GBPUSD": 1.234567890123456789}} # High precision
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
|
json.dump(data, f)
|
|
filepath = f.name
|
|
|
|
try:
|
|
result = read_cached_rates(filepath, currencies)
|
|
|
|
assert isinstance(result["USD"], Decimal)
|
|
# Should preserve reasonable precision from the JSON
|
|
# Python's JSON precision may be limited to float precision
|
|
expected = Decimal("1.234567890123456789")
|
|
assert abs(result["USD"] - expected) < Decimal("0.0000000000000001")
|
|
|
|
finally:
|
|
os.unlink(filepath)
|
|
|
|
def test_read_cached_rates_various_currency_codes(self) -> None:
|
|
"""Test with various currency codes."""
|
|
currencies = ["USD", "EUR", "JPY", "CHF", "CAD", "AUD"]
|
|
data = {
|
|
"quotes": {
|
|
"GBPUSD": 1.25,
|
|
"GBPEUR": 1.15,
|
|
"GBPJPY": 150.0,
|
|
"GBPCHF": 1.12,
|
|
"GBPCAD": 1.70,
|
|
"GBPAUD": 1.85,
|
|
}
|
|
}
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
|
json.dump(data, f)
|
|
filepath = f.name
|
|
|
|
try:
|
|
result = read_cached_rates(filepath, currencies)
|
|
|
|
assert len(result) == 6
|
|
for currency in currencies:
|
|
assert currency in result
|
|
assert isinstance(result[currency], Decimal)
|
|
|
|
finally:
|
|
os.unlink(filepath)
|
|
|
|
|
|
def test_get_rates_ignores_old_usage_limit_attempt_when_using_frankfurter() -> None:
|
|
"""An old provider usage-limit response should not block Frankfurter refreshes."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
fx_dir = os.path.join(temp_dir, "fx")
|
|
os.mkdir(fx_dir)
|
|
suffix = "EUR,USD_to_GBP.json"
|
|
valid_file = os.path.join(fx_dir, f"2026-06-19_23:35_{suffix}")
|
|
usage_limit_file = os.path.join(fx_dir, f"2026-06-20_11:35_{suffix}")
|
|
|
|
with open(valid_file, "w") as f:
|
|
json.dump(
|
|
{"success": True, "quotes": {"GBPEUR": 1.15, "GBPUSD": 1.25}},
|
|
f,
|
|
)
|
|
|
|
with open(usage_limit_file, "w") as f:
|
|
json.dump(
|
|
{
|
|
"success": False,
|
|
"error": {
|
|
"code": 104,
|
|
"type": "usage_limit_reached",
|
|
"info": "Your monthly usage limit has been reached.",
|
|
},
|
|
},
|
|
f,
|
|
)
|
|
|
|
config = {
|
|
"CURRENCIES": ["USD", "EUR"],
|
|
"DATA_DIR": temp_dir,
|
|
"EXCHANGERATE_ACCESS_KEY": "key",
|
|
"OFFLINE_MODE": False,
|
|
"FX_CACHE_TTL_HOURS": 1,
|
|
"FX_FAILURE_RETRY_HOURS": 24,
|
|
}
|
|
|
|
with patch("agenda.fx.datetime") as mock_datetime, patch(
|
|
"agenda.fx.httpx.Client"
|
|
) as mock_client:
|
|
mock_datetime.now.return_value = datetime(2026, 6, 20, 12, 35)
|
|
mock_datetime.strptime.side_effect = datetime.strptime
|
|
mock_response = (
|
|
mock_client.return_value.__enter__.return_value.get.return_value
|
|
)
|
|
mock_response.text = json.dumps(
|
|
[
|
|
{
|
|
"date": "2026-06-20",
|
|
"base": "GBP",
|
|
"quote": "EUR",
|
|
"rate": 1.16,
|
|
},
|
|
{
|
|
"date": "2026-06-20",
|
|
"base": "GBP",
|
|
"quote": "USD",
|
|
"rate": 1.26,
|
|
},
|
|
]
|
|
)
|
|
|
|
result = get_rates(config)
|
|
|
|
assert result == {"USD": Decimal("1.26"), "EUR": Decimal("1.16")}
|
|
mock_client.return_value.__enter__.return_value.get.assert_called_once_with(
|
|
"https://api.frankfurter.dev/v2/rates",
|
|
params={"base": "GBP", "quotes": "EUR,USD"},
|
|
timeout=10,
|
|
)
|