6efd03570a
Wires the IE + Schwab email parsers into an actual runnable sync. Walks the IMAP mailbox, routes each message by sender domain: - *@investengine.com → invest_engine.parse_invest_engine_email - *@schwab.com → schwab.parse_schwab_email then pushes the resulting Activities through the shared pipeline. broker-sync imap-ingest — new CLI command taking IMAP_HOST/USER/PASSWORD/ DIRECTORY (mirrors the old wealthfolio-sync image's env shape so the Terraform CronJob's existing env wiring works unchanged). Verified: poetry run pytest -q → 109 passed + 1 skipped; mypy strict clean (37 files); ruff + yapf clean.
276 lines
8.7 KiB
Python
276 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from broker_sync.models import Account, AccountType, Activity, ActivityType
|
|
from broker_sync.sinks.wealthfolio import (
|
|
ImportValidationError,
|
|
WealthfolioSink,
|
|
WealthfolioUnauthorizedError,
|
|
)
|
|
|
|
|
|
def _buy() -> Activity:
|
|
return Activity(
|
|
external_id="t212:1",
|
|
account_id="t212-isa",
|
|
account_type=AccountType.ISA,
|
|
date=datetime(2026, 4, 1, 10, 30, tzinfo=UTC),
|
|
activity_type=ActivityType.BUY,
|
|
symbol="VUAG",
|
|
quantity=Decimal("1"),
|
|
unit_price=Decimal("100"),
|
|
currency="GBP",
|
|
)
|
|
|
|
|
|
def _client(handler: httpx.MockTransport, session_path: Path) -> WealthfolioSink:
|
|
return WealthfolioSink(
|
|
base_url="https://wf.test",
|
|
username="viktor",
|
|
password="hunter2",
|
|
session_path=session_path,
|
|
transport=handler,
|
|
)
|
|
|
|
|
|
def _login_ok(req: httpx.Request) -> httpx.Response:
|
|
assert req.url.path == "/api/v1/auth/login"
|
|
body = json.loads(req.content)
|
|
# Wealthfolio 3.2 LoginRequest is password-only.
|
|
assert body == {"password": "hunter2"}
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"authenticated": True,
|
|
"expiresIn": 604800
|
|
},
|
|
headers={"set-cookie": "wf_token=abc123; Path=/api; HttpOnly"},
|
|
)
|
|
|
|
|
|
# -- Login --
|
|
|
|
|
|
async def test_login_persists_cookie(tmp_path: Path) -> None:
|
|
|
|
async def handler(req: httpx.Request) -> httpx.Response:
|
|
return _login_ok(req)
|
|
|
|
sp = tmp_path / "session.json"
|
|
sink = _client(httpx.MockTransport(handler), sp)
|
|
await sink.login()
|
|
|
|
data = json.loads(sp.read_text())
|
|
assert "wf_token" in data["cookies"]
|
|
assert data["cookies"]["wf_token"] == "abc123"
|
|
|
|
|
|
async def test_login_raises_on_401(tmp_path: Path) -> None:
|
|
|
|
async def handler(req: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(401, json={"error": "bad creds"})
|
|
|
|
sink = _client(httpx.MockTransport(handler), tmp_path / "s.json")
|
|
with pytest.raises(WealthfolioUnauthorizedError):
|
|
await sink.login()
|
|
|
|
|
|
# -- Session reuse --
|
|
|
|
|
|
async def test_session_reused_from_disk(tmp_path: Path) -> None:
|
|
sp = tmp_path / "s.json"
|
|
sp.write_text(json.dumps({"cookies": {"wf_token": "cached"}}))
|
|
|
|
calls: list[str] = []
|
|
|
|
async def handler(req: httpx.Request) -> httpx.Response:
|
|
calls.append(req.url.path)
|
|
assert "wf_token=cached" in req.headers.get("cookie", "")
|
|
return httpx.Response(200, json=[])
|
|
|
|
sink = _client(httpx.MockTransport(handler), sp)
|
|
await sink.list_accounts()
|
|
assert calls == ["/api/v1/accounts"]
|
|
|
|
|
|
async def test_401_triggers_single_reauth_and_retry(tmp_path: Path) -> None:
|
|
sp = tmp_path / "s.json"
|
|
sp.write_text(json.dumps({"cookies": {"wf_token": "stale"}}))
|
|
|
|
path_calls: list[tuple[str, str]] = []
|
|
|
|
async def handler(req: httpx.Request) -> httpx.Response:
|
|
path_calls.append((req.method, req.url.path))
|
|
if req.url.path == "/api/v1/accounts" and req.headers.get("cookie", "").endswith("stale"):
|
|
return httpx.Response(401)
|
|
if req.url.path == "/api/v1/auth/login":
|
|
return _login_ok(req)
|
|
# After login the stored cookie is fresh; second GET should succeed.
|
|
return httpx.Response(200, json=[{"id": "a1"}])
|
|
|
|
sink = _client(httpx.MockTransport(handler), sp)
|
|
out = await sink.list_accounts()
|
|
assert out == [{"id": "a1"}]
|
|
assert path_calls == [
|
|
("GET", "/api/v1/accounts"),
|
|
("POST", "/api/v1/auth/login"),
|
|
("GET", "/api/v1/accounts"),
|
|
]
|
|
|
|
|
|
# -- Account ensure --
|
|
|
|
|
|
async def test_ensure_account_returns_existing_wf_id(tmp_path: Path) -> None:
|
|
sp = tmp_path / "s.json"
|
|
sp.write_text(json.dumps({"cookies": {"wf_token": "fresh"}}))
|
|
|
|
posts: list[dict[str, Any]] = []
|
|
|
|
async def handler(req: httpx.Request) -> httpx.Response:
|
|
if req.method == "GET" and req.url.path == "/api/v1/accounts":
|
|
# Wealthfolio stores its own UUID for id; providerAccountId is
|
|
# what we gave it.
|
|
return httpx.Response(
|
|
200,
|
|
json=[{
|
|
"id": "uuid-wf-123",
|
|
"name": "Trading212 ISA",
|
|
"provider": "trading212",
|
|
"providerAccountId": "t212-isa",
|
|
}],
|
|
)
|
|
if req.method == "POST":
|
|
posts.append(json.loads(req.content))
|
|
return httpx.Response(500)
|
|
|
|
sink = _client(httpx.MockTransport(handler), sp)
|
|
acc = Account(
|
|
id="t212-isa",
|
|
name="Trading212 ISA",
|
|
account_type=AccountType.ISA,
|
|
currency="GBP",
|
|
provider="trading212",
|
|
)
|
|
wf_id = await sink.ensure_account(acc)
|
|
assert wf_id == "uuid-wf-123"
|
|
assert posts == [] # no create
|
|
|
|
|
|
async def test_ensure_account_creates_if_missing(tmp_path: Path) -> None:
|
|
sp = tmp_path / "s.json"
|
|
sp.write_text(json.dumps({"cookies": {"wf_token": "fresh"}}))
|
|
|
|
posted: list[dict[str, Any]] = []
|
|
|
|
async def handler(req: httpx.Request) -> httpx.Response:
|
|
if req.method == "GET" and req.url.path == "/api/v1/accounts":
|
|
return httpx.Response(200, json=[])
|
|
if req.method == "POST" and req.url.path == "/api/v1/accounts":
|
|
posted.append(json.loads(req.content))
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"id": "uuid-new-456",
|
|
"provider": "trading212",
|
|
"providerAccountId": "t212-isa",
|
|
},
|
|
)
|
|
return httpx.Response(500)
|
|
|
|
sink = _client(httpx.MockTransport(handler), sp)
|
|
acc = Account(
|
|
id="t212-isa",
|
|
name="Trading212 ISA",
|
|
account_type=AccountType.ISA,
|
|
currency="GBP",
|
|
provider="trading212",
|
|
)
|
|
wf_id = await sink.ensure_account(acc)
|
|
assert wf_id == "uuid-new-456"
|
|
assert len(posted) == 1
|
|
# We no longer send our logical id — Wealthfolio would ignore it. We
|
|
# DO send our id as providerAccountId so Wealthfolio preserves it for
|
|
# future matching.
|
|
assert "id" not in posted[0]
|
|
assert posted[0]["providerAccountId"] == "t212-isa"
|
|
assert posted[0]["accountType"] == "ISA"
|
|
assert posted[0]["currency"] == "GBP"
|
|
assert posted[0]["isActive"] is True
|
|
assert posted[0]["isDefault"] is False
|
|
|
|
|
|
# -- Activity import --
|
|
|
|
|
|
async def test_import_dry_run_then_real(tmp_path: Path) -> None:
|
|
sp = tmp_path / "s.json"
|
|
sp.write_text(json.dumps({"cookies": {"wf_token": "fresh"}}))
|
|
|
|
calls: list[str] = []
|
|
|
|
async def handler(req: httpx.Request) -> httpx.Response:
|
|
calls.append(req.url.path)
|
|
if req.url.path == "/api/v1/activities/import/check":
|
|
# /import/check hydrates and returns a list of ActivityImport.
|
|
return httpx.Response(200,
|
|
json=[
|
|
{
|
|
"symbol": "VUAG",
|
|
"isValid": True,
|
|
"errors": None,
|
|
"assetId": "enriched-asset-uuid",
|
|
"exchangeMic": "XLON",
|
|
},
|
|
])
|
|
if req.url.path == "/api/v1/activities/import":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"activities": [
|
|
{
|
|
"id": "wf-1",
|
|
"external_id": "t212:1"
|
|
},
|
|
],
|
|
},
|
|
)
|
|
return httpx.Response(500)
|
|
|
|
sink = _client(httpx.MockTransport(handler), sp)
|
|
out = await sink.import_activities([_buy()])
|
|
assert calls == [
|
|
"/api/v1/activities/import/check",
|
|
"/api/v1/activities/import",
|
|
]
|
|
assert out == [{"id": "wf-1", "external_id": "t212:1"}]
|
|
|
|
|
|
async def test_import_halts_on_validation_failure(tmp_path: Path) -> None:
|
|
sp = tmp_path / "s.json"
|
|
sp.write_text(json.dumps({"cookies": {"wf_token": "fresh"}}))
|
|
|
|
calls: list[str] = []
|
|
|
|
async def handler(req: httpx.Request) -> httpx.Response:
|
|
calls.append(req.url.path)
|
|
if req.url.path == "/api/v1/activities/import/check":
|
|
return httpx.Response(
|
|
400,
|
|
json={"errors": ["row 1: unknown symbol"]},
|
|
)
|
|
return httpx.Response(500)
|
|
|
|
sink = _client(httpx.MockTransport(handler), sp)
|
|
with pytest.raises(ImportValidationError, match="unknown symbol"):
|
|
await sink.import_activities([_buy()])
|
|
assert calls == ["/api/v1/activities/import/check"] # real import never hit
|