63 lines
2.4 KiB
Python
63 lines
2.4 KiB
Python
|
from datetime import date
|
||
|
|
from decimal import Decimal
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from broker_sync.fx import FxCache, convert_to_gbp
|
||
|
|
from broker_sync.models import FxRateSource
|
||
|
|
|
||
|
|
|
||
|
|
def test_cache_roundtrip(tmp_path: Path) -> None:
|
||
|
|
c = FxCache(tmp_path / "fx.db")
|
||
|
|
c.put("USD", date(2026, 4, 1), Decimal("0.79"), FxRateSource.ECB_LIVE)
|
||
|
|
hit = c.get("USD", date(2026, 4, 1))
|
||
|
|
assert hit is not None
|
||
|
|
assert hit[0] == Decimal("0.79")
|
||
|
|
assert hit[1] is FxRateSource.ECB_LIVE
|
||
|
|
|
||
|
|
|
||
|
|
def test_cache_miss(tmp_path: Path) -> None:
|
||
|
|
c = FxCache(tmp_path / "fx.db")
|
||
|
|
assert c.get("USD", date(2026, 4, 1)) is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_put_overwrites_on_source_upgrade(tmp_path: Path) -> None:
|
||
|
|
# Reconciliation job upgrades ECB_LIVE → HMRC_MONTHLY for historical rows
|
||
|
|
c = FxCache(tmp_path / "fx.db")
|
||
|
|
c.put("USD", date(2026, 3, 1), Decimal("0.79"), FxRateSource.ECB_LIVE)
|
||
|
|
c.put("USD", date(2026, 3, 1), Decimal("0.791"), FxRateSource.HMRC_MONTHLY)
|
||
|
|
hit = c.get("USD", date(2026, 3, 1))
|
||
|
|
assert hit is not None
|
||
|
|
assert hit[0] == Decimal("0.791")
|
||
|
|
assert hit[1] is FxRateSource.HMRC_MONTHLY
|
||
|
|
|
||
|
|
|
||
|
|
def test_gbp_passthrough_is_identity() -> None:
|
||
|
|
# currency == GBP → rate is 1.0, no network needed
|
||
|
|
amount_gbp, rate, source = convert_to_gbp(amount=Decimal("100"),
|
||
|
|
currency="GBP",
|
||
|
|
on_date=date(2026, 4, 1),
|
||
|
|
cache=None)
|
||
|
|
assert amount_gbp == Decimal("100")
|
||
|
|
assert rate == Decimal("1")
|
||
|
|
assert source is FxRateSource.ECB_LIVE
|
||
|
|
|
||
|
|
|
||
|
|
def test_convert_uses_cached_rate(tmp_path: Path) -> None:
|
||
|
|
c = FxCache(tmp_path / "fx.db")
|
||
|
|
c.put("USD", date(2026, 4, 1), Decimal("0.80"), FxRateSource.ECB_LIVE)
|
||
|
|
amount_gbp, rate, source = convert_to_gbp(amount=Decimal("100"),
|
||
|
|
currency="USD",
|
||
|
|
on_date=date(2026, 4, 1),
|
||
|
|
cache=c)
|
||
|
|
assert amount_gbp == Decimal("80.00")
|
||
|
|
assert rate == Decimal("0.80")
|
||
|
|
assert source is FxRateSource.ECB_LIVE
|
||
|
|
|
||
|
|
|
||
|
|
def test_convert_raises_on_cache_miss(tmp_path: Path) -> None:
|
||
|
|
c = FxCache(tmp_path / "fx.db")
|
||
|
|
with pytest.raises(LookupError, match="USD.*2026-04-01"):
|
||
|
|
convert_to_gbp(amount=Decimal("100"), currency="USD", on_date=date(2026, 4, 1), cache=c)
|