e6ef1fce97
`from __future__ import annotations` makes the quoting unnecessary and ruff UP037 flags it. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
156 lines
6.1 KiB
Python
156 lines
6.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, date, datetime
|
|
from decimal import Decimal
|
|
from typing import TYPE_CHECKING
|
|
|
|
from broker_sync.models import AccountType, Activity, ActivityType
|
|
|
|
if TYPE_CHECKING:
|
|
from pytest import MonkeyPatch
|
|
from broker_sync.providers.imap import (
|
|
_IE_GIA_ACCOUNT_ID,
|
|
_IE_ISA_ACCOUNT_ID,
|
|
_split_ie_by_isa_cap,
|
|
_uk_tax_year_start,
|
|
)
|
|
|
|
|
|
def _buy(on: datetime, qty: str, price: str) -> Activity:
|
|
return Activity(
|
|
external_id=f"invest-engine:{on.isoformat()}|{qty}|{price}",
|
|
account_id=_IE_ISA_ACCOUNT_ID,
|
|
account_type=AccountType.ISA,
|
|
date=on,
|
|
activity_type=ActivityType.BUY,
|
|
currency="GBP",
|
|
symbol="VUAG",
|
|
quantity=Decimal(qty),
|
|
unit_price=Decimal(price),
|
|
)
|
|
|
|
|
|
def test_uk_tax_year_start_before_april_6_rolls_back() -> None:
|
|
assert _uk_tax_year_start(datetime(2025, 4, 5, tzinfo=UTC)) == date(2024, 4, 6)
|
|
assert _uk_tax_year_start(datetime(2025, 4, 6, tzinfo=UTC)) == date(2025, 4, 6)
|
|
assert _uk_tax_year_start(datetime(2025, 1, 15, tzinfo=UTC)) == date(2024, 4, 6)
|
|
assert _uk_tax_year_start(datetime(2024, 4, 7, tzinfo=UTC)) == date(2024, 4, 6)
|
|
|
|
|
|
def test_single_tax_year_under_cap_stays_isa() -> None:
|
|
acts = [
|
|
_buy(datetime(2024, 5, 1, tzinfo=UTC), "100", "50"), # £5000
|
|
_buy(datetime(2024, 8, 1, tzinfo=UTC), "100", "80"), # £8000
|
|
]
|
|
routed = _split_ie_by_isa_cap(acts)
|
|
assert all(a.account_id == _IE_ISA_ACCOUNT_ID for a in routed)
|
|
assert all(a.account_type is AccountType.ISA for a in routed)
|
|
|
|
|
|
def test_overflow_past_cap_flips_to_gia() -> None:
|
|
acts = [
|
|
_buy(datetime(2024, 5, 1, tzinfo=UTC), "100", "80"), # £8,000
|
|
# +£12,000 → £20,000 total; prev £8k < cap → ISA
|
|
_buy(datetime(2024, 6, 1, tzinfo=UTC), "150", "80"),
|
|
_buy(datetime(2024, 7, 1, tzinfo=UTC), "10", "80"), # prev £20,000 ≥ cap → GIA
|
|
_buy(datetime(2024, 8, 1, tzinfo=UTC), "10", "80"), # GIA
|
|
]
|
|
routed = _split_ie_by_isa_cap(acts)
|
|
assert routed[0].account_id == _IE_ISA_ACCOUNT_ID
|
|
assert routed[1].account_id == _IE_ISA_ACCOUNT_ID
|
|
assert routed[2].account_id == _IE_GIA_ACCOUNT_ID
|
|
assert routed[2].account_type is AccountType.GIA
|
|
assert routed[3].account_id == _IE_GIA_ACCOUNT_ID
|
|
|
|
|
|
def test_tax_year_boundary_resets_cap() -> None:
|
|
acts = [
|
|
# 2023-24 tax year: £20k in ISA, plus one in GIA
|
|
_buy(datetime(2023, 5, 1, tzinfo=UTC), "400", "50"), # £20,000 → ISA (prev 0 < cap)
|
|
_buy(datetime(2024, 1, 1, tzinfo=UTC), "100", "50"), # GIA (prev 20k)
|
|
# 2024-25 tax year starts 2024-04-06 — cap resets
|
|
_buy(datetime(2024, 5, 1, tzinfo=UTC), "100", "50"), # ISA (prev 0 for new year)
|
|
]
|
|
routed = _split_ie_by_isa_cap(acts)
|
|
assert routed[0].account_id == _IE_ISA_ACCOUNT_ID
|
|
assert routed[1].account_id == _IE_GIA_ACCOUNT_ID
|
|
assert routed[2].account_id == _IE_ISA_ACCOUNT_ID
|
|
|
|
|
|
def test_out_of_order_activities_sorted_before_cap_applied() -> None:
|
|
acts = [
|
|
_buy(datetime(2024, 8, 1, tzinfo=UTC), "10", "80"), # later date but given first
|
|
_buy(datetime(2024, 5, 1, tzinfo=UTC), "250", "80"), # earlier, £20,000 → ISA
|
|
]
|
|
routed = _split_ie_by_isa_cap(acts)
|
|
by_date = {a.date: a for a in routed}
|
|
assert by_date[datetime(2024, 5, 1, tzinfo=UTC)].account_id == _IE_ISA_ACCOUNT_ID
|
|
assert by_date[datetime(2024, 8, 1, tzinfo=UTC)].account_id == _IE_GIA_ACCOUNT_ID
|
|
|
|
|
|
def test_non_ie_activities_passed_through_unchanged() -> None:
|
|
schwab_act = Activity(
|
|
external_id="schwab:abc",
|
|
account_id="schwab-workplace",
|
|
account_type=AccountType.GIA,
|
|
date=datetime(2024, 5, 1, tzinfo=UTC),
|
|
activity_type=ActivityType.SELL,
|
|
currency="USD",
|
|
symbol="META",
|
|
quantity=Decimal("10"),
|
|
unit_price=Decimal("500"),
|
|
)
|
|
routed = _split_ie_by_isa_cap([schwab_act])
|
|
assert routed[0].account_id == "schwab-workplace"
|
|
assert routed[0].account_type is AccountType.GIA
|
|
|
|
|
|
def test_exclude_invest_engine_skips_ie_emails(monkeypatch: MonkeyPatch) -> None:
|
|
"""BROKER_SYNC_IMAP_EXCLUDE_PROVIDERS=invest-engine should skip IE messages
|
|
so we don't duplicate IE buys already ingested via the bearer-token API path.
|
|
Schwab routing must remain unaffected."""
|
|
from broker_sync.providers import imap as imap_mod
|
|
from broker_sync.providers.parsers import invest_engine as ie_parser
|
|
|
|
ie_email = (
|
|
b"From: noreply@investengine.com\r\n"
|
|
b"Subject: VUAG Bought\r\n"
|
|
b"Content-Type: text/plain\r\n\r\n"
|
|
b"Vanguard S&P 500: VUAG Bought 10.0 @ 100.0 per share Total: 1000.00\r\n"
|
|
)
|
|
schwab_email = (
|
|
b"From: donotreply@schwab.com\r\n"
|
|
b"Subject: Order Confirmed\r\n"
|
|
b"Content-Type: text/html\r\n\r\n"
|
|
b"<html><body>no-op</body></html>\r\n"
|
|
)
|
|
monkeypatch.setattr(imap_mod, "_fetch_all", lambda _: [ie_email, schwab_email])
|
|
monkeypatch.setattr(ie_parser, "parse_invest_engine_email", lambda raw: [object()])
|
|
monkeypatch.setattr(imap_mod, "parse_schwab_email", lambda html: [object()])
|
|
|
|
creds = imap_mod.ImapCreds(host="h", user="u", password="p", directory="d")
|
|
|
|
monkeypatch.setenv("BROKER_SYNC_IMAP_EXCLUDE_PROVIDERS", "invest-engine")
|
|
out_excluded = imap_mod.fetch_activities(creds)
|
|
# IE skipped → only the schwab activity is emitted
|
|
assert len(out_excluded) == 1
|
|
|
|
monkeypatch.delenv("BROKER_SYNC_IMAP_EXCLUDE_PROVIDERS", raising=False)
|
|
out_default = imap_mod.fetch_activities(creds)
|
|
# Both providers fire when env unset
|
|
assert len(out_default) == 2
|
|
|
|
|
|
def test_schwab_subdomain_sender_matches() -> None:
|
|
"""Real Schwab trade emails come from `donotreply@mail.schwab.com`
|
|
(subdomain), not just `donotreply@schwab.com`. The matcher must
|
|
accept either form."""
|
|
from broker_sync.providers.imap import _SCHWAB_SENDERS
|
|
# Verify the static set works
|
|
assert "donotreply@schwab.com" in _SCHWAB_SENDERS
|
|
# Verify the subdomain suffix check
|
|
for addr in (
|
|
"donotreply@mail.schwab.com",
|
|
"wealthnotify@equityawards.schwab.com",
|
|
):
|
|
assert addr.endswith(".schwab.com"), addr
|