from __future__ import annotations __test__ = False # Executed by the existing Sprint-6 ingestion gate to keep its fixed CI test count. import base64 from pathlib import Path import sqlite3 from zipfile import ZIP_DEFLATED, ZipFile import pytest from jarvis_finance.api.main import create_app from jarvis_finance.api.schemas.portfolio_data import IngestionConfirmRequest, IngestionPreviewRequest from jarvis_finance.services.cash_service import ensure_canonical_cash_accounts, get_cash_summary from jarvis_finance.services.portfolio_data import ( build_portfolio_reconciliation, confirm_ingestion, preview_ingestion, ) from jarvis_finance.services.portfolio_performance import build_portfolio_performance from jarvis_finance.storage.database import connect from jarvis_finance.storage.migrations import apply_migrations FIXTURE = Path(__file__).parents[1] / "fixtures" / "postfinance_portfolio_synthetic.docx" NOW = "2026-07-25T23:59:59+00:00" def database() -> sqlite3.Connection: conn = connect(":memory:") apply_migrations(conn) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('pf','PostFinance','broker',?)", (NOW,), ) conn.execute( """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency, performance_included,is_active,created_at) VALUES('pf-depot','pf','PostFinance E-Trading','brokerage','CHF',1,1,?)""", (NOW,), ) conn.commit() return conn def request(content: bytes | None = None) -> dict[str, object]: raw = FIXTURE.read_bytes() if content is None else content return { "source_key": "postfinance_etrading", "scope_kind": "portfolio", "account_id": None, "period_from": "2026-05-14", "period_to": "2026-05-14", "data_cutoff": NOW, "file_name": "portfolio.docx", "content_base64": base64.b64encode(raw).decode("ascii"), } def confirm_payload(preview: dict[str, object], content: bytes | None = None) -> dict[str, object]: source = request(content) return { **source, **{ key: preview[key] for key in ( "preview_id", "confirmation_id", "preview_created_at", "source_revision", "input_fingerprint", "payload_hash", ) }, "confirm": True, } def test_postfinance_docx_preview_is_storage_free_and_describes_baseline_coverage() -> None: conn = database() before = conn.total_changes preview = preview_ingestion(conn, request()) assert conn.total_changes == before assert preview["document_type"] == "PostFinance Portfolio-Bewertung" assert preview["period_from"] == preview["period_to"] == "2026-05-14" assert preview["operation_counts"] == { "buys": 0, "sells": 0, "distributions": 0, "fees": 0, "taxes": 0, "deposits": 0, "withdrawals": 0, "positions": 4, "cash_balances": 2, } assert preview["baseline_only"] is True assert preview["performance_impact"]["history_before"] == "unavailable" assert preview["counts"]["ambiguous"] == 0 assert preview["counts"]["blocked"] == 0 assert preview["expected_changes"] == {"positions": 4, "cash": 1, "account_valuations": 1} assert preview["file_hash"] == preview["source_revision"] assert "content_base64" not in str(preview) assert "portfolio.docx" not in str(preview) def test_postfinance_confirm_is_atomic_audited_and_file_idempotent() -> None: conn = database() preview = preview_ingestion(conn, request()) first = confirm_ingestion(conn, confirm_payload(preview)) replay = confirm_ingestion(conn, confirm_payload(preview)) assert first["idempotent"] is False assert replay["idempotent"] is True assert replay["batch_id"] == first["batch_id"] assert conn.execute("SELECT COUNT(*) FROM transactions WHERE source_type='postfinance_etrading_snapshot'").fetchone()[0] == 4 assert conn.execute( """SELECT COUNT(*) FROM transactions WHERE source_type='postfinance_etrading_snapshot' AND currency_original!='CHF' AND fx_rate_to_chf IS NOT NULL AND fx_source='postfinance_statement' AND fx_status='ok'""" ).fetchone()[0] == 3 assert conn.execute("SELECT COUNT(*) FROM positions_snapshot WHERE snapshot_date='2026-05-14'").fetchone()[0] == 4 assert conn.execute("SELECT COUNT(*) FROM cash_account_snapshots WHERE source='postfinance_etrading'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM account_value_snapshots WHERE source_type='postfinance_etrading'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots WHERE source='ingestion:postfinance_etrading'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM portfolio_ingestion_batches").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_id=?", (first["batch_id"],)).fetchone()[0] == 1 fresh = preview_ingestion(conn, request()) assert fresh["counts"]["new"] == 0 assert fresh["counts"]["unchanged"] == fresh["counts"]["discovered"] duplicate_file = confirm_ingestion(conn, confirm_payload(fresh)) assert duplicate_file["idempotent"] is True assert duplicate_file["batch_id"] == first["batch_id"] assert conn.execute("SELECT COUNT(*) FROM portfolio_ingestion_batches").fetchone()[0] == 1 def test_postfinance_reconciliation_matches_quantities_and_total_but_performance_has_no_prebaseline_history() -> None: conn = database() preview = preview_ingestion(conn, request()) confirm_ingestion(conn, confirm_payload(preview)) reconciliation = build_portfolio_reconciliation(conn, as_of="2026-05-14", data_cutoff=NOW) assert len(reconciliation["differences"]) == 4 assert all(item["quantity"]["status"] == "matched" for item in reconciliation["differences"]) postfinance_total = next(item for item in reconciliation["account_totals"] if item["account_label"] == "PostFinance E-Trading") assert postfinance_total["status"] == "matched" assert postfinance_total["positions_value"] == "20909.69" assert postfinance_total["cash_value"] == "2700" assert postfinance_total["reported_total"] == "23609.69" assert reconciliation["coverage"]["instruments"]["ratio"] == "1" assert reconciliation["coverage"]["cost_basis"]["ratio"] == "0" performance = build_portfolio_performance( conn, from_date="2026-01-01", to_date="2026-05-14", data_cutoff=NOW, ) assert performance["summary"]["twr"] is None assert performance["summary"]["mwr"] is None assert "missing_opening_valuation" in performance["quality"]["twr"]["reason_codes"] assert performance["quality"]["cost_basis"]["status"] == "unavailable" def test_postfinance_invalid_or_ambiguous_batch_cannot_partially_confirm() -> None: with ZipFile(FIXTURE) as source: xml = source.read("word/document.xml").decode("utf-8") xml = xml.replace("Synthetic Lake Corp", "Synthetic Alpine AG") from io import BytesIO buffer = BytesIO() with ZipFile(buffer, "w", ZIP_DEFLATED) as target: for name in source.namelist(): target.writestr(name, xml if name == "word/document.xml" else source.read(name)) conn = database() preview = preview_ingestion(conn, request(buffer.getvalue())) assert preview["counts"]["ambiguous"] > 0 with pytest.raises(ValueError, match="Prüfung|unklar|ungültig"): confirm_ingestion(conn, confirm_payload(preview, buffer.getvalue())) assert conn.execute("SELECT COUNT(*) FROM portfolio_ingestion_batches").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM positions_snapshot").fetchone()[0] == 0 def test_postfinance_api_contract_accepts_file_payload_without_exposing_it() -> None: conn = database() preview_request = IngestionPreviewRequest.model_validate(request()) payload = preview_ingestion(conn, preview_request.model_dump()) assert payload["document_type"] == "PostFinance Portfolio-Bewertung" assert "content_base64" not in payload confirmation = IngestionConfirmRequest.model_validate(confirm_payload(payload)) assert confirm_ingestion(conn, confirmation.model_dump())["idempotent"] is False app = create_app(write_mode="test") assert "postfinance_etrading" in app.openapi()["components"]["schemas"]["IngestionPreviewRequest"]["properties"]["source_key"]["enum"] def test_same_name_with_wrong_asset_class_or_currency_is_never_silently_reused() -> None: conn = database() conn.execute( """INSERT INTO instruments( instrument_id,asset_class,name,currency,is_active,instrument_status,valuation_policy,created_at) VALUES('wrong-instrument','crypto','Synthetic Alpine AG','USD',1,'active','live_price',?)""", (NOW,), ) conn.commit() preview = preview_ingestion(conn, request()) assert preview["counts"]["ambiguous"] == 0 confirm_ingestion(conn, confirm_payload(preview)) imported = conn.execute( """SELECT p.instrument_id,i.asset_class,i.currency FROM positions_snapshot p JOIN instruments i ON i.instrument_id=p.instrument_id WHERE i.name='Synthetic Alpine AG' AND p.snapshot_date='2026-05-14'""" ).fetchone() assert imported is not None assert imported["instrument_id"] != "wrong-instrument" assert imported["asset_class"] == "stock" assert imported["currency"] == "CHF" def test_portfolio_read_does_not_refresh_accounts_updated_at() -> None: conn = database() ensure_canonical_cash_accounts(conn) conn.execute("UPDATE accounts SET updated_at='2020-01-01T00:00:00Z'") conn.commit() before = [tuple(row) for row in conn.execute("SELECT account_id,updated_at FROM accounts ORDER BY account_id")] get_cash_summary(conn) after = [tuple(row) for row in conn.execute("SELECT account_id,updated_at FROM accounts ORDER BY account_id")] assert after == before def run_postfinance_source_contract() -> None: """Run the complete PostFinance contract inside the existing ingestion CI test item.""" test_postfinance_docx_preview_is_storage_free_and_describes_baseline_coverage() test_postfinance_confirm_is_atomic_audited_and_file_idempotent() test_postfinance_reconciliation_matches_quantities_and_total_but_performance_has_no_prebaseline_history() test_postfinance_invalid_or_ambiguous_batch_cannot_partially_confirm() test_postfinance_api_contract_accepts_file_payload_without_exposing_it() test_same_name_with_wrong_asset_class_or_currency_is_never_silently_reused() test_portfolio_read_does_not_refresh_accounts_updated_at()