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_invest_engine_skipped_by_default(monkeypatch: MonkeyPatch) -> None: """InvestEngine messages MUST be skipped by default, even with no env set. Post-mortem 2026-05-27: any code path that doesn't set the cron's env (e.g. `kubectl run --rm` or devvm `poetry run`) was re-importing IE BUYs through this IMAP path. The opt-out env var was a foot-gun. Invariant now: structural default skip; opt back in only with BROKER_SYNC_IMAP_INCLUDE_PROVIDERS. """ 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"no-op\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") # Default (no env): IE skipped, Schwab parsed. monkeypatch.delenv("BROKER_SYNC_IMAP_EXCLUDE_PROVIDERS", raising=False) monkeypatch.delenv("BROKER_SYNC_IMAP_INCLUDE_PROVIDERS", raising=False) out_default = imap_mod.fetch_activities(creds) assert len(out_default) == 1, "IE must be skipped by default; only Schwab emitted" def test_invest_engine_opt_in_via_include_env(monkeypatch: MonkeyPatch) -> None: """Setting BROKER_SYNC_IMAP_INCLUDE_PROVIDERS=invest-engine re-enables IE parsing (escape hatch for the legacy IMAP path).""" 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\r\nirrelevant\r\n" schwab_email = b"From: donotreply@schwab.com\r\n\r\n\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_INCLUDE_PROVIDERS", "invest-engine") monkeypatch.delenv("BROKER_SYNC_IMAP_EXCLUDE_PROVIDERS", raising=False) out = imap_mod.fetch_activities(creds) assert len(out) == 2, "INCLUDE=invest-engine must re-enable IE parsing" def test_exclude_schwab_still_works(monkeypatch: MonkeyPatch) -> None: """EXCLUDE env still works for other providers (forward-compat).""" from broker_sync.providers import imap as imap_mod from broker_sync.providers.parsers import invest_engine as ie_parser schwab_email = b"From: donotreply@schwab.com\r\n\r\n\r\n" monkeypatch.setattr(imap_mod, "_fetch_all", lambda _: [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", "schwab") monkeypatch.delenv("BROKER_SYNC_IMAP_INCLUDE_PROVIDERS", raising=False) out = imap_mod.fetch_activities(creds) assert len(out) == 0, "Schwab must be skipped when in EXCLUDE list" def test_include_overrides_default_and_exclude(monkeypatch: MonkeyPatch) -> None: """INCLUDE wins over both the structural default and EXCLUDE env var.""" from broker_sync.providers import imap as imap_mod monkeypatch.setenv("BROKER_SYNC_IMAP_EXCLUDE_PROVIDERS", "invest-engine,schwab") monkeypatch.setenv("BROKER_SYNC_IMAP_INCLUDE_PROVIDERS", "invest-engine") resolved = imap_mod._resolve_excluded_providers() assert "invest-engine" not in resolved assert "schwab" in resolved 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