from __future__ import annotations import base64 from dataclasses import replace from decimal import Decimal from hashlib import sha256 import pytest from jarvis_finance.imports.postfinance_documents import ( PFBundle, PFDocument, PFEvent, PFSnapshot, PFSnapshotCash, PFSnapshotPosition, ) from jarvis_finance.services import postfinance_service as service from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations DEPOT = "account_pf_depot_test" CASH = "account_pf_trading_cash_test" def make_conn(): conn = connect_memory() apply_migrations(conn) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,country,default_currency,is_active,created_at) VALUES('pf','PostFinance','bank','CH','CHF',1,'2026-01-01')" ) conn.execute( "INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,performance_included,is_active,created_at,portfolio_bucket) VALUES(?,'pf','PostFinance E-Trading','brokerage','CHF',1,1,'2026-01-01','postfinance')", (DEPOT,), ) conn.execute( "INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,performance_included,is_active,created_at,portfolio_bucket) VALUES(?,'pf','PostFinance Cash-Konto','cash','CHF',1,1,'2026-01-01','cash')", (CASH,), ) for index in range(22): instrument = f"instrument_pf_{index:02d}" label = f"Synthetic Asset {index + 1:02d}" conn.execute( "INSERT INTO instruments(instrument_id,name,ticker,isin,asset_class,currency,is_active,created_at) VALUES(?,?,?,?,?,?,1,?)", ( instrument, label, f"S{index:02d}", f"CH{index:010d}", "equity" if index < 11 else "etf", "CHF", "2026-01-01", ), ) conn.execute( "INSERT INTO positions_snapshot(position_snapshot_id,snapshot_date,account_id,platform_id,instrument_id,quantity,average_cost_original,cost_basis_original,market_price_original,market_value_original,market_fx_rate_to_chf,market_value_chf,category,data_quality_status,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", ( f"baseline_{index}", "2026-06-30", DEPOT, "pf", instrument, "1", "50", "50", "90", "90", "1", "90", "equity" if index < 11 else "etf", "complete", "2026-06-30T00:00:00Z", ), ) conn.commit() return conn def bundle( zip_raw: bytes = b"synthetic-zip", overview_raw: bytes = b"%PDF-synthetic-overview" ) -> PFBundle: zip_hash = sha256(zip_raw).hexdigest() overview_hash = sha256(overview_raw).hexdigest() doc_raw = b"%PDF-synthetic-trade" document = PFDocument( sha256(doc_raw).hexdigest(), sha256(b"trade.pdf").hexdigest(), "trade_confirmation", "synthetic-reference", sha256(b"semantic").hexdigest(), "2026-07-01", sha256(b"account").hexdigest(), "etrading_cash", 1, doc_raw, "synthetic", ) event = PFEvent( sha256(b"event").hexdigest(), document.document_hash, document.semantic_reference, "buy", "etrading_cash", "2026-07-01", "2026-07-03", "out", "Synthetic Asset 01", "CH0000000000", Decimal("1"), Decimal("50"), Decimal("50"), Decimal("1"), Decimal("0"), Decimal("51"), "CHF", Decimal("1"), None, "complete", (), ) transfer_out = PFEvent( sha256(b"transfer-out").hexdigest(), document.document_hash, "transfer-reference", "internal_transfer", "efinance", "2026-07-02", "2026-07-02", "out", None, None, None, None, Decimal("10"), Decimal("0"), Decimal("0"), Decimal("10"), "CHF", Decimal("1"), "synthetic-transfer-group", "partial", ("internal_transfer_not_income_or_expense",), ) transfer_in = PFEvent( sha256(b"transfer-in").hexdigest(), document.document_hash, "transfer-reference", "internal_transfer", "etrading_cash", "2026-07-02", "2026-07-02", "in", None, None, None, None, Decimal("10"), Decimal("0"), Decimal("0"), Decimal("10"), "CHF", Decimal("1"), "synthetic-transfer-group", "partial", ("internal_transfer_not_income_or_expense",), ) positions = tuple( PFSnapshotPosition( sha256(f"row-{index}".encode()).hexdigest()[:32], f"Synthetic Asset {index + 1:02d}", f"synthetic asset {index + 1:02d}", "stock" if index < 11 else "etf", Decimal("1"), Decimal("50"), Decimal("50"), Decimal("100"), "CHF", Decimal("100"), Decimal("4.17"), ) for index in range(22) ) snapshot = PFSnapshot( overview_hash, 5, "2026-07-27T09:14:30+02:00", positions, ( PFSnapshotCash("EUR", Decimal("100"), Decimal("1"), Decimal("100")), PFSnapshotCash("USD", Decimal("100"), Decimal("1"), Decimal("100")), ), Decimal("2200"), Decimal("200"), Decimal("2400"), Decimal("1100"), Decimal("1100"), 0, ) return PFBundle( zip_hash, overview_hash, sha256((zip_hash + overview_hash).encode()).hexdigest(), (document,), (event, transfer_out, transfer_in), snapshot, ("synthetic-transfer-group",), (), ) def request(zip_raw: bytes = b"synthetic-zip", overview_raw: bytes = b"%PDF-synthetic-overview"): return { "zip_file_name": "documents.zip", "overview_file_name": "overview.pdf", "zip_content_base64": base64.b64encode(zip_raw).decode(), "overview_content_base64": base64.b64encode(overview_raw).decode(), } def test_preview_confirm_and_reparse_are_atomic_idempotent_and_keep_roles_separate( monkeypatch, tmp_path ): conn = make_conn() calls = [] monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path)) monkeypatch.setattr( service, "parse_postfinance_bundle", lambda zip_raw, overview_raw: ( calls.append((zip_raw, overview_raw)) or bundle(zip_raw, overview_raw) ), ) payload = request() before = conn.total_changes preview = service.preview_postfinance_import(conn, payload) assert conn.total_changes == before assert preview["mapped_position_count"] == 22 first = service.confirm_postfinance_import( conn, { **payload, "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "confirm": True, }, ) again_preview = service.preview_postfinance_import(conn, payload) second = service.confirm_postfinance_import( conn, { **payload, "preview_id": again_preview["preview_id"], "confirmation_id": again_preview["confirmation_id"], "confirm": True, }, ) assert len(calls) >= 4 assert first["idempotent"] is False and second["idempotent"] is True safe_summary = service.get_postfinance_summary(conn) serialized = str(safe_summary) assert "account_pf_" not in serialized and "CH0000000000" not in serialized assert "snapshot_id" not in safe_summary["latest_snapshot"] assert "instrument_id" not in safe_summary["positions"][0] assert "batch_id" not in safe_summary["imports"][0] assert conn.execute("SELECT COUNT(*) FROM postfinance_import_batches").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM postfinance_snapshots").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM postfinance_snapshot_positions").fetchone()[0] == 22 assert conn.execute("SELECT COUNT(*) FROM postfinance_account_roles").fetchone()[0] == 3 roles = dict(conn.execute("SELECT role,account_id FROM postfinance_account_roles")) assert roles["etrading_depot"] == DEPOT assert roles["etrading_cash"] == CASH assert roles["efinance"] not in {DEPOT, CASH} assert len(set(roles.values())) == 3 assert ( conn.execute( "SELECT account_name FROM accounts WHERE account_id=?", (roles["efinance"],) ).fetchone()[0] == "PostFinance E-Finance" ) transfer_rows = conn.execute( "SELECT account_id,activity_kind FROM transactions WHERE internal_transfer_group_id='synthetic-transfer-group' ORDER BY account_id" ).fetchall() assert {row["account_id"] for row in transfer_rows} == {roles["efinance"], CASH} assert {row["activity_kind"] for row in transfer_rows} == {"internal_transfer"} assert ( conn.execute( "SELECT activity_kind FROM transactions WHERE source_type='postfinance_official_import'" ).fetchone()[0] == "trade" ) assert len(list((tmp_path / "imports" / "postfinance" / "archive").glob("*"))) == 3 def test_provider_and_computed_cost_basis_are_separate(monkeypatch, tmp_path): conn = make_conn() monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path)) monkeypatch.setattr( service, "parse_postfinance_bundle", lambda zip_raw, overview_raw: bundle(zip_raw, overview_raw), ) payload = request() preview = service.preview_postfinance_import(conn, payload) service.confirm_postfinance_import( conn, { **payload, "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "confirm": True, }, ) first = conn.execute( "SELECT provider_cost_total_original,computed_cost_basis_original,computed_cost_basis_status FROM postfinance_snapshot_positions WHERE computed_cost_basis_original IS NOT NULL" ).fetchone() assert first[0] == "50" and first[1] == "51" and first[2] == "complete" assert conn.execute("SELECT COUNT(*) FROM postfinance_cost_basis_lots").fetchone()[0] == 1 def test_sell_consumes_documented_fifo_lot_before_cost_basis_persistence(monkeypatch, tmp_path): conn = make_conn() monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path)) original = bundle() buy = replace( original.events[0], quantity=Decimal("2"), gross_original=Decimal("100"), net_original=Decimal("101"), ) sell = replace( buy, event_fingerprint=sha256(b"sell-event").hexdigest(), event_type="sell", occurred_on="2026-07-10", direction="in", quantity=Decimal("1"), price_original=Decimal("60"), gross_original=Decimal("60"), fee_original=Decimal("1"), net_original=Decimal("59"), ) fifo_bundle = replace(original, events=(buy, sell, *original.events[1:])) monkeypatch.setattr(service, "parse_postfinance_bundle", lambda _zip, _overview: fifo_bundle) payload = request() preview = service.preview_postfinance_import(conn, payload) service.confirm_postfinance_import( conn, { **payload, "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "confirm": True, }, ) lot = conn.execute( "SELECT quantity_acquired,quantity_remaining FROM postfinance_cost_basis_lots" ).fetchone() assert tuple(lot) == ("2", "1") computed = conn.execute( "SELECT computed_cost_basis_original,computed_cost_basis_status FROM postfinance_snapshot_positions WHERE computed_cost_basis_original IS NOT NULL" ).fetchone() assert tuple(computed) == ("50.5", "complete") def test_same_snapshot_with_different_original_source_is_blocked(monkeypatch, tmp_path): conn = make_conn() monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path)) monkeypatch.setattr( service, "parse_postfinance_bundle", lambda zip_raw, overview_raw: bundle(zip_raw, overview_raw), ) payload = request() preview = service.preview_postfinance_import(conn, payload) service.confirm_postfinance_import( conn, { **payload, "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "confirm": True, }, ) alternate = request(b"synthetic-zip-revision", b"%PDF-synthetic-overview-revision") conflict = service.preview_postfinance_import(conn, alternate) assert conflict["conflict"] is True with pytest.raises(ValueError, match="conflict"): service.confirm_postfinance_import( conn, { **alternate, "preview_id": conflict["preview_id"], "confirmation_id": conflict["confirmation_id"], "confirm": True, }, ) assert conn.execute("SELECT COUNT(*) FROM postfinance_import_batches").fetchone()[0] == 1 def test_existing_same_day_canonical_projection_is_blocked_during_preview(monkeypatch): conn = make_conn() conn.execute( """INSERT INTO positions_snapshot(position_snapshot_id,snapshot_date,account_id,platform_id, instrument_id,quantity,market_value_chf,data_quality_status,created_at) VALUES('same-day','2026-07-27',?,'pf','instrument_pf_00','1','100','complete','2026-07-27')""", (DEPOT,), ) conn.commit() monkeypatch.setattr(service, "parse_postfinance_bundle", lambda _zip, _overview: bundle()) preview = service.preview_postfinance_import(conn, request()) assert preview["conflict"] is True assert preview["conflicting_batch_id"] == "canonical_projection_date_conflict" def test_immutable_postfinance_source_and_snapshot_rows(monkeypatch, tmp_path): conn = make_conn() monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path)) monkeypatch.setattr( service, "parse_postfinance_bundle", lambda zip_raw, overview_raw: bundle(zip_raw, overview_raw), ) payload = request() preview = service.preview_postfinance_import(conn, payload) result = service.confirm_postfinance_import( conn, { **payload, "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "confirm": True, }, ) with pytest.raises(Exception, match="immutable"): conn.execute( "UPDATE postfinance_snapshots SET total_chf='0' WHERE snapshot_id=?", (result["snapshot_id"],), ) with pytest.raises(Exception, match="cannot be deleted"): conn.execute( "DELETE FROM postfinance_snapshots WHERE snapshot_id=?", (result["snapshot_id"],) ) immutable_columns = { "postfinance_batch_documents": "document_hash=document_hash", "postfinance_snapshot_positions": "created_at=created_at", "postfinance_snapshot_cash": "created_at=created_at", "postfinance_event_components": "created_at=created_at", } for table, assignment in immutable_columns.items(): with pytest.raises(Exception, match="immutable"): conn.execute(f"UPDATE {table} SET {assignment}") with pytest.raises(Exception, match="cannot be deleted"): conn.execute(f"DELETE FROM {table}") canonical_tables = { "positions_snapshot": "source_type='postfinance_official_import'", "cash_balances": "source_type='postfinance_official_import'", "transactions": "source_type='postfinance_official_import'", } for table, predicate in canonical_tables.items(): with pytest.raises(Exception, match="immutable"): conn.execute(f"UPDATE {table} SET created_at=created_at WHERE {predicate}") with pytest.raises(Exception, match="cannot be deleted"): conn.execute(f"DELETE FROM {table} WHERE {predicate}")