from __future__ import annotations
from decimal import Decimal
from broker_sync.models import AccountType, ActivityType
from broker_sync.providers.parsers.schwab import parse_schwab_email
_SELL = """
| Date | Jan 23, 2025 |
| Action | Sold |
| Quantity | 100.0 |
| Ticker | META |
| Price | $612.34 |
"""
_BUY = """
| 2024-11-15 |
| Bought |
| 5.5 |
| AAPL |
| $225.00 |
"""
_MALFORMED = "no transaction here"
_MISSING_CELLS = """
"""
def test_sell_email_parses_to_one_sell_activity() -> None:
acts = parse_schwab_email(_SELL)
assert len(acts) == 1
a = acts[0]
assert a.activity_type is ActivityType.SELL
assert a.symbol == "META"
assert a.quantity == Decimal("100.0")
assert a.unit_price == Decimal("612.34")
assert a.currency == "USD"
assert a.account_id == "schwab-workplace"
assert a.account_type is AccountType.GIA
assert a.date.date().isoformat() == "2025-01-23"
def test_buy_email_becomes_buy_activity() -> None:
acts = parse_schwab_email(_BUY)
assert len(acts) == 1
a = acts[0]
assert a.activity_type is ActivityType.BUY
assert a.symbol == "AAPL"
assert a.quantity == Decimal("5.5")
assert a.unit_price == Decimal("225.00")
def test_malformed_email_returns_empty_list() -> None:
# No matching td cells at all.
assert parse_schwab_email(_MALFORMED) == []
def test_missing_cells_returns_empty_list() -> None:
# Only 2 of the 5 required cells — parser must bail cleanly.
assert parse_schwab_email(_MISSING_CELLS) == []
def test_external_id_is_stable_across_reruns() -> None:
# Same email → same external_id (deterministic, not timestamp-based).
a1 = parse_schwab_email(_SELL)[0]
a2 = parse_schwab_email(_SELL)[0]
assert a1.external_id == a2.external_id
def test_price_with_commas_parses() -> None:
html = _SELL.replace("$612.34", "$1,612.34")
# The first activity is the inferred BUY (date 2025-01-23 ≥ 2026-04-01? no →
# only one activity for this old-dated email), so index 0 is the SELL.
acts = parse_schwab_email(html)
sell = next(a for a in acts if a.activity_type is ActivityType.SELL)
assert sell.unit_price == Decimal("1612.34")
# --- Inferred vest BUY ---------------------------------------------------
def _recent_sell(date_iso: str = "2026-05-19", qty: str = "55", price: str = "609.35") -> str:
return f"""
| {date_iso} |
| Sold |
| {qty} |
| META |
| ${price} |
"""
def test_recent_sell_emits_paired_buy() -> None:
"""SELL dated on/after the synthesis boundary triggers a paired BUY."""
acts = parse_schwab_email(_recent_sell())
assert len(acts) == 2
buy = next(a for a in acts if a.activity_type is ActivityType.BUY)
sell = next(a for a in acts if a.activity_type is ActivityType.SELL)
assert buy.quantity == sell.quantity == Decimal("55")
assert buy.unit_price == sell.unit_price == Decimal("609.35")
assert buy.date == sell.date
assert buy.symbol == sell.symbol == "META"
assert "schwab-vest-inferred-from-same-day-sell" in (buy.notes or "")
assert buy.external_id == "schwab:vest:2026-05-19:META:BUY:55"
assert sell.external_id == "schwab:2026-05-19:META:SELL:55"
def test_old_sell_emits_only_sell() -> None:
"""SELL dated before 2026-04-01 (default boundary) skips the paired BUY —
those vests already have csv-sourced BUY rows in Wealthfolio."""
acts = parse_schwab_email(_recent_sell(date_iso="2025-08-19"))
assert len(acts) == 1
assert acts[0].activity_type is ActivityType.SELL
def test_boundary_env_var_overrides(monkeypatch: object) -> None:
"""The synthesis boundary is configurable via env var."""
import os
os.environ["SCHWAB_VEST_INFER_FROM_DATE"] = "2025-01-01"
try:
acts = parse_schwab_email(_recent_sell(date_iso="2025-08-19"))
assert len(acts) == 2 # now in scope
finally:
del os.environ["SCHWAB_VEST_INFER_FROM_DATE"]
def test_buy_email_does_not_emit_inferred_buy() -> None:
"""BUY-direction emails (rare for workplace account) don't get paired."""
acts = parse_schwab_email(_BUY.replace("2024-11-15", "2026-05-15"))
assert len(acts) == 1
assert acts[0].activity_type is ActivityType.BUY