from __future__ import annotations import hashlib import json from decimal import Decimal from sqlite3 import Connection import pytest from fastapi import HTTPException from fastapi.testclient import TestClient import jarvis_finance.services.household_import as household_import_service from jarvis_finance.api.dependencies import get_db from jarvis_finance.api.main import create_app from jarvis_finance.services.budget_categories import confirm_create_category from jarvis_finance.services.budget_overview import get_budget_charts_2026, get_monthly_summary from jarvis_finance.services.budget_transactions import ( confirm_budget_transaction, confirm_transfer, create_budget_transaction_preview, ) from jarvis_finance.services.household_financials import get_household_financial_summary from jarvis_finance.services.household_import import ( configure_source_mapping, confirm_household_import, confirm_household_review_action, get_household_overview, get_household_review, list_household_transactions, preview_household_import, preview_household_review_action, ) from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, get_schema_version def database() -> Connection: conn = connect_memory() apply_migrations(conn) conn.execute("INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES ('p_syn','Synthetic','bank','2026-01-01')") for key, label in (("rai", "Household A"), ("akb", "Household B"), ("visa", "Card A")): account_type = "credit_card" if key == "visa" else "cash" conn.execute( """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,portfolio_bucket,created_at) VALUES (?,?,?,?, 'CHF',?,'2026-01-01')""", (f"acct_{key}", "p_syn", label, account_type, "liability" if key == "visa" else "cash")) conn.execute( """INSERT INTO budget_accounts(budget_account_id,linked_account_id,name,account_type,currency,created_at) VALUES (?,?,?,?, 'CHF','2026-01-01')""", (f"bacc_{key}", f"acct_{key}", label, account_type)) conn.commit() configure_source_mapping(conn, {"confirm": True, "source_type": "raiffeisen_bank", "source_reference": "SYN-RAI-001", "budget_account_id": "bacc_rai"}) configure_source_mapping(conn, {"confirm": True, "source_type": "akb_bank", "source_reference": "SYN-AKB-001", "budget_account_id": "bacc_akb"}) configure_source_mapping(conn, {"confirm": True, "source_type": "visa_credit_card", "source_reference": "SYN-CARD-001", "budget_account_id": "bacc_visa"}) return conn def raiffeisen_csv(amount: str = "-100.00", text: str = "SWISSLOS E-COMMERCE", tx_date: str = "2026-01-26") -> str: return f"IBAN;Booked At;Text;Credit/Debit Amount;Valuta Date\nSYN-RAI-001;{tx_date};{text};{amount};{tx_date}\n" def akb_csv(amount: str = "100.00", text: str = "Own account transfer", tx_date: str = "2026-01-26") -> str: return f"Buchung;Valuta;Buchungstext;Belastung;Gutschrift\n{tx_date};{tx_date};{text};;{amount}\n" def transfer_payload() -> dict: return {"files": [ {"profile": "raiffeisen_bank", "csv_text": raiffeisen_csv()}, {"profile": "akb_bank", "source_reference": "SYN-AKB-001", "csv_text": akb_csv()}, ]} def confirm_payload(source: dict, preview: dict) -> dict: return source | {"confirm": True, "preview_fingerprint": preview["preview_fingerprint"], "baseline_fingerprint": preview["baseline_fingerprint"]} def business_ready_preview(conn: Connection, source: dict) -> tuple[dict, dict]: initial = preview_household_import(conn, source) by_type = { str(category.get("category_type")): str(category["category_id"]) for category in initial["categories"] if category.get("category_type") in {"expense", "income"} } overrides = { str(item["row_token"]): by_type[str(item["transaction_semantics"])] for item in initial["items"] if item["user_state"] != "proposal_ready" and str(item["transaction_semantics"]) in by_type } resolved = source | {"category_overrides": overrides} preview = preview_household_import(conn, resolved) assert preview["business_ready_for_confirm"] is True return resolved, preview def test_schema_47_has_household_contract_and_three_unique_layers() -> None: conn = database() assert get_schema_version(conn) == 47 tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} assert {"household_account_source_mappings", "household_import_batches", "household_import_files", "household_import_items", "household_migros_links"}.issubset(tables) indexes = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='index'")} assert {"ux_budget_candidates_household_source_row", "ux_budget_candidates_household_logical"}.issubset(indexes) def test_preview_is_strictly_read_only_deterministic_and_redacted() -> None: conn = database(); payload = transfer_payload() conn.commit(); before_changes = conn.total_changes before_dump = "\n".join(conn.iterdump()) first = preview_household_import(conn, payload) second = preview_household_import(conn, payload) first_stable = {key: value for key, value in first.items() if key != "performance"} second_stable = {key: value for key, value in second.items() if key != "performance"} assert first_stable == second_stable assert first["preview_fingerprint"] == second["preview_fingerprint"] assert first["performance"]["api_total_seconds"] >= 0 assert len(first["performance"]["parsing_by_file"]) == 2 assert conn.total_changes == before_changes assert "\n".join(conn.iterdump()) == before_dump assert first["confirmable"] is True assert first["counts"]["safe_transfer_pairs"] == 1 serialized = json.dumps(first) assert "SYN-RAI-001" not in serialized and "SYN-AKB-001" not in serialized assert "source_reference" not in serialized def test_preview_bulk_loads_repeated_source_mapping_once_per_binding() -> None: conn = database() rows = "".join( f"SYN-RAI-001;2026-01-{(index % 28) + 1:02d};Merchant {index};-{index + 1}.00;2026-01-{(index % 28) + 1:02d}\n" for index in range(40) ) payload = { "files": [{ "profile": "raiffeisen_bank", "csv_text": "IBAN;Booked At;Text;Credit/Debit Amount;Valuta Date\n" + rows, }] } statements: list[str] = [] conn.set_trace_callback(statements.append) preview_household_import(conn, payload) conn.set_trace_callback(None) mapping_selects = [ statement for statement in statements if statement.lstrip().upper().startswith("SELECT M.MAPPING_ID") and "household_account_source_mappings" in statement ] assert len(mapping_selects) == 1 def test_golden_raiffeisen_2026_01_26_swisslos_chf100_is_safe_v3_transfer_and_confirm_is_noop_twice() -> None: conn = database(); payload = transfer_payload() preview = preview_household_import(conn, payload) pair = preview["transfer_pairs"][0] assert pair["pairing_class"] == "safe" assert pair["amount"] == "100.00" and pair["currency"] == "CHF" assert preview["pairing_version"] == "transfer_pairing_v3" first = confirm_household_import(conn, confirm_payload(payload, preview)) second = confirm_household_import(conn, confirm_payload(payload, preview)) assert first["transfer_pair_count"] if "transfer_pair_count" in first else first["counts"]["transfer_pair_count"] == 1 assert second["idempotent"] is True and second["batch_id"] == first["batch_id"] assert conn.execute("SELECT COUNT(*) FROM budget_transfer_pairs WHERE status='confirmed' AND quality_status='safe'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM budget_transfers").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM budget_transactions WHERE transaction_type='transfer'").fetchone()[0] == 2 audit = conn.execute("SELECT new_values_json FROM audit_log WHERE action='household_import_confirmed'").fetchone()[0] assert "SYN-" not in audit and "SWISSLOS" not in audit and "csv_text" not in audit def test_generic_payment_text_cannot_turn_unrelated_rows_into_a_safe_transfer() -> None: conn = database() payload = {"files": [ {"profile": "raiffeisen_bank", "csv_text": raiffeisen_csv("-10.00", "Payment Coffee")}, {"profile": "akb_bank", "source_reference": "SYN-AKB-001", "csv_text": akb_csv("10.00", "Salary correction")}, ]} preview = preview_household_import(conn, payload) assert preview["counts"]["safe_transfer_pairs"] == 0 assert all(row["disposition"] != "transfer_confirmed" for row in preview["rows"]) def test_stale_confirm_rejected_before_any_batch_write() -> None: conn = database(); payload = transfer_payload(); preview = preview_household_import(conn, payload) conn.execute("UPDATE household_account_source_mappings SET updated_at='2026-02-01' WHERE source_type='akb_bank'") conn.commit() with pytest.raises(HTTPException) as exc: confirm_household_import(conn, confirm_payload(payload, preview)) assert exc.value.status_code == 409 assert conn.execute("SELECT COUNT(*) FROM household_import_batches").fetchone()[0] == 0 def test_file_source_row_and_logical_duplicates_are_distinguished() -> None: conn = database() visa = "TransactionId,CardId,Date,Amount,Currency,MerchantName\nT-001,SYN-CARD-001,2026-02-01,-12.50,CHF,Synthetic Shop\n" payload = {"files": [{"profile": "visa_credit_card", "csv_text": visa}]} payload, preview = business_ready_preview(conn, payload) confirm_household_import(conn, confirm_payload(payload, preview)) same_file = preview_household_import(conn, payload) assert same_file["rows"][0]["disposition"] == "duplicate_file" overlap = {"files": [{"profile": "visa_credit_card", "csv_text": visa.replace("T-001", "T-002")}]} logical = preview_household_import(conn, overlap) assert logical["rows"][0]["disposition"] == "review" assert logical["rows"][0]["classification"] == "possible_logical_duplicate" source_overlap = {"files": [{"profile": "visa_credit_card", "csv_text": visa + "\n"}]} assert preview_household_import(conn, source_overlap)["rows"][0]["disposition"] == "duplicate_source_row" def test_credit_card_pending_final_refund_and_payment_are_explicit() -> None: conn = database() csv_text = "TransactionId,CardId,Date,Amount,Currency,MerchantName,Status\n" \ "T-P,SYN-CARD-001,2026-02-02,-20.00,CHF,Synthetic Market,pending\n" \ "T-P,SYN-CARD-001,2026-02-02,-20.00,CHF,Synthetic Market,booked\n" \ "T-R,SYN-CARD-001,2026-02-03,5.00,CHF,Synthetic Refund,booked\n" preview = preview_household_import(conn, {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]}) dispositions = [row["disposition"] for row in preview["rows"]] classes = [row["classification"] for row in preview["rows"]] assert "superseded_pending" in dispositions and "candidate" in dispositions assert "credit_card_refund" in classes bank_payment = preview_household_import(conn, {"files": [{"profile": "raiffeisen_bank", "csv_text": raiffeisen_csv("-55.00", "VISA CARD PAYMENT")}]}) assert bank_payment["rows"][0]["classification"] == "credit_card_payment" assert bank_payment["transfer_pairs"][0]["pairing_class"] == "unmatched" def test_migros_links_detail_without_double_counting_and_difference_over_cent_reviews() -> None: conn = database() visa = "TransactionId,CardId,Date,Amount,Currency,MerchantName\nM-1,SYN-CARD-001,2026-03-01,-60.00,CHF,Migros Synthetic\n" receipt = "Datum;Zeit;Filiale;Kassennummer;Transaktionsnummer;Artikel;Umsatz\n" \ "2026-03-01;12:00;Synthetic Store;1;99;Item A;24.00\n" \ "2026-03-01;12:00;Synthetic Store;1;99;Item B;36.00\n" payload = {"files": [{"profile": "visa_credit_card", "csv_text": visa}, {"profile": "migros_receipts", "csv_text": receipt}]} preview = preview_household_import(conn, payload) assert preview["receipt_links"][0]["status"] == "linked" result = confirm_household_import(conn, confirm_payload(payload, preview)) assert result["counts"]["receipt_link_count"] == 1 assert conn.execute("SELECT COUNT(*) FROM budget_import_line_items").fetchone()[0] == 2 # The card entry is the only money movement; receipt detail never creates another. assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM household_migros_links WHERE status='linked'").fetchone()[0] == 1 conn2 = database() mismatch = payload | {"files": [payload["files"][0], {"profile": "migros_receipts", "csv_text": receipt.replace("36.00", "36.02")}]} review = preview_household_import(conn2, mismatch) assert review["receipt_links"][0]["status"] == "review" assert review["receipt_links"][0]["difference"] == "0.02" def test_api_contract_gets_are_side_effect_free_and_write_gate_is_preserved() -> None: conn = database() def override_db(): yield conn app = create_app(write_mode="test"); app.dependency_overrides[get_db] = override_db client = TestClient(app) paths = client.get("/openapi.json").json()["paths"] for path in ("/api/budget/household/source-mappings", "/api/budget/household/import/preview", "/api/budget/household/import/confirm", "/api/budget/household/import/batches"): assert path in paths before = conn.total_changes assert client.get("/api/budget/household/source-mappings").status_code == 200 assert client.get("/api/budget/household/import/batches").status_code == 200 assert conn.total_changes == before blocked_app = create_app(write_mode="disabled"); blocked_app.dependency_overrides[get_db] = override_db blocked = TestClient(blocked_app) assert blocked.post("/api/budget/household/import/confirm", json={}).status_code == 403 def test_idempotent_confirm_recomputes_input_identity_before_noop() -> None: conn = database(); payload = transfer_payload(); preview = preview_household_import(conn, payload) confirm_household_import(conn, confirm_payload(payload, preview)) stolen = transfer_payload() stolen["files"][0]["csv_text"] = raiffeisen_csv("-101.00") with pytest.raises(HTTPException) as exc: confirm_household_import(conn, confirm_payload(stolen, preview)) assert exc.value.status_code == 409 assert "SYN-" not in str(exc.value.detail) def test_selected_mapping_is_bound_and_mismatch_fails_closed() -> None: conn = database() payload = {"files": [{"profile": "raiffeisen_bank", "csv_text": raiffeisen_csv("-10.00")}]} mapping_ids = { row["source_type"]: row["mapping_id"] for row in conn.execute("SELECT mapping_id,source_type FROM household_account_source_mappings") } payload["files"][0]["mapping_id"] = mapping_ids["visa_credit_card"] mismatch = preview_household_import(conn, payload) assert mismatch["confirmable"] is False assert mismatch["errors"][0]["code"] == "account_source_mapping_missing_or_mismatch" payload["files"][0]["mapping_id"] = mapping_ids["raiffeisen_bank"] payload, matched = business_ready_preview(conn, payload) assert matched["confirmable"] is True confirm_household_import(conn, confirm_payload(payload, matched)) payload["files"][0]["mapping_id"] = mapping_ids["visa_credit_card"] with pytest.raises(HTTPException) as exc: confirm_household_import(conn, confirm_payload(payload, matched)) assert exc.value.status_code == 409 def test_baseline_covers_accounts_transactions_transfers_and_household_relations() -> None: mutations = ( "UPDATE accounts SET account_name='Changed' WHERE account_id='acct_rai'", "UPDATE budget_accounts SET updated_at='changed' WHERE budget_account_id='bacc_rai'", "INSERT INTO budget_transactions(budget_transaction_id,account_id,transaction_type,transaction_date,description,amount_original,currency_original,amount_chf,fx_status,status,source_type,created_at,updated_at) VALUES ('drift','bacc_rai','expense','2026-01-01','x','1','CHF','1','not_needed','confirmed','manual','x','x')", ) for sql in mutations: conn = database(); payload = transfer_payload(); preview = preview_household_import(conn, payload) conn.execute(sql); conn.commit() with pytest.raises(HTTPException) as exc: confirm_household_import(conn, confirm_payload(payload, preview)) assert exc.value.status_code == 409 def test_limits_empty_unsupported_and_errors_fail_closed_without_raw_leaks() -> None: conn = database() bad_payloads = ( {"files": []}, {"files": [{"profile": "visa_credit_card", "csv_text": ""}]}, {"files": [{"profile": "unknown", "csv_text": "secret-account-reference"}]}, {"files": [{"profile": "auto", "csv_text": "secret-account-reference"}]}, {"files": [{"profile": "visa_credit_card", "csv_text": "x" * (5 * 1024 * 1024 + 1)}]}, ) for payload in bad_payloads: with pytest.raises(HTTPException) as exc: preview_household_import(conn, payload) assert exc.value.status_code in {413, 422} assert "secret-account-reference" not in str(exc.value.detail) def test_public_preview_exposes_protocol_bindings_but_no_internal_file_or_row_hashes() -> None: conn = database() payload = transfer_payload() preview = preview_household_import(conn, payload) encoded = str(preview) assert preview["preview_fingerprint"] and preview["baseline_fingerprint"] assert "input_fingerprint" not in preview assert "file_fingerprint" not in encoded assert "source_row_fingerprint" not in encoded assert "logical_fingerprint" not in encoded assert "money_candidate_id" not in encoded assert "SYN-" not in encoded assert all(str(row["row_token"]).startswith("row_") for row in preview["rows"]) def test_same_request_duplicate_and_pending_final_have_one_writable_final() -> None: conn = database() rows = "TransactionId,CardId,Date,Amount,Currency,MerchantName,Status\n" \ "P-1,SYN-CARD-001,2026-02-02,-20.00,CHF,Synthetic,pending\n" \ "P-1,SYN-CARD-001,2026-02-02,-20.00,CHF,Synthetic,booked\n" \ "P-1,SYN-CARD-001,2026-02-02,-20.00,CHF,Synthetic,booked\n" payload = {"files": [{"profile": "visa_credit_card", "csv_text": rows}]} payload, preview = business_ready_preview(conn, payload) assert preview["counts"]["duplicates"] == 2 assert [row["disposition"] for row in preview["rows"]].count("candidate") == 1 result = confirm_household_import(conn, confirm_payload(payload, preview)) assert result["counts"]["candidate_count"] == 1 assert conn.execute("SELECT COUNT(*) FROM budget_transaction_candidates WHERE household_batch_id IS NOT NULL").fetchone()[0] == 1 conn2 = database() duplicate_rows = "TransactionId,CardId,Date,Amount,Currency,MerchantName\n" \ "D-1,SYN-CARD-001,2026-02-03,-5.00,CHF,Synthetic\n" \ "D-1,SYN-CARD-001,2026-02-03,-5.00,CHF,Synthetic\n" duplicate_payload = {"files": [{"profile": "visa_credit_card", "csv_text": duplicate_rows}]} duplicate_payload, duplicate_preview = business_ready_preview(conn2, duplicate_payload) assert [row["disposition"] for row in duplicate_preview["rows"]] == ["candidate", "duplicate_source_row"] confirm_household_import(conn2, confirm_payload(duplicate_payload, duplicate_preview)) assert conn2.execute("SELECT COUNT(*) FROM budget_transaction_candidates WHERE household_batch_id IS NOT NULL").fetchone()[0] == 1 def test_distinct_provider_transactions_with_same_business_fields_are_not_silently_suppressed() -> None: conn = database() csv_text = "TransactionId,CardId,Date,Amount,Currency,MerchantName\n" \ "REAL-1,SYN-CARD-001,2026-02-03,-5.00,CHF,Synthetic Coffee\n" \ "REAL-2,SYN-CARD-001,2026-02-03,-5.00,CHF,Synthetic Coffee\n" payload = {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]} payload, preview = business_ready_preview(conn, payload) dispositions = [row["disposition"] for row in preview["rows"]] assert dispositions == ["candidate", "review"] assert preview["rows"][1]["classification"] == "possible_logical_duplicate" confirm_household_import(conn, confirm_payload(payload, preview)) assert conn.execute( "SELECT COUNT(*) FROM budget_transaction_candidates WHERE household_batch_id IS NOT NULL" ).fetchone()[0] == 2 def test_legacy_productive_overlap_is_reviewed_not_counted_again() -> None: conn = database() account_id = conn.execute( "SELECT budget_account_id FROM household_account_source_mappings WHERE source_type='raiffeisen_bank'" ).fetchone()[0] conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,booking_date, description,amount_original,currency_original,fx_rate_to_chf,amount_chf,fx_status, status,source_type,created_at,updated_at) VALUES ('legacy-overlap',?,'expense','2026-01-26','2026-01-26', 'Synthetic Grocery','-25.50','CHF','1','-25.50','not_needed', 'confirmed','manual','2026-01-15T00:00:00Z','2026-01-15T00:00:00Z')""", (account_id,), ) conn.commit() payload = {"files": [{"profile": "raiffeisen_bank", "csv_text": raiffeisen_csv("-25.50", "Synthetic Grocery")}]} payload, preview = business_ready_preview(conn, payload) assert preview["rows"][0]["classification"] == "possible_legacy_duplicate" assert preview["rows"][0]["disposition"] == "review" confirm_household_import(conn, confirm_payload(payload, preview)) assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 1 group = next(g for g in get_household_review(conn)["groups"] if g["title"] == "Possible Legacy Duplicate") assert "duplicate" in group["supported_actions"] action = {"group_key": group["group_key"], "action": "duplicate"} action_preview = preview_household_review_action(conn, action) confirm_household_review_action(conn, {**action, **action_preview, "confirm": True}) assert conn.execute( "SELECT COUNT(*) FROM budget_transaction_candidates WHERE status='duplicate'" ).fetchone()[0] == 1 def test_household_confirm_preserves_outer_transaction_and_rolls_back_only_its_savepoint() -> None: conn = database(); payload = transfer_payload(); preview = preview_household_import(conn, payload) conn.execute("INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES ('caller','Caller','bank','x')") result = confirm_household_import(conn, confirm_payload(payload, preview)) assert result["idempotent"] is False and conn.in_transaction conn.rollback() assert conn.execute("SELECT COUNT(*) FROM household_import_batches").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM platforms WHERE platform_id='caller'").fetchone()[0] == 0 conn = database(); preview = preview_household_import(conn, payload) conn.execute("INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES ('caller','Caller','bank','x')") conn.execute("CREATE TEMP TRIGGER fail_household_candidate BEFORE INSERT ON budget_transaction_candidates BEGIN SELECT RAISE(ABORT,'synthetic failure'); END") with pytest.raises(Exception): confirm_household_import(conn, confirm_payload(payload, preview)) assert conn.in_transaction assert conn.execute("SELECT COUNT(*) FROM platforms WHERE platform_id='caller'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM household_import_batches").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE action='household_import_confirmed'").fetchone()[0] == 0 def test_transfer_false_positive_card_refund_is_not_safe() -> None: conn = database() bank = raiffeisen_csv("-10.00", "Ordinary purchase") card = "TransactionId,CardId,Date,Amount,Currency,MerchantName\nR-1,SYN-CARD-001,2026-01-26,10.00,CHF,Ordinary refund\n" preview = preview_household_import(conn, {"files": [ {"profile": "raiffeisen_bank", "csv_text": bank}, {"profile": "visa_credit_card", "csv_text": card}, ]}) assert preview["counts"]["safe_transfer_pairs"] == 0 def test_clear_chf_card_refund_becomes_productive_refund_only_after_confirm() -> None: conn = database() csv_text = "TransactionId,CardId,Date,Amount,Currency,MerchantName\nRF-1,SYN-CARD-001,2026-02-10,12.50,CHF,Synthetic refund\n" payload = {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]} preview = preview_household_import(conn, payload) assert preview["rows"][0]["classification"] == "credit_card_refund" assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 0 confirm_household_import(conn, confirm_payload(payload, preview)) tx = conn.execute( "SELECT transaction_type,amount_original,status,source_type FROM budget_transactions" ).fetchone() assert tuple(tx) == ("refund", "12.50", "confirmed", "import_candidate") overview = get_household_overview(conn, month="2026-02") assert overview["income_chf"] == "0.00" and overview["expense_chf"] == "0.00" assert overview["unlinked_refund_count"] == 1 conn2 = database() eur_csv = csv_text.replace(",CHF,", ",EUR,").replace("RF-1", "RF-2") eur_payload = {"files": [{"profile": "visa_credit_card", "csv_text": eur_csv}]} eur_preview = preview_household_import(conn2, eur_payload) assert eur_preview["rows"][0]["requires_review"] is True confirm_household_import(conn2, confirm_payload(eur_payload, eur_preview)) assert conn2.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 0 def test_unique_reversal_deterministically_offsets_original_transaction() -> None: conn = database() purchase = { "files": [{ "profile": "visa_credit_card", "csv_text": "Date,Amount,Currency,MerchantName,TransactionId,CardId\n2026-02-01,-8.25,CHF,Synthetic reverse purchase,rev-purchase,SYN-CARD-001\n", }] } purchase, purchase_preview = business_ready_preview(conn, purchase) confirm_household_import(conn, confirm_payload(purchase, purchase_preview)) original_id = conn.execute( "SELECT budget_transaction_id FROM budget_transactions WHERE description='Synthetic reverse purchase'" ).fetchone()[0] reversal = { "files": [{ "profile": "visa_credit_card", "csv_text": "Date,Amount,Currency,MerchantName,TransactionId,CardId\n2026-02-03,8.25,CHF,Storno Synthetic reverse purchase,rev-refund,SYN-CARD-001\n", }] } reversal_preview = preview_household_import(conn, reversal) row = reversal_preview["rows"][0] assert row["classification"] == "credit_card_reversal" assert row["requires_review"] is False assert "reversal_of_transaction_id" not in row confirm_household_import(conn, confirm_payload(reversal, reversal_preview)) result = conn.execute( """SELECT transaction_type,amount_original,reversal_of_transaction_id,notes FROM budget_transactions WHERE description LIKE 'Storno %'""" ).fetchone() assert result["transaction_type"] == "refund" assert Decimal(result["amount_original"]) == Decimal("8.25") assert result["reversal_of_transaction_id"] == original_id assert json.loads(result["notes"])["reversal_of_transaction_id"] == original_id total = conn.execute( "SELECT SUM(CAST(amount_chf AS REAL)) FROM budget_transactions WHERE description LIKE '%reverse purchase'" ).fetchone()[0] assert total == 0 financials = get_household_financial_summary(conn, period="2026-02") assert financials["expense_chf"] == "0.00" assert financials["net_chf"] == "0.00" assert financials["unlinked_refund_count"] == 0 same_file = preview_household_import(conn, reversal) assert same_file["rows"][0]["disposition"] == "duplicate_file" same_business_new_provider_id = { "files": [{ "profile": "visa_credit_card", "csv_text": reversal["files"][0]["csv_text"].replace("rev-refund", "rev-refund-new-id"), }] } logical_overlap = preview_household_import(conn, same_business_new_provider_id) assert logical_overlap["rows"][0]["classification"] == "possible_logical_duplicate" assert logical_overlap["rows"][0]["disposition"] == "review" def test_ambiguous_reversal_remains_review_and_never_posts_automatically() -> None: conn = database() purchases = { "files": [{ "profile": "visa_credit_card", "csv_text": ( "Date,Amount,Currency,MerchantName,TransactionId,CardId\n" "2026-02-01,-8.25,CHF,Repeated purchase,amb-1,SYN-CARD-001\n" "2026-02-02,-8.25,CHF,Repeated purchase,amb-2,SYN-CARD-001\n" ), }] } purchases, purchase_preview = business_ready_preview(conn, purchases) confirm_household_import(conn, confirm_payload(purchases, purchase_preview)) reversal = { "files": [{ "profile": "visa_credit_card", "csv_text": "Date,Amount,Currency,MerchantName,TransactionId,CardId\n2026-02-03,8.25,CHF,Storno Repeated purchase,amb-r,SYN-CARD-001\n", }] } reversal_preview = preview_household_import(conn, reversal) assert reversal_preview["rows"][0]["classification"] == "credit_card_reversal_unmatched" assert reversal_preview["rows"][0]["requires_review"] is True assert reversal_preview["business_ready_for_confirm"] is True confirm_household_import(conn, confirm_payload(reversal, reversal_preview)) assert conn.execute("SELECT COUNT(*) FROM budget_transactions WHERE description LIKE 'Storno %'").fetchone()[0] == 0 def test_review_expense_preview_confirm_is_bound_atomic_and_idempotent() -> None: conn = database() header = "TransactionId,CardId,Date,Amount,Currency,MerchantName\n" known = "".join( f"SAFE-{index},SYN-CARD-001,2026-{'01' if index < 31 else '02'}-{index + 1 if index < 31 else index - 30:02d},-3.00,CHF,Migros\n" for index in range(33) ) csv_text = header + known + ( "RV-1,SYN-CARD-001,2026-02-28,-8.25,CHF,Synthetic review expense\n" ) payload = {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]} import_preview = preview_household_import(conn, payload) confirm_household_import(conn, confirm_payload(payload, import_preview)) review = get_household_review(conn) group = next(item for item in review["groups"] if item["title"] == "Expense Candidate") assert "expense" in group["supported_actions"] and group["can_confirm"] is True before = conn.total_changes action_preview = preview_household_review_action( conn, {"group_key": group["group_key"], "action": "expense"} ) assert conn.total_changes == before request = { "group_key": group["group_key"], "action": "expense", "preview_fingerprint": action_preview["preview_fingerprint"], "baseline_fingerprint": action_preview["baseline_fingerprint"], } result = confirm_household_review_action(conn, request) assert result["confirmed_count"] == 1 and result["idempotent"] is False tx = conn.execute( "SELECT transaction_type,amount_original,status FROM budget_transactions " "WHERE description='Synthetic review expense'" ).fetchone() assert tuple(tx) == ("expense", "-8.25", "confirmed") filtered = list_household_transactions( conn, period="2026-02", account="Card A", source="Import", merchant="Synthetic review expense", review="reviewed" ) assert filtered["total"] == 1 assert filtered["items"][0]["review_status"] == "reviewed" assert filtered["items"][0]["source_label"] == "Import" assert list_household_transactions(conn, review="ignored")["total"] == 0 assert confirm_household_review_action(conn, request)["idempotent"] is True def test_migros_receipt_links_to_existing_money_movement_across_batches() -> None: conn = database() visa = "TransactionId,CardId,Date,Amount,Currency,MerchantName\nMX-1,SYN-CARD-001,2026-03-01,-10.00,CHF,Migros Synthetic\n" money_payload = {"files": [{"profile": "visa_credit_card", "csv_text": visa}]} money_preview = preview_household_import(conn, money_payload) confirm_household_import(conn, confirm_payload(money_payload, money_preview)) transaction_id = conn.execute("SELECT budget_transaction_id FROM budget_transactions").fetchone()[0] receipt = "Datum;Zeit;Filiale;Kassennummer;Transaktionsnummer;Artikel;Umsatz\n" \ "2026-03-01;12:00;Synthetic Store;1;100;A;4.00\n" \ "2026-03-01;12:00;Synthetic Store;1;100;B;6.00\n" receipt_payload = {"files": [{"profile": "migros_receipts", "csv_text": receipt}]} receipt_preview = preview_household_import(conn, receipt_payload) assert receipt_preview["receipt_links"][0]["status"] == "linked" confirm_household_import(conn, confirm_payload(receipt_payload, receipt_preview)) link = conn.execute("SELECT money_candidate_id,money_transaction_id,status FROM household_migros_links").fetchone() assert tuple(link) == (None, transaction_id, "linked") assert conn.execute("SELECT COUNT(*) FROM budget_import_line_items").fetchone()[0] == 2 assert conn.execute("SELECT COUNT(*) FROM budget_transactions WHERE transaction_type='expense'").fetchone()[0] == 1 second_receipt = receipt.replace(";100;", ";102;") second_payload = {"files": [{"profile": "migros_receipts", "csv_text": second_receipt}]} second_preview = preview_household_import(conn, second_payload) assert second_preview["receipt_links"][0]["status"] == "unmatched" confirm_household_import(conn, confirm_payload(second_payload, second_preview)) assert conn.execute( "SELECT COUNT(*) FROM household_migros_links WHERE status='linked'" ).fetchone()[0] == 1 second_link = conn.execute( "SELECT money_candidate_id,money_transaction_id,status FROM household_migros_links " "WHERE status='unmatched'" ).fetchone() assert second_link is None assert conn.execute("SELECT COUNT(*) FROM household_migros_links").fetchone()[0] == 1 def test_migros_receipt_links_to_productive_transaction_not_consumed_candidate() -> None: conn = database() visa = "TransactionId,CardId,Date,Amount,Currency,MerchantName\nMX-2,SYN-CARD-001,2026-03-02,-10.00,CHF,Migros Productive\n" money_payload = {"files": [{"profile": "visa_credit_card", "csv_text": visa}]} money_preview = preview_household_import(conn, money_payload) confirm_household_import(conn, confirm_payload(money_payload, money_preview)) transaction_id = conn.execute("SELECT budget_transaction_id FROM budget_transactions").fetchone()[0] receipt = ( "Datum;Zeit;Filiale;Kassennummer;Transaktionsnummer;Artikel;Umsatz\n" "2026-03-02;12:00;Synthetic Store;1;101;A;4.00\n" "2026-03-02;12:00;Synthetic Store;1;101;B;6.00\n" ) receipt_payload = {"files": [{"profile": "migros_receipts", "csv_text": receipt}]} receipt_preview = preview_household_import(conn, receipt_payload) assert receipt_preview["receipt_links"][0]["status"] == "linked" confirm_household_import(conn, confirm_payload(receipt_payload, receipt_preview)) link = conn.execute("SELECT money_candidate_id,money_transaction_id,status FROM household_migros_links").fetchone() assert tuple(link) == (None, transaction_id, "linked") second_payload = { "files": [{"profile": "migros_receipts", "csv_text": receipt.replace(";101;", ";103;")}] } second_preview = preview_household_import(conn, second_payload) assert second_preview["receipt_links"][0]["status"] == "unmatched" confirm_household_import(conn, confirm_payload(second_payload, second_preview)) assert conn.execute( "SELECT COUNT(*) FROM household_migros_links WHERE status='linked'" ).fetchone()[0] == 1 def test_plural_api_and_read_models_are_side_effect_free() -> None: conn = database() def override_db(): yield conn app = create_app(write_mode="test"); app.dependency_overrides[get_db] = override_db client = TestClient(app) paths = client.get("/openapi.json").json()["paths"] required = ("/api/budget/household/imports/preview", "/api/budget/household/imports/confirm", "/api/budget/household/imports/options", "/api/budget/household/imports/history", "/api/budget/household/imports", "/api/budget/household/overview", "/api/budget/household/transactions", "/api/budget/household/review", "/api/budget/household/review/preview", "/api/budget/household/review/confirm") assert set(required).issubset(paths) before = conn.total_changes for path in required[2:8]: response = client.get(path) assert response.status_code == 200, response.text assert "SYN-" not in response.text and "csv_text" not in response.text assert conn.total_changes == before overview = client.get("/api/budget/household/overview?month=2026-01").json() assert {"month", "income_chf", "expense_chf", "balance_chf", "review_count", "data_status"}.issubset(overview) transactions = client.get("/api/budget/household/transactions?period=2026-01").json() assert {"items", "total", "total_all", "filtered_total", "page_count", "has_more", "next_cursor"}.issubset(transactions) review = client.get("/api/budget/household/review").json() assert review["unsafe_groups_can_confirm"] is False and "groups" in review for path in required[-2:]: assert client.post(path, json={}).status_code in {404, 409, 422} def test_authenticated_owner_attestation_api_binds_exact_cluster_and_confirms_neutral_transfer( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("JARVIS_FINANCE_OPERATOR_APPROVAL_KEY", "synthetic-operator-key") conn = database() def override_db(): yield conn app = create_app(write_mode="test") app.dependency_overrides[get_db] = override_db client = TestClient(app) csv_text = ( "IBAN;Booked At;Text;Credit/Debit Amount;Valuta Date\n" "SYN-RAI-001;2026-07-01;Synthetic Household Counterparty;-20.00;2026-07-01\n" "SYN-RAI-001;2026-07-02;Synthetic Household Counterparty;-30.00;2026-07-02\n" ) payload = {"files": [{"profile": "raiffeisen_bank", "csv_text": csv_text}]} base = client.post("/api/budget/household/imports/preview", json=payload) assert base.status_code == 200, base.text cluster = base.json()["merchant_clusters"][0] approval_body = { "confirm_owner_attestation": True, "preview_request": payload, "approval_evidence": "owner_attested_known_household_counterparty_v1", "cluster_token": cluster["cluster_token"], "approved_row_tokens": cluster["row_tokens"], } denied = client.post( "/api/budget/household/imports/owner-neutral-cluster-approval", headers={"X-Jarvis-Operator-Approval": "wrong"}, json=approval_body, ) assert denied.status_code == 403 subset = client.post( "/api/budget/household/imports/owner-neutral-cluster-approval", headers={"X-Jarvis-Operator-Approval": "synthetic-operator-key"}, json=approval_body | {"approved_row_tokens": cluster["row_tokens"][:1]}, ) assert subset.status_code == 422 approved = client.post( "/api/budget/household/imports/owner-neutral-cluster-approval", headers={"X-Jarvis-Operator-Approval": "synthetic-operator-key"}, json=approval_body, ) assert approved.status_code == 200, approved.text forged_decision = approved.json() public_material = { "scope": "owner_confirmed_neutral_household_transfer_v2", "approval_evidence": forged_decision["approval_evidence"], "cluster_token": forged_decision["cluster_token"], "row_tokens": sorted(forged_decision["approved_row_tokens"]), "baseline_fingerprint": forged_decision["approval_baseline_fingerprint"], "issued_at": forged_decision["approval_issued_at"], "expires_at": forged_decision["approval_expires_at"], "nonce": forged_decision["approval_nonce"], } forged_decision["approval_token"] = hashlib.sha256( json.dumps(public_material, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode() ).hexdigest() forged_preview = client.post( "/api/budget/household/imports/preview", json={**payload, "cluster_decisions": [forged_decision]}, ) assert forged_preview.status_code == 422 signed_payload = {**payload, "cluster_decisions": [approved.json()]} preview_response = client.post("/api/budget/household/imports/preview", json=signed_payload) assert preview_response.status_code == 200, preview_response.text preview = preview_response.json() assert preview["business_ready_for_confirm"] is True assert {item["transaction_semantics"] for item in preview["items"]} == { "user_confirmed_unmatched_transfer" } confirm_request = confirm_payload(signed_payload, preview) confirmed = client.post( "/api/budget/household/imports/confirm", json=confirm_request, ) assert confirmed.status_code == 200, confirmed.text immediate_replay = client.post( "/api/budget/household/imports/confirm", json=confirm_request, ) assert immediate_replay.status_code == 200 assert immediate_replay.json()["idempotent"] is True monkeypatch.setattr( household_import_service.time, "time", lambda: int(approved.json()["approval_expires_at"]) + 1, ) expired_replay = client.post( "/api/budget/household/imports/confirm", json=confirm_request, ) assert expired_replay.status_code == 409 overview = client.get("/api/budget/household/overview?month=2026-07").json() assert overview["income_chf"] == "0.00" assert overview["expense_chf"] == "0.00" audit = conn.execute( "SELECT new_values_json FROM audit_log WHERE action='household_import_confirmed' ORDER BY rowid DESC LIMIT 1" ).fetchone() audit_values = json.loads(audit["new_values_json"]) assert audit_values["owner_attested_neutral_cluster_count"] == 1 assert audit_values["owner_attestation_evidence_version"] == ( "owner_attested_known_household_counterparty_v1" ) def test_review_summary_is_canonical_filtered_and_cursor_paginated() -> None: conn = database() header = "TransactionId,CardId,Date,Amount,Currency,MerchantName\n" known = "".join( f"SAFE-R-{index},SYN-CARD-001,2026-03-{index + 1:02d},-3.00,CHF,Migros\n" for index in range(24) ) payload = {"files": [{"profile": "visa_credit_card", "csv_text": header + known + "OPEN-1,SYN-CARD-001,2026-03-25,-8.25,CHF,Synthetic review expense\n"}]} preview = preview_household_import(conn, payload) confirm_household_import(conn, confirm_payload(payload, preview)) original = dict(conn.execute( "SELECT * FROM budget_transaction_candidates WHERE description='Synthetic review expense'" ).fetchone()) expense_category = conn.execute( "SELECT category_id,name FROM budget_categories WHERE is_active=1 AND category_type='expense' LIMIT 1" ).fetchone() income_category = conn.execute( "SELECT category_id,name FROM budget_categories WHERE is_active=1 AND category_type='income' LIMIT 1" ).fetchone() def clone( candidate_id: str, *, status: str, classification: str, proposed=None, source_type="visa_credit_card", signed_amount: str | None = None, ) -> None: row = original | { "transaction_candidate_id": candidate_id, "source_type": source_type, "status": status, "classification": classification, "description": candidate_id, "merchant": candidate_id, "source_row_fingerprint": f"row-{candidate_id}", "logical_fingerprint": f"logical-{candidate_id}", "proposed_category_id": proposed[0] if proposed else None, "proposed_category_name": proposed[1] if proposed else None, "amount_original": signed_amount or original["amount_original"], "signed_amount_original": signed_amount or original["signed_amount_original"], "updated_at": f"2026-03-25T00:00:0{len(candidate_id)}Z", } keys = list(row) conn.execute( f"INSERT INTO budget_transaction_candidates({','.join(keys)}) VALUES ({','.join('?' for _ in keys)})", [row[key] for key in keys], ) clone( "proposal", status="needs_review", classification="income_candidate", proposed=income_category, signed_amount="8.25", ) clone( "wrong-proposal", status="needs_review", classification="expense", proposed=income_category, source_type="raiffeisen_bank", ) clone("inactive-proposal", status="needs_review", classification="expense", proposed=expense_category) clone("special", status="needs_review", classification="possible_logical_duplicate", proposed=expense_category) clone("covered", status="covered_by_source", classification="migros_receipt_detail", source_type="migros_receipts") # Lifecycle status is authoritative: a stale legacy flag cannot close needs_review, # and a non-open legacy status cannot be reopened by requires_review=1. conn.execute("UPDATE budget_transaction_candidates SET requires_review=0 WHERE transaction_candidate_id='proposal'") conn.execute( """UPDATE budget_transaction_candidates SET status='pending',requires_review=1,classification='possible_logical_duplicate' WHERE transaction_candidate_id='covered'""" ) conn.execute("UPDATE budget_categories SET is_active=0 WHERE category_id=?", (expense_category[0],)) conn.commit() first = get_household_review(conn, limit=1) assert (first["total_open"], first["proposal_ready"], first["decision_needed"], first["special_cases"]) == (5, 1, 3, 1) assert first["review_count"] == first["total_open"] and first["filtered_total"] == 5 assert {item["value"] for item in first["source_options"]} == {"visa_credit_card", "raiffeisen_bank"} assert len(first["items"]) == 1 and first["has_more"] is True and first["next_cursor"] second = get_household_review(conn, limit=1, cursor=first["next_cursor"]) assert second["data_version"] == first["data_version"] assert second["items"][0]["item_token"] != first["items"][0]["item_token"] conn.execute( "UPDATE budget_transaction_candidates SET amount_original='999.00' WHERE transaction_candidate_id='wrong-proposal'" ) conn.commit() with pytest.raises(HTTPException) as stale_review: get_household_review(conn, limit=1, cursor=first["next_cursor"]) assert stale_review.value.status_code == 409 fresh = get_household_review(conn, limit=1) conn.execute( """INSERT INTO budget_review_rules( rule_id,merchant_contains,source_type,category_id,target_status,is_active, priority,confidence,created_at,updated_at) VALUES ('review-rule-mutation','wrong-proposal','raiffeisen_bank',?,'review',1, 100,'1.0','x','x')""", (expense_category[0],), ) conn.commit() with pytest.raises(HTTPException) as stale_classifier: get_household_review(conn, limit=1, cursor=fresh["next_cursor"]) assert stale_classifier.value.status_code == 409 special = get_household_review(conn, limit=25, special_only=True) assert special["total_open"] == 5 and special["filtered_total"] == 1 assert special["items"][0]["user_state"] == "special_case" assert special["items"][0]["can_confirm"] is False needs_decision = get_household_review(conn, limit=25, state="needs_decision") assert needs_decision["filtered_total"] == 4 assert all(item["user_state"] != "proposal_ready" for item in needs_decision["items"]) proposals = get_household_review(conn, limit=25, state="proposal_ready") assert proposals["filtered_total"] == 1 assert {item["user_state"] for item in proposals["items"]} == {"proposal_ready"} decisions = get_household_review(conn, limit=25, state="decision_needed") assert decisions["filtered_total"] == 3 assert {item["user_state"] for item in decisions["items"]} == {"decision_needed"} def test_financial_semantics_reconcile_refunds_transfers_categories_and_net() -> None: conn = database() expense_category = conn.execute( "SELECT category_id FROM budget_categories WHERE is_active=1 AND category_type='expense' LIMIT 1" ).fetchone()[0] expense_category_name = conn.execute( "SELECT name FROM budget_categories WHERE category_id=?", (expense_category,), ).fetchone()[0] income_category = conn.execute( "SELECT category_id FROM budget_categories WHERE is_active=1 AND category_type='income' LIMIT 1" ).fetchone()[0] def transaction(tx_id: str, tx_type: str, amount: str, category_id=None, reversal=None) -> None: conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,description, amount_original,currency_original,amount_chf,fx_status,category_id,status,source_type, reversal_of_transaction_id,created_at,updated_at) VALUES (?,?,?,'2026-04-15',?,?,'CHF',?,'not_needed',?,'confirmed','manual',?,'2026-04-15T00:00:00Z','2026-04-15T00:00:00Z')""", (tx_id, "bacc_rai", tx_type, tx_id, amount, amount, category_id, reversal), ) transaction("expense", "expense", "-100.00", expense_category) transaction("income", "income", "200.00", income_category) transaction("linked-refund", "refund", "20.00", None, "expense") conn.execute( "UPDATE budget_transactions SET reversal_of_transaction_id=NULL,notes=? WHERE budget_transaction_id='linked-refund'", (json.dumps({"reversal_of_transaction_id": "expense"}),), ) transaction("unlinked-refund", "refund", "5.00") transaction("transfer-out", "transfer", "-50.00") transaction("card-settlement", "transfer", "50.00") conn.commit() summary = get_household_financial_summary(conn, period="2026-04") assert summary["income_chf"] == "200.00" assert summary["expense_chf"] == "80.00" assert summary["net_chf"] == "120.00" assert summary["unlinked_refund_count"] == 1 assert summary["data_status"] == "partial" assert summary["neutral_transfer_count"] == 2 assert sum(Decimal(row["expense_chf"]) for row in summary["categories"]) == Decimal("80.00") assert sum(Decimal(row["income_chf"]) for row in summary["categories"]) == Decimal("200.00") overview = get_household_overview(conn, month="2026-04") assert (overview["income_chf"], overview["expense_chf"], overview["balance_chf"]) == ("200.00", "80.00", "120.00") assert overview["neutral_transfers_chf"] == "0.00" assert overview["data_status"]["status"] == "partial" assert "nicht verknüpfte Rückerstattung" in overview["data_status"]["message"] conn.execute( "UPDATE budget_transactions SET updated_at='2026-04-17T00:00:00Z' WHERE budget_transaction_id='income'" ) conn.commit() page = list_household_transactions(conn, period="2026-04", limit=2) assert page["total_all"] == 6 and page["filtered_total"] == 6 and page["page_count"] == 2 assert page["has_more"] is True and page["next_cursor"] conn.execute("UPDATE budget_transactions SET amount_chf='999.00' WHERE budget_transaction_id='expense'") conn.commit() with pytest.raises(HTTPException) as stale: list_household_transactions(conn, period="2026-04", limit=2, cursor=page["next_cursor"]) assert stale.value.status_code == 409 category_page = list_household_transactions( conn, period="2026-04", category=expense_category_name, limit=10, ) assert category_page["filtered_total"] == 2 assert {item["transaction_type"] for item in category_page["items"]} == {"expense", "refund"} assert {item["category_name"] for item in category_page["items"]} == {expense_category_name} def test_unconverted_foreign_currency_is_partial_and_never_invented_as_chf() -> None: conn = database() category_id = conn.execute( "SELECT category_id FROM budget_categories WHERE category_type='expense' AND is_active=1 LIMIT 1" ).fetchone()[0] conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,description, amount_original,currency_original,amount_chf,fx_status,category_id,status,source_type, created_at,updated_at) VALUES ('foreign-missing','bacc_rai','expense','2026-05-01','Synthetic foreign', '100.00','EUR',NULL,'missing',?,'confirmed','manual','x','x')""", (category_id,), ) conn.commit() summary = get_household_financial_summary(conn, period="2026-05") assert summary["expense_chf"] == "0.00" and summary["net_chf"] == "0.00" assert summary["unavailable_chf_count"] == 1 and summary["data_status"] == "partial" overview = get_household_overview(conn, month="2026-05") assert overview["expense_chf"] == "0.00" assert overview["unavailable_chf_count"] == 1 assert overview["data_status"]["status"] == "partial" monthly = get_monthly_summary(conn, month="2026-05") assert monthly["unavailable_chf_count"] == 1 assert monthly["data_status"] == "partial" page = list_household_transactions(conn, period="2026-05") assert page["items"][0]["amount_chf"] is None assert page["items"][0]["amount_original"] == "100.00" assert page["items"][0]["currency"] == "EUR" def test_refund_links_are_eligible_bounded_and_writer_validated() -> None: conn = database() expense_category = conn.execute( "SELECT category_id FROM budget_categories WHERE category_type='expense' AND is_active=1 LIMIT 1" ).fetchone()[0] income_category_row = conn.execute( "SELECT category_id,name FROM budget_categories WHERE category_type='income' AND is_active=1 LIMIT 1" ).fetchone() income_category = income_category_row["category_id"] def transaction(tx_id: str, tx_type: str, amount: str, category_id: str, reversal: str | None = None) -> None: conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,description, amount_original,currency_original,amount_chf,fx_status,category_id,status,source_type, reversal_of_transaction_id,created_at,updated_at) VALUES (?, 'bacc_rai', ?, '2026-06-01', ?, ?, 'CHF', ?, 'not_needed', ?, 'confirmed','manual',? ,?,?)""", (tx_id, tx_type, tx_id, amount, amount, category_id, reversal, tx_id, tx_id), ) transaction("expense-origin", "expense", "100.00", expense_category) transaction("income-origin", "income", "100.00", income_category) transaction("refund-valid", "refund", "60.00", expense_category, "expense-origin") transaction("refund-excess", "refund", "50.00", expense_category, "expense-origin") transaction("refund-income", "refund", "20.00", expense_category, "income-origin") conn.commit() summary = get_household_financial_summary(conn, period="2026-06") assert (summary["income_chf"], summary["expense_chf"], summary["net_chf"]) == ( "100.00", "50.00", "50.00", ) assert summary["unlinked_refund_count"] == 2 and summary["data_status"] == "partial" income_page = list_household_transactions( conn, period="2026-06", category=str(income_category_row["name"]), limit=10, ) assert income_page["filtered_total"] == 1 assert income_page["items"][0]["transaction_type"] == "income" charts = get_budget_charts_2026(conn, year="2026") june = next(row for row in charts["cashflow_by_month"] if row["month"] == "2026-06") assert june["data_status"] == "partial" invalid_income_link = { "account_id": "bacc_rai", "transaction_type": "refund", "transaction_date": "2026-06-02", "description": "Synthetic invalid refund", "amount_original": "10.00", "currency_original": "CHF", "category_id": expense_category, "reversal_of_transaction_id": "income-origin", } with pytest.raises(HTTPException) as invalid_origin: create_budget_transaction_preview(conn, invalid_income_link) assert invalid_origin.value.status_code == 422 with pytest.raises(HTTPException) as excessive: confirm_budget_transaction( conn, invalid_income_link | { "description": "Synthetic excessive refund", "reversal_of_transaction_id": "expense-origin", }, ) assert excessive.value.status_code == 422 def test_missing_fx_transfer_volume_is_unavailable_not_zero() -> None: conn = database() conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,description, amount_original,currency_original,amount_chf,fx_status,status,source_type,created_at,updated_at) VALUES ('foreign-transfer','bacc_rai','transfer','2026-06-01','Synthetic FX transfer', '100.00','EUR',NULL,'missing','confirmed','manual','x','x')""" ) conn.commit() summary = get_household_financial_summary(conn, period="2026-06") assert summary["neutral_transfer_count"] == 1 assert summary["neutral_transfer_volume_chf"] is None assert summary["unavailable_transfer_count"] == 1 assert summary["unavailable_chf_count"] == 1 and summary["data_status"] == "partial" monthly = get_monthly_summary(conn, month="2026-06") assert monthly["neutral_transfer_volume_chf"] is None assert monthly["unavailable_transfer_count"] == 1 and monthly["data_status"] == "partial" charts = get_budget_charts_2026(conn, year="2026") june = next(row for row in charts["cashflow_by_month"] if row["month"] == "2026-06") assert june["data_status"] == "partial" def test_paired_transfer_volume_is_counted_once_and_unmatched_once() -> None: conn = database() confirm_transfer(conn, { "from_account_id": "bacc_rai", "to_account_id": "bacc_akb", "amount_original": "900.00", "currency_original": "CHF", "transaction_date": "2026-06-02", "description": "Synthetic internal transfer", }) summary = get_household_financial_summary(conn, period="2026-06") assert summary["neutral_transfer_count"] == 2 assert summary["neutral_transfer_volume_chf"] == "900.00" assert summary["unavailable_transfer_count"] == 0 overview = get_household_overview(conn, month="2026-06") assert overview["neutral_transfers_chf"] == "0.00" assert overview["neutral_transfer_volume_chf"] == "900.00" for account_name in ("Household A", "Household B"): scoped = get_household_financial_summary(conn, period="2026-06", account=account_name) assert scoped["neutral_transfer_volume_chf"] == "900.00" conn.execute( """INSERT INTO budget_transfers( transfer_id,from_transaction_id,to_transaction_id,from_account_id,to_account_id, amount_original,currency_original,fx_rate_to_chf,created_at,transfer_type) SELECT 'duplicate-membership',from_transaction_id,to_transaction_id,from_account_id,to_account_id, amount_original,currency_original,fx_rate_to_chf,'x',transfer_type FROM budget_transfers LIMIT 1""" ) conflict = get_household_financial_summary(conn, period="2026-06") assert conflict["confirmed_transaction_count"] == 2 assert conflict["neutral_transfer_count"] == 2 assert conflict["neutral_transfer_volume_chf"] is None assert conflict["unavailable_chf_count"] == 0 assert conflict["transfer_membership_conflict_count"] == 2 assert conflict["data_status"] == "partial" assert [item["code"] for item in conflict["warnings"]] == ["transfer_membership_conflict"] conflict_month = get_monthly_summary(conn, month="2026-06") assert conflict_month["transfer_membership_conflict_count"] == 2 assert conflict_month["data_status"] == "partial" conflict_charts = get_budget_charts_2026(conn, year="2026") conflict_june = next(row for row in conflict_charts["cashflow_by_month"] if row["month"] == "2026-06") assert conflict_june["data_status"] == "partial" conn.execute("DELETE FROM budget_transfers WHERE transfer_id='duplicate-membership'") malformed_category = confirm_create_category( conn, {"name": "Malformed membership category", "category_type": "expense"} )["entity_id"] confirm_budget_transaction(conn, { "account_id": "bacc_rai", "category_id": malformed_category, "transaction_type": "expense", "transaction_date": "2026-06-02", "description": "Malformed membership expense", "amount_original": "50.00", "currency_original": "CHF", }) confirm_transfer(conn, { "from_account_id": "bacc_rai", "to_account_id": "bacc_akb", "amount_original": "30.00", "currency_original": "CHF", "transaction_date": "2026-06-02", "description": "Malformed membership counterpart", }) malformed_pair = conn.execute( "SELECT from_transaction_id,to_transaction_id FROM budget_transfers WHERE amount_original='30.00'" ).fetchone() expense_id = conn.execute( "SELECT budget_transaction_id FROM budget_transactions WHERE description='Malformed membership expense'" ).fetchone()[0] conn.execute("DELETE FROM budget_transfers WHERE amount_original='30.00'") for index, counterpart_id in enumerate(malformed_pair, 1): conn.execute( """INSERT INTO budget_transfers( transfer_id,from_transaction_id,to_transaction_id,from_account_id,to_account_id, amount_original,currency_original,created_at,transfer_type) VALUES(?,?,?,?,?,?,?,'x','internal')""", (f"malformed-{index}", expense_id, counterpart_id, "bacc_rai", "bacc_akb", str(index * 10), "CHF"), ) malformed = get_household_financial_summary(conn, period="2026-06") assert malformed["confirmed_transaction_count"] == 5 assert malformed["neutral_transfer_count"] == 4 assert malformed["expense_chf"] == "50.00" assert next( row["expense_chf"] for row in malformed["categories"] if row["category_id"] == malformed_category ) == "50.00" assert malformed["neutral_transfer_volume_chf"] is None assert malformed["unavailable_transfer_count"] == 1 assert malformed["transfer_membership_conflict_count"] == 1 malformed_overview = get_household_overview(conn, month="2026-06") assert malformed_overview["data_status"]["status"] == "partial" assert malformed_overview["transfer_membership_conflict_count"] == 1 conn.execute("DELETE FROM budget_transfers WHERE transfer_id LIKE 'malformed-%'") conn.execute("DELETE FROM budget_transactions WHERE description LIKE 'Malformed membership%'") confirm_budget_transaction(conn, { "account_id": "bacc_rai", "transaction_type": "transfer", "transaction_date": "2026-06-03", "description": "Synthetic unmatched transfer", "amount_original": "100.00", "currency_original": "CHF", }) with_unmatched = get_household_financial_summary(conn, period="2026-06") assert with_unmatched["neutral_transfer_count"] == 3 assert with_unmatched["neutral_transfer_volume_chf"] == "1000.00" confirm_transfer(conn, { "from_account_id": "bacc_rai", "to_account_id": "bacc_akb", "amount_original": "50.00", "currency_original": "EUR", "transaction_date": "2026-07-02", "description": "Synthetic unconverted pair", }) unavailable_pair = get_household_financial_summary(conn, period="2026-07") assert unavailable_pair["neutral_transfer_count"] == 2 assert unavailable_pair["neutral_transfer_volume_chf"] is None assert unavailable_pair["unavailable_transfer_count"] == 1