from __future__ import annotations import json from collections import Counter from io import BytesIO from sqlite3 import Connection import pytest from fastapi import HTTPException from openpyxl import Workbook import jarvis_finance.services.household_import as household_import_service from jarvis_finance.services.budget_csv_imports import parse_budget_csv_text from jarvis_finance.services.budget_recurring import confirm_manual_recurring_payment from jarvis_finance.services.household_classification import ( CLASSIFICATION_VERSION, ClassificationLookupCache, classify_household_row, extract_counterparty, normalize_merchant, ) from jarvis_finance.services.household_import import ( _mapping_account_is_compatible, authorize_owner_neutral_cluster, configure_source_mapping, confirm_household_import, confirm_household_review_batch, get_household_overview, get_household_review, preview_household_import, preview_household_review_batch, ) from jarvis_finance.services.household_source_analysis import ( analyze_akb_balance_workbook, parse_raiffeisen_physical_records, summarize_migros_rows, summarize_visa_rows, ) from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations def database() -> Connection: conn = connect_memory() apply_migrations(conn) conn.execute("INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES ('p','Synthetic','bank','2026-01-01')") for key, account_type in (("bank", "cash"), ("bank2", "cash"), ("card", "credit_card")): 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", f"Synthetic {key}", account_type, "liability" if account_type == "credit_card" 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}", f"Synthetic {key}", account_type), ) for key, label in (("food", "Essen & Haushalt"), ("mobility", "Auto Melanie")): conn.execute( "INSERT INTO budget_categories(category_id,name,category_type,is_active,sort_order,created_at) VALUES (?,?, 'expense',1,1,'2026-01-01')", (f"cat_{key}", label), ) conn.execute( "INSERT INTO budget_categories(category_id,name,category_type,is_active,sort_order,created_at) " "VALUES ('cat_other_income','Sonstige Einnahmen','income',1,3,'2026-01-01')" ) conn.commit() configure_source_mapping(conn, {"confirm": True, "source_type": "raiffeisen_bank", "source_reference": "SYN-BANK-1", "budget_account_id": "bacc_bank"}) configure_source_mapping(conn, {"confirm": True, "source_type": "akb_bank", "source_reference": "SYN-BANK-2", "budget_account_id": "bacc_bank2"}) configure_source_mapping(conn, {"confirm": True, "source_type": "visa_credit_card", "source_reference": "SYN-CARD-1", "budget_account_id": "bacc_card"}) return conn def row(*, merchant: str = "Example Shop", source: str = "visa_credit_card", amount: str = "-10.00", tx_date: str = "2026-07-01") -> dict: account = "bacc_card" if source == "visa_credit_card" else "bacc_bank" return { "source_type": source, "description": merchant, "merchant": merchant, "signed_amount": amount, "amount": amount.lstrip("-"), "currency": "CHF", "transaction_date": tx_date, "classification": "expense_candidate", "disposition": "candidate", "requires_review": True, "mapping": {"budget_account_id": account}, } def confirmed_history(conn: Connection, *, merchant: str, category: str, source: str = "visa_credit_card", index: int = 1) -> None: candidate_id = f"hist_c_{category}_{index}" transaction_id = f"hist_t_{category}_{index}" conn.execute( """INSERT INTO budget_transaction_candidates( transaction_candidate_id,source_file_label,source_row_or_range,source_type,transaction_date, description,merchant,amount_original,signed_amount_original,currency_original,proposed_category_id, confidence,requires_review,status,classification,account_source,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,0,'confirmed','expense_candidate',?, '2026-01-01','2026-01-01')""", (candidate_id, "synthetic", f"R{index}", source, f"2026-01-{index:02d}", merchant, merchant, "10", "-10", "CHF", category, "1", "bacc_card"), ) conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,description,payee,amount_original, currency_original,amount_chf,fx_status,category_id,status,source_type,source_candidate_id,created_at,updated_at) VALUES (?,?, 'expense',?,?,?,?,?,'-10','not_needed',?,'confirmed','import_candidate',?,'2026-01-01','2026-01-01')""", (transaction_id, "bacc_card", f"2026-01-{index:02d}", merchant, merchant, "-10", "CHF", category, candidate_id), ) conn.execute( "UPDATE budget_transaction_candidates SET confirmed_transaction_id=? WHERE transaction_candidate_id=?", (transaction_id, candidate_id), ) conn.commit() def visa_payload(description: str = "Example Shop", amount: str = "10.00") -> dict: csv_text = f"TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\nT-NEW,SYN-CARD-1,2026-07-01,{amount},CHF,{description},booked\n" return {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]} def review_ready_visa_payload(description: str = "Example Shop") -> dict: header = "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" known = "".join( f"T-KNOWN-{index},SYN-CARD-1,2026-{'07' if index < 28 else '08'}-{index + 1 if index < 28 else index - 27:02d},10.00,CHF,Migros,booked\n" for index in range(33) ) csv_text = header + known + ( f"T-REVIEW,SYN-CARD-1,2026-07-28,13.00,CHF,{description},booked\n" ) return {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]} def confirm_payload(payload: dict, preview: dict) -> dict: return payload | {"confirm": True, "preview_fingerprint": preview["preview_fingerprint"], "baseline_fingerprint": preview["baseline_fingerprint"]} def test_source_mapping_rejects_account_role_mismatches() -> None: conn = database() with pytest.raises(HTTPException) as visa_cash: configure_source_mapping(conn, { "confirm": True, "source_type": "visa_credit_card", "source_reference": "SYN-WRONG-CARD", "budget_account_id": "bacc_bank", }) assert visa_cash.value.status_code == 422 with pytest.raises(HTTPException) as bank_card: configure_source_mapping(conn, { "confirm": True, "source_type": "raiffeisen_bank", "source_reference": "SYN-WRONG-BANK", "budget_account_id": "bacc_card", }) assert bank_card.value.status_code == 422 conn.execute("UPDATE accounts SET portfolio_bucket='other' WHERE account_id='acct_bank'") conn.commit() with pytest.raises(HTTPException) as bank_non_cash_bucket: configure_source_mapping(conn, { "confirm": True, "source_type": "raiffeisen_bank", "source_reference": "SYN-WRONG-BUCKET", "budget_account_id": "bacc_bank", }) assert bank_non_cash_bucket.value.status_code == 422 def test_source_mapping_requires_explicit_chf_non_performance_portfolio_roles() -> None: base = { "budget_account_type": "checking", "canonical_account_type": "cash", "budget_currency": "CHF", "canonical_currency": "CHF", "performance_included": 0, "portfolio_bucket": "cash", } assert _mapping_account_is_compatible("raiffeisen_bank", base) assert not _mapping_account_is_compatible("raiffeisen_bank", base | {"performance_included": 1}) assert not _mapping_account_is_compatible( "raiffeisen_bank", base | {"budget_currency": "EUR", "canonical_currency": "EUR"} ) card = base | { "budget_account_type": "credit_card", "canonical_account_type": "credit_card_liability", "portfolio_bucket": "liability", } assert _mapping_account_is_compatible("visa_credit_card", card) assert not _mapping_account_is_compatible("visa_credit_card", card | {"portfolio_bucket": "investments"}) assert not _mapping_account_is_compatible("visa_credit_card", card | {"performance_included": 1}) assert not _mapping_account_is_compatible( "visa_credit_card", card | {"budget_currency": "EUR", "canonical_currency": "EUR"} ) def test_source_mapping_identical_retry_is_audited_noop() -> None: conn = database() before = conn.execute( "SELECT COUNT(*) FROM audit_log WHERE action='account_source_mapping_confirmed'" ).fetchone()[0] repeated = configure_source_mapping(conn, { "confirm": True, "source_type": "raiffeisen_bank", "source_reference": "SYN-BANK-1", "budget_account_id": "bacc_bank", }) after = conn.execute( "SELECT COUNT(*) FROM audit_log WHERE action='account_source_mapping_confirmed'" ).fetchone()[0] assert repeated["status"] == "unchanged" and repeated["idempotent"] is True assert after == before def test_stored_source_mapping_fails_closed_after_account_role_drift() -> None: visa_conn = database() visa_conn.execute("UPDATE budget_accounts SET account_type='cash' WHERE budget_account_id='bacc_card'") visa_conn.execute("UPDATE accounts SET account_type='cash',portfolio_bucket='cash' WHERE account_id='acct_card'") visa_conn.commit() visa_preview = preview_household_import(visa_conn, visa_payload()) assert visa_preview["confirmable"] is False assert {error["code"] for error in visa_preview["errors"]} == {"account_source_mapping_missing_or_mismatch"} bank_conn = database() bank_conn.execute("UPDATE budget_accounts SET account_type='credit_card' WHERE budget_account_id='bacc_bank'") bank_conn.execute("UPDATE accounts SET account_type='credit_card',portfolio_bucket='liability' WHERE account_id='acct_bank'") bank_conn.commit() bank_csv = ( "IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\n" "SYN-BANK-1;2026-07-01;Synthetic Shop;-10.00;90.00;2026-07-01\n" ) bank_preview = preview_household_import(bank_conn, {"files": [{"profile": "raiffeisen_bank", "csv_text": bank_csv}]}) assert bank_preview["confirmable"] is False assert {error["code"] for error in bank_preview["errors"]} == {"account_source_mapping_missing_or_mismatch"} def test_auto_detects_legacy_comma_raiffeisen_with_unquoted_description_delimiter() -> None: parsed = parse_budget_csv_text( "Account,Booked At,Text,Credit/Debit Amount,Balance,Valuta Date\n" "SYN-BANK-1,2026-07-01,Synthetic Shop, City,-10.00,90.00,2026-07-01\n" ) assert parsed["profile"] == "raiffeisen_bank" assert parsed["logical_row_count"] == 1 assert parsed["rows"][0]["Text"] == "Synthetic Shop, City" def test_classification_v2_is_deterministic_and_normalizes_merchant() -> None: conn = database() first = classify_household_row(conn, row(merchant=" EXAMPLE SHOP ")) second = classify_household_row(conn, row(merchant="example shop")) assert CLASSIFICATION_VERSION == "household_classification_v2.3" assert normalize_merchant(" EXAMPLE SHOP ") == "example shop" assert first == second def test_health_insurance_family_precedes_medical_providers_and_accident_insurance_fails_closed() -> None: conn = database() conn.execute( "INSERT INTO budget_categories(category_id,name,category_type,is_active,sort_order,created_at) " "VALUES ('cat_health','Gesundheit','expense',1,4,'2026-01-01')" ) conn.execute( "INSERT INTO budget_categories(category_id,name,category_type,is_active,sort_order,created_at) " "VALUES ('cat_health_insurance','Krankenkasse','expense',1,5,'2026-01-01')" ) insurer = classify_household_row(conn, row(merchant="ÖKK Krankenversicherung")) accident = classify_household_row(conn, row(merchant="Synthetic Unfallversicherung")) assert insurer["origin"] == "stable_merchant_family" assert insurer["category_id"] == "cat_health_insurance" assert insurer["user_state"] == "proposal_ready" assert accident["category_id"] is None assert accident["user_state"] == "decision_needed" def test_two_consistent_confirmations_create_a_future_only_learned_suggestion() -> None: conn = database() confirmed_history(conn, merchant="Example Shop", category="cat_food", index=1) confirmed_history(conn, merchant="Example Shop", category="cat_food", index=2) result = classify_household_row(conn, row()) assert result["category_id"] == "cat_food" assert result["user_state"] == "proposal_ready" assert result["learned_rule"] is True assert result["evidence_count"] == 2 assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 2 def test_one_confirmation_can_suggest_but_does_not_create_global_rule() -> None: conn = database() confirmed_history(conn, merchant="Example Shop", category="cat_food", index=1) result = classify_household_row(conn, row()) assert result["category_id"] == "cat_food" assert result["learned_rule"] is False assert result["evidence_count"] == 1 def test_conflicting_confirmed_history_requires_decision() -> None: conn = database() confirmed_history(conn, merchant="Example Shop", category="cat_food", index=1) confirmed_history(conn, merchant="Example Shop", category="cat_mobility", index=2) result = classify_household_row(conn, row()) assert result["category_id"] is None assert result["user_state"] == "decision_needed" assert result["conflict_count"] == 2 def test_confirmed_history_is_isolated_by_source_even_for_same_merchant_family() -> None: conn = database() confirmed_history(conn, merchant="Coop Filiale 100", category="cat_food", source="akb_bank", index=1) confirmed_history(conn, merchant="Coop Store 200", category="cat_food", source="akb_bank", index=2) result = classify_household_row( conn, row(merchant="Coop Terminal ABC123", source="raiffeisen_bank"), ) assert result["category_id"] == "cat_food" assert result["origin"] == "stable_merchant_family" assert result["evidence_count"] == 0 def test_explicit_exact_merchant_rule_precedes_history() -> None: conn = database() conn.execute("INSERT INTO budget_merchants(merchant_id,display_name,normalized_name,default_category_id,is_active,created_at) VALUES ('m1','Example Shop','example shop','cat_mobility',1,'2026-01-01')") conn.commit() confirmed_history(conn, merchant="Example Shop", category="cat_food", index=1) confirmed_history(conn, merchant="Example Shop", category="cat_food", index=2) result = classify_household_row(conn, row()) assert result["category_id"] == "cat_mobility" assert result["origin"] == "exact_merchant_rule" def test_unmatched_bank_text_never_becomes_a_transaction_specific_transfer_exception() -> None: conn = database() result = classify_household_row( conn, row(merchant="Synthetic Lottery Mobile", source="raiffeisen_bank", amount="-91.00", tx_date="2026-02-11"), ) assert result["user_state"] == "decision_needed" assert result["transaction_semantics"] == "expense" assert result["budget_effect_chf"] is None assert result["create_counterbooking"] is False def test_card_payment_is_never_income_and_unmatched_stays_neutral() -> None: conn = database() item = row(merchant="IHRE ZAHLUNG – DANKE", source="visa_credit_card", amount="865.00") item["classification"] = "credit_card_payment_counterpost" result = classify_household_row(conn, item) assert result["transaction_semantics"] == "credit_card_payment" assert result["budget_effect_chf"] == "0.00" assert result["category_id"] is None def test_category_override_changes_preview_fingerprint_and_confirm_writes_exact_category() -> None: conn = database() payload = visa_payload() base = preview_household_import(conn, payload) overridden_payload = payload | {"category_overrides": {"row_1": "cat_food"}} overridden = preview_household_import(conn, overridden_payload) assert base["preview_fingerprint"] != overridden["preview_fingerprint"] assert overridden["items"][0]["category_id"] == "cat_food" result = confirm_household_import(conn, confirm_payload(overridden_payload, overridden)) assert result["idempotent"] is False tx = conn.execute("SELECT category_id FROM budget_transactions WHERE source_type='import_candidate'").fetchone() assert tx["category_id"] == "cat_food" assert confirm_household_import(conn, confirm_payload(overridden_payload, overridden))["idempotent"] is True def test_stale_override_preview_is_rejected() -> None: conn = database() payload = visa_payload() | {"category_overrides": {"row_1": "cat_food"}} preview = preview_household_import(conn, payload) conn.execute("UPDATE budget_categories SET updated_at='changed' WHERE category_id='cat_food'") conn.commit() with pytest.raises(HTTPException) as exc: confirm_household_import(conn, confirm_payload(payload, preview)) assert exc.value.status_code == 409 def test_bank_interest_reuses_existing_income_category_and_is_not_a_transfer() -> None: result = classify_household_row( database(), row(merchant="Zinsgutschrift", source="raiffeisen_bank", amount="1.25"), ) assert result["category_id"] == "cat_other_income" assert result["category_name"] == "Sonstige Einnahmen" assert result["transaction_semantics"] == "income" assert result["user_state"] == "proposal_ready" assert result["origin"] == "bank_interest_existing_category" def test_card_settlement_three_day_edge_is_ambiguous_while_two_days_is_safe() -> None: payload = {"files": [ { "profile": "akb_bank", "source_reference": "SYN-BANK-2", "csv_text": ( "Buchung;Valuta;Buchungstext;Belastung;Gutschrift\n" "01.07.2026;01.07.2026;Viseca payment;100.00;\n" "10.07.2026;10.07.2026;Viseca payment;200.00;\n" ), }, { "profile": "visa_credit_card", "csv_text": ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType,OriginalAmount\n" "PAY-1,SYN-CARD-1,2026-07-04,-100.00,CHF,Ihre Zahlung,BOOKED,-100.00\n" "PAY-2,SYN-CARD-1,2026-07-12,-200.00,CHF,Ihre Zahlung,BOOKED,-200.00\n" ), }, ]} preview = preview_household_import(database(), payload) assert preview["confirmable"] is True assert preview["counts"]["safe_transfer_pairs"] == 1 assert Counter(pair["pairing_class"] for pair in preview["transfer_pairs"]) == { "safe": 1, "ambiguous": 1, } def test_legacy_migros_receipt_candidate_is_not_treated_as_money_movement() -> None: conn = database() conn.execute( """INSERT INTO budget_transaction_candidates( transaction_candidate_id,source_file_label,source_row_or_range,source_type, transaction_date,description,merchant,amount_original,signed_amount_original, currency_original,confidence,requires_review,status,classification,created_at,updated_at) VALUES ('legacy_receipt','legacy','R1','migros_receipt','2026-07-01', 'Migros receipt','Migros','10.00','-10.00','CHF','1',0,'confirmed', 'migros_receipt_detail','2026-01-01','2026-01-01')""" ) conn.commit() preview = preview_household_import(conn, {"files": [{ "profile": "migros_receipts", "csv_text": ( "Datum,Zeit,Filiale,Kassennummer,Transaktionsnummer,Umsatz,Artikel\n" "2026-07-01,12:00,Store,1,TX-1,10.00,Item\n" ), }]}) assert preview["counts"]["linked_migros_receipts"] == 0 assert preview["counts"]["unlinked_migros_receipts"] == 1 def test_migros_links_every_bidirectionally_unique_receipt_without_amount_threshold() -> None: payload = {"files": [ { "profile": "visa_credit_card", "csv_text": ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType,OriginalAmount\n" "M-1,SYN-CARD-1,2026-07-01,60.00,CHF,Migros,BOOKED,60.00\n" "M-2,SYN-CARD-1,2026-07-02,40.00,CHF,Migros,BOOKED,40.00\n" ), }, { "profile": "migros_receipts", "csv_text": ( "Datum,Zeit,Filiale,Kassennummer,Transaktionsnummer,Umsatz,Artikel\n" "2026-07-01,12:00,Store 1,1,R-1,60.00,High value basket\n" "2026-07-02,12:00,Store 1,1,R-2,40.00,Small basket\n" ), }, ]} preview = preview_household_import(database(), payload) assert preview["counts"]["linked_migros_receipts"] == 2 assert preview["counts"]["unlinked_migros_receipts"] == 0 small = next(row for row in preview["rows"] if row["source_type"] == "migros_receipts" and row["amount"] == "40.00") assert small["requires_review"] is False assert small["disposition"] == "receipt_detail" def test_raiffeisen_same_day_same_amount_rows_with_distinct_balances_are_never_fuzzy_deduplicated() -> None: csv_text = ( "IBAN,Booked At,Text,Credit/Debit Amount,Balance,Valuta Date\n" "SYN-REF-A,2026-07-01,COOP BASEL 1,-10.00,100.00,2026-07-01\n" "SYN-REF-A,2026-07-01,COOP BASEL 2,-10.00,90.00,2026-07-01\n" ) parsed = parse_budget_csv_text(csv_text, "raiffeisen_bank") assert parsed["physical_row_count"] == 2 assert parsed["logical_row_count"] == 2 def test_raiffeisen_forced_profile_recovers_unquoted_description_commas() -> None: csv_text = ( "IBAN,Booked At,Text,Credit/Debit Amount,Valuta Date\n" "SYN-REF-A,2026-07-01,Shop, Zurich,-10.00,2026-07-01\n" ) parsed = parse_budget_csv_text(csv_text, "raiffeisen_bank") assert parsed["delimiter"] == "," assert parsed["logical_row_count"] == 1 assert parsed["rows"][0]["Credit/Debit Amount"] == "-10.00" def test_raiffeisen_127_lines_become_69_logical_rows_across_two_references() -> None: header = ["IBAN", "Booked At", "Text", "Credit/Debit Amount", "Balance", "Valuta Date"] physical: list[list[str]] = [] for index in range(69): is_interest = index == 68 physical.append([ "SYN-REF-B" if is_interest else "SYN-REF-A", "2026-07-28", "Interest credit" if is_interest else f"Synthetic booking {index}", "1.25" if is_interest else f"-{index + 1}.00", "20.00" if is_interest else "1000.00", "2026-07-28", ]) if index < 57: physical.append(["", "", f"Continuation detail {index}"]) logical = parse_raiffeisen_physical_records(physical) assert 1 + len(physical) == 127 # provider count includes the header assert len(header) == 6 and len(logical) == 69 assert Counter(item["IBAN"] for item in logical) == {"SYN-REF-A": 68, "SYN-REF-B": 1} assert "Continuation detail 0" in logical[0]["Text"] interest = next(item for item in logical if item["IBAN"] == "SYN-REF-B") assert (interest["Credit/Debit Amount"], interest["Balance"]) == ("1.25", "20.00") assert logical[67]["Balance"] == "1000.00" def test_one_file_can_bind_two_confirmed_source_mappings() -> None: conn = database() configure_source_mapping(conn, { "confirm": True, "source_type": "raiffeisen_bank", "source_reference": "SYN-BANK-2", "budget_account_id": "bacc_bank2", }) mapping_ids = [ str(row[0]) for row in conn.execute( "SELECT mapping_id FROM household_account_source_mappings " "WHERE source_type='raiffeisen_bank' ORDER BY mapping_id" ).fetchall() ] csv_text = ( "IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\n" "SYN-BANK-1;2026-01-05;Synthetic expense;-10.00;1000.00;2026-01-05\n" "SYN-BANK-2;2026-01-06;Interest booking;1.25;20.00;2026-01-06\n" ) base_file = {"profile": "raiffeisen_bank", "csv_text": csv_text} rejected = preview_household_import( conn, {"files": [base_file | {"mapping_id": mapping_ids[0]}]}, ) assert rejected["confirmable"] is False payload = {"files": [base_file | {"mapping_ids": mapping_ids}]} preview = preview_household_import(conn, payload) assert preview["confirmable"] is True assert preview["counts"]["rows"] == 2 assert len({item["account_hint"] for item in preview["items"]}) == 2 reversed_selection = {"files": [base_file | {"mapping_ids": list(reversed(mapping_ids))}]} assert preview_household_import(conn, reversed_selection)["preview_fingerprint"] == preview["preview_fingerprint"] def test_akb_xlsx_creates_balance_snapshots_and_no_transactions() -> None: workbook = Workbook() sheet = workbook.active sheet.append(["Account", "Balance", "Currency", "As of"]) sheet.append(["Mortgage interest", 100.00, "CHF", "2026-07-28"]) sheet.append(["Savings", 200.00, "CHF", "2026-07-28"]) sheet.append(["Household", 300.00, "CHF", "2026-07-28"]) output = BytesIO() workbook.save(output) result = analyze_akb_balance_workbook(output.getvalue()) assert result["snapshot_count"] == 3 assert result["total_chf"] == "600.00" assert result["transaction_count"] == 0 assert result["as_of"] == "2026-07-28" def test_akb_csv_preserves_rows_and_the_final_balance() -> None: lines = ["Buchung;Valuta;Buchungstext;Belastung;Gutschrift;Saldo"] for index in range(12): tx_date = "04.08.2025" if index == 0 else ("28.07.2026" if index == 11 else "01.01.2026") lines.append(f"{tx_date};{tx_date};Synthetic booking {index};1.00;;{100 + (11 - index):.2f}") parsed = parse_budget_csv_text("\n".join(lines) + "\n", "akb_bank") assert parsed["physical_row_count"] == parsed["logical_row_count"] == 12 assert parsed["rows"][0]["Buchung"] == "04.08.2025" assert parsed["rows"][-1]["Buchung"] == "28.07.2026" assert parsed["rows"][-1]["Saldo"] == "100.00" def test_visa_summary_uses_content_dates_unique_ids_and_payment_aggregate() -> None: rows = [] for index in range(12): rows.append({"TransactionId": f"T{index}", "Date": "2024-07-11" if index == 0 else ("2026-07-13" if index == 11 else "2025-01-01"), "Amount": "100.00" if index == 0 else ("200.00" if index == 1 else "-1.00"), "Currency": "CHF", "MerchantName": "IHRE ZAHLUNG – DANKE" if index < 2 else "Synthetic"}) summary = summarize_visa_rows(rows) assert summary == {"booked_transactions": 12, "period_start": "2024-07-11", "period_end": "2026-07-13", "unique_transaction_ids": 12, "payment_count": 2, "payment_total_chf": "300.00"} def test_migros_summary_preserves_duplicate_items_and_has_no_amount_review_rule() -> None: rows = [] totals = [40.00, 60.00, 80.00] for receipt in range(3): item_count = 2 for item in range(item_count): rows.append({"Datum": "2026-01-02" if receipt == 0 else ("2026-07-27" if receipt == 2 else "2026-04-01"), "Zeit": "12:00", "Filiale": "Synthetic", "Kassennummer": "1", "Transaktionsnummer": str(receipt), "Artikel": "Repeated item", "Umsatz": str(totals[receipt] / item_count)}) assert len(rows) == 6 summary = summarize_migros_rows(rows) assert summary["article_rows"] == 6 and summary["receipt_count"] == 3 assert summary["period_start"] == "2026-01-02" and summary["period_end"] == "2026-07-27" assert summary["median_receipt_chf"] == "60.00" assert summary["receipts_at_least_50"] == 2 assert summary["amount_threshold_review_count"] == 0 assert summary["duplicate_article_rows_removed"] == 0 @pytest.mark.parametrize("profile", ["visa_credit_card", "akb_bank", "migros_receipts"]) def test_explicit_profile_rejects_unrelated_headers(profile: str) -> None: with pytest.raises(ValueError, match="unsupported"): parse_budget_csv_text("foo,bar\nvalue,value\n", profile) def test_valid_header_with_no_importable_rows_is_not_confirmable() -> None: conn = database() preview = preview_household_import(conn, { "files": [{ "profile": "visa_credit_card", "csv_text": "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\nT0,SYN-CARD-1,,,CHF,Synthetic,booked\n", }] }) assert preview["confirmable"] is False assert {error["code"] for error in preview["errors"]} == {"source_file_contains_no_importable_rows"} def test_unlinked_migros_receipt_never_auto_creates_second_expense() -> None: conn = database() receipt = "Datum;Zeit;Filiale;Kassennummer;Transaktionsnummer;Artikel;Umsatz\n2026-07-01;12:00;Synthetic;1;1;Item;75.00\n" payload = {"files": [{"profile": "migros_receipts", "csv_text": receipt}]} preview = preview_household_import(conn, payload) assert preview["receipt_links"][0]["status"] == "unmatched" confirm_household_import(conn, confirm_payload(payload, preview)) assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 0 def test_public_preview_uses_human_states_without_reason_codes_or_raw_references() -> None: conn = database() preview = preview_household_import(conn, visa_payload()) serialized = json.dumps(preview, ensure_ascii=False) assert preview["classification_version"] == CLASSIFICATION_VERSION assert preview["items"][0]["user_state"] in {"proposal_ready", "decision_needed", "special_case"} assert "reason_codes" not in preview["items"][0] assert "SYN-CARD-1" not in serialized assert "source_row_fingerprint" not in serialized def test_preview_declares_expected_writes_and_is_database_read_only() -> None: conn = database() payload = visa_payload() before = "\n".join(conn.iterdump()) result = preview_household_import(conn, payload) assert "expected_writes" in result and "expected_budget_effect" in result assert "\n".join(conn.iterdump()) == before def test_review_batch_fails_closed_after_mapping_account_role_drift() -> None: conn = database() payload = review_ready_visa_payload() import_preview = preview_household_import(conn, payload) confirm_household_import(conn, confirm_payload(payload, import_preview)) item = get_household_review(conn)["items"][0] selection = {"items": [{"item_token": item["item_token"], "category_id": "cat_food"}]} conn.execute("UPDATE budget_accounts SET account_type='checking' WHERE budget_account_id='bacc_card'") conn.execute("UPDATE accounts SET account_type='cash',portfolio_bucket='cash' WHERE account_id='acct_card'") conn.commit() with pytest.raises(HTTPException) as stale_mapping: preview_household_review_batch(conn, selection) assert stale_mapping.value.status_code == 409 def test_review_inbox_batch_is_fingerprint_bound_stale_safe_and_idempotent() -> None: conn = database() payload = review_ready_visa_payload() import_preview = preview_household_import(conn, payload) confirm_household_import(conn, confirm_payload(payload, import_preview)) inbox = get_household_review(conn) assert inbox["proposal_count"] == 0 and inbox["decision_count"] == 1 item = inbox["items"][0] assert "confidence" not in item and "internal_reason" not in item transaction_count_before = conn.execute( "SELECT COUNT(*) FROM budget_transactions" ).fetchone()[0] with pytest.raises(HTTPException) as wrong_category_type: preview_household_review_batch(conn, { "items": [{ "item_token": item["item_token"], "category_id": "cat_other_income", }] }) assert wrong_category_type.value.status_code == 409 assert conn.execute( "SELECT COUNT(*) FROM budget_transactions" ).fetchone()[0] == transaction_count_before selection = {"items": [{"item_token": item["item_token"], "category_id": "cat_food"}]} before = "\n".join(conn.iterdump()) preview = preview_household_review_batch(conn, selection) assert "\n".join(conn.iterdump()) == before result = confirm_household_review_batch(conn, selection | { "confirm": True, "preview_fingerprint": preview["preview_fingerprint"], "baseline_fingerprint": preview["baseline_fingerprint"], }) assert result["confirmed_count"] == 1 and result["idempotent"] is False assert conn.execute("SELECT category_id FROM budget_transactions WHERE source_type='import_candidate'").fetchone()[0] == "cat_food" second = confirm_household_review_batch(conn, selection | { "confirm": True, "preview_fingerprint": preview["preview_fingerprint"], "baseline_fingerprint": preview["baseline_fingerprint"], }) assert second["idempotent"] is True with pytest.raises(HTTPException) as altered_retry: confirm_household_review_batch(conn, { "items": [{"item_token": item["item_token"], "category_id": "cat_mobility"}], "confirm": True, "preview_fingerprint": preview["preview_fingerprint"], "baseline_fingerprint": preview["baseline_fingerprint"], }) assert altered_retry.value.status_code == 409 with pytest.raises(HTTPException) as altered_baseline_retry: confirm_household_review_batch(conn, { "items": [{"item_token": item["item_token"], "category_id": "cat_food"}], "confirm": True, "preview_fingerprint": preview["preview_fingerprint"], "baseline_fingerprint": "different-nonempty-baseline", }) assert altered_baseline_retry.value.status_code == 409 stale_conn = database() stale_import = review_ready_visa_payload(description="Stale preview shop") stale_import_preview = preview_household_import(stale_conn, stale_import) confirm_household_import(stale_conn, confirm_payload(stale_import, stale_import_preview)) stale_item = get_household_review(stale_conn)["items"][0] stale_selection = {"items": [{"item_token": stale_item["item_token"], "category_id": "cat_food"}]} stale_preview = preview_household_review_batch(stale_conn, stale_selection) stale_conn.execute("UPDATE budget_categories SET updated_at='changed' WHERE category_id='cat_food'") stale_conn.commit() with pytest.raises(HTTPException) as exc: confirm_household_review_batch(stale_conn, stale_selection | { "confirm": True, "preview_fingerprint": stale_preview["preview_fingerprint"], "baseline_fingerprint": stale_preview["baseline_fingerprint"], }) assert exc.value.status_code == 409 assert stale_conn.execute( "SELECT COUNT(*) FROM budget_transactions WHERE description='Stale preview shop'" ).fetchone()[0] == 0 def test_productive_canonical_categories_win_over_legacy_aliases() -> None: conn = database() conn.execute( "INSERT INTO budget_categories(category_id,name,category_type,is_active,sort_order,created_at) " "VALUES ('legacy_food','Essen + Haushalt','expense',1,99,'2026-01-01')" ) conn.commit() before = conn.execute("SELECT COUNT(*) FROM budget_categories").fetchone()[0] food = classify_household_row(conn, row(merchant="Migros Filiale 4711")) auto = classify_household_row(conn, row(merchant="Parkhaus Terminal ABC123")) assert (food["category_id"], food["category_name"]) == ("cat_food", "Essen & Haushalt") assert (auto["category_id"], auto["category_name"]) == ("cat_mobility", "Auto Melanie") assert conn.execute("SELECT COUNT(*) FROM budget_categories").fetchone()[0] == before @pytest.mark.parametrize("amount", ["10.00", "40.00", "60.00"]) def test_migros_receipt_matching_has_no_amount_threshold(amount: str) -> None: payload = {"files": [ {"profile": "visa_credit_card", "csv_text": ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" f"M-{amount},SYN-CARD-1,2026-07-01,{amount},CHF,Migros,booked\n" )}, {"profile": "migros_receipts", "csv_text": ( "Datum;Zeit;Filiale;Kassennummer;Transaktionsnummer;Artikel;Umsatz\n" f"2026-07-01;12:00;Synthetic;1;R-{amount};Item;{amount}\n" )}, ]} preview = preview_household_import(database(), payload) assert preview["receipt_links"][0]["status"] == "linked" def test_migros_without_counterpart_never_creates_an_expense_and_ambiguous_match_reviews() -> None: receipt = ( "Datum;Zeit;Filiale;Kassennummer;Transaktionsnummer;Artikel;Umsatz\n" "2026-07-01;12:00;Synthetic;1;R-1;Item;10.00\n" ) conn = database() unmatched_payload = {"files": [{"profile": "migros_receipts", "csv_text": receipt}]} unmatched = preview_household_import(conn, unmatched_payload) assert unmatched["receipt_links"][0]["status"] == "unmatched" confirm_household_import(conn, confirm_payload(unmatched_payload, unmatched)) assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM household_migros_links").fetchone()[0] == 0 ambiguous_payload = {"files": [ {"profile": "visa_credit_card", "csv_text": ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" "A,SYN-CARD-1,2026-07-01,10.00,CHF,Migros,booked\n" "B,SYN-CARD-1,2026-07-01,10.00,CHF,Migros,booked\n" )}, {"profile": "migros_receipts", "csv_text": receipt}, ]} ambiguous = preview_household_import(database(), ambiguous_payload) assert ambiguous["receipt_links"][0]["status"] == "review" pending_conn = database() pending_payload = {"files": [ {"profile": "visa_credit_card", "csv_text": ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" "PENDING,SYN-CARD-1,2026-07-01,10.00,CHF,Migros,pending\n" )}, {"profile": "migros_receipts", "csv_text": receipt}, ]} pending_preview = preview_household_import(pending_conn, pending_payload) assert pending_preview["receipt_links"][0]["status"] == "unmatched" confirm_household_import( pending_conn, confirm_payload(pending_payload, pending_preview) ) assert pending_conn.execute( "SELECT COUNT(*) FROM household_migros_links WHERE status='linked'" ).fetchone()[0] == 0 assert pending_conn.execute("SELECT COUNT(*) FROM household_migros_links").fetchone()[0] == 0 def test_user_confirmed_unmatched_transfer_is_reachable_neutral_and_stale_safe() -> None: payload = {"files": [{ "profile": "raiffeisen_bank", "csv_text": ( "IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\n" "SYN-BANK-1;2026-07-01;Synthetic own account transfer;-91.00;909.00;2026-07-01\n" ), }]} conn = database() undecided = preview_household_import(conn, payload) assert "user_confirmed_unmatched_transfer" in undecided["items"][0]["decision_actions"] private_row_token = undecided["items"][0]["row_token"] assert private_row_token.startswith("row_") and private_row_token != "row_1" decided_payload = payload | { "user_decisions": {private_row_token: "user_confirmed_unmatched_transfer"} } decided = preview_household_import(conn, decided_payload) assert decided["preview_fingerprint"] != undecided["preview_fingerprint"] item = decided["items"][0] assert item["selected_user_decision"] == "user_confirmed_unmatched_transfer" assert item["explanation"] == ( "Nutzerbestätigter interner Transfer. Die Gegenbuchung ist in den vorhandenen " "Importen noch nicht gefunden." ) assert decided["expected_budget_effect"]["income_chf"] == "0.00" assert decided["expected_budget_effect"]["expense_chf"] == "0.00" assert decided["counts"]["safe_transfer_pairs"] == 0 result = confirm_household_import(conn, confirm_payload(decided_payload, decided)) assert result["idempotent"] is False assert confirm_household_import(conn, confirm_payload(decided_payload, decided))["idempotent"] is True tx = conn.execute("SELECT transaction_type,category_id FROM budget_transactions").fetchone() assert tuple(tx) == ("transfer", None) transfer_overview = get_household_overview(conn, month="2026-07") assert transfer_overview["neutral_transfers_chf"] == "0.00" assert transfer_overview["neutral_transfer_volume_chf"] == "91.00" assert conn.execute("SELECT COUNT(*) FROM budget_transfer_pairs").fetchone()[0] == 0 stale_conn = database() stale = preview_household_import(stale_conn, decided_payload) with pytest.raises(HTTPException) as exc: confirm_household_import(stale_conn, confirm_payload(payload, stale)) assert exc.value.status_code == 409 def test_merchant_normalization_removes_only_volatile_safe_suffixes() -> None: assert normalize_merchant("Coop Filiale 4711 Terminal ABC123") == normalize_merchant( "COOP Store 9988 TID XYZ999" ) == "coop" assert normalize_merchant("Coop Restaurant") != normalize_merchant("Coop Tankstelle") def test_bank_envelope_extracts_counterparty_without_losing_semantic_text() -> None: incoming = ( "Zahlungseingang Ref Nr 123456 Ursprünglicher Auftraggeber " "Synthetic Employer AG Mitteilung Salary July" ) outgoing = "Belastung E-Banking Synthetic Utility AG Mitteilung Invoice 4711" outgoing_with_owner = "Belastung E-Banking Synthetic Retailer AG Mitteilung Invoice Ursprünglicher Auftraggeber Synthetic Account Owner" salary_credit = "Gutschrift Synthetic Employer AG Lohnzahlung CHF 9000.00" assert extract_counterparty(incoming, "akb_bank") == "synthetic employer ag" assert extract_counterparty(outgoing, "raiffeisen_bank") == "synthetic utility ag" assert extract_counterparty(outgoing_with_owner, "akb_bank") == "synthetic retailer ag" assert extract_counterparty(salary_credit, "raiffeisen_bank") == "synthetic employer ag" assert "ref nr" not in extract_counterparty(incoming, "akb_bank") def test_payment_processor_is_unwrapped_only_with_a_real_counterparty() -> None: assert extract_counterparty("PAYPAL * MIGROS STORE 4711", "visa_credit_card") == "migros" assert extract_counterparty("PAYPAL", "visa_credit_card") == "paypal" assert extract_counterparty("TWINT Terminal 998877", "akb_bank") == "twint" def test_income_patterns_are_explicit_and_transfer_safe_without_fallback_category() -> None: conn = database() salary = classify_household_row( conn, row( merchant="Zahlungseingang Ursprünglicher Auftraggeber Synthetic Employer AG Mitteilung Lohn Juli", source="akb_bank", amount="1000.00", ), ) assert salary["income_kind"] == "salary" assert salary["user_state"] == "decision_needed" assert salary["category_id"] is None assert salary["origin"] == "clear_income_pattern" payment = classify_household_row( conn, row( merchant="Zahlungseingang Ursprünglicher Auftraggeber Synthetic Person Mitteilung Danke", source="akb_bank", amount="25.00", ), ) assert payment["income_kind"] == "payment_incoming" assert payment["user_state"] == "decision_needed" assert payment["budget_effect_chf"] is None possible_transfer = classify_household_row( conn, row( merchant="Zahlungseingang Own account transfer", source="akb_bank", amount="250.00", ), ) assert possible_transfer["income_kind"] == "possible_internal_transfer" assert possible_transfer["user_state"] == "decision_needed" assert possible_transfer["category_id"] is None salary_named_transfer = classify_household_row( conn, row( merchant="Zahlungseingang Übertrag von Lohnkonto", source="akb_bank", amount="250.00", ), ) assert salary_named_transfer["income_kind"] == "possible_internal_transfer" assert salary_named_transfer["category_id"] is None def test_bank_counterparty_envelopes_form_one_finite_decision_group_without_raw_references() -> None: conn = database() rows = [ "2026-06-25;2026-06-25;Zahlungseingang Ursprünglicher Auftraggeber Synthetic Employer AG Mitteilung Lohn June Ref 111;;1000.00", "2026-07-25;2026-07-25;Zahlungseingang Ursprünglicher Auftraggeber Synthetic Employer AG Mitteilung Lohn July Ref 222;;1000.00", ] payload = { "files": [{ "profile": "akb_bank", "source_reference": "SYN-BANK-2", "csv_text": "Buchung;Valuta;Buchungstext;Belastung;Gutschrift\n" + "\n".join(rows) + "\n", }] } preview = preview_household_import(conn, payload) clusters = [cluster for cluster in preview["merchant_clusters"] if cluster["count"] == 2] assert len(clusters) == 1 assert clusters[0]["transaction_semantics"] == "income" assert clusters[0]["count"] == 2 assert len(clusters[0]["members"]) == 2 serialized = json.dumps(preview, ensure_ascii=False).casefold() assert "ref 111" not in serialized assert "ref 222" not in serialized assert "synthetic employer ag" in serialized def test_active_recurring_pattern_requires_exact_merchant_amount_and_cadence() -> None: conn = database() created = confirm_manual_recurring_payment(conn, { "name": "Synthetic Recurring Vendor", "merchant_name": "Synthetic Recurring Vendor", "category_id": "cat_food", "account_id": "bacc_card", "expected_amount_text": "20.00", "currency": "CHF", "frequency": "monthly", "expected_day_of_month": 15, "date_tolerance_days": 2, "recurring_type": "subscription", }) matched = classify_household_row(conn, { "source_type": "visa_credit_card", "merchant": "Synthetic Recurring Vendor", "description": "Synthetic Recurring Vendor", "amount": "20.50", "signed_amount": "-20.50", "currency": "CHF", "transaction_date": "2026-07-16", "classification": "charge", "disposition": "booked", "mapping": {"budget_account_id": "bacc_card"}, }) wrong_amount = classify_household_row(conn, { "source_type": "visa_credit_card", "merchant": "Synthetic Recurring Vendor", "description": "Synthetic Recurring Vendor", "amount": "30.00", "signed_amount": "-30.00", "currency": "CHF", "transaction_date": "2026-07-16", "classification": "charge", "disposition": "booked", "mapping": {"budget_account_id": "bacc_card"}, }) wrong_account = classify_household_row(conn, { "source_type": "visa_credit_card", "merchant": "Synthetic Recurring Vendor", "description": "Synthetic Recurring Vendor", "amount": "20.50", "signed_amount": "-20.50", "currency": "CHF", "transaction_date": "2026-07-16", "classification": "charge", "disposition": "booked", "mapping": {"budget_account_id": "bacc_bank"}, }) wrong_currency = classify_household_row(conn, { "source_type": "visa_credit_card", "merchant": "Synthetic Recurring Vendor", "description": "Synthetic Recurring Vendor", "amount": "20.50", "signed_amount": "-20.50", "currency": "EUR", "transaction_date": "2026-07-16", "classification": "charge", "disposition": "booked", "mapping": {"budget_account_id": "bacc_card"}, }) conn.execute("UPDATE budget_recurring_payments SET account_id=NULL WHERE recurring_id=?", (created["entity_id"],)) accountless = classify_household_row(conn, { "source_type": "visa_credit_card", "merchant": "Synthetic Recurring Vendor", "description": "Synthetic Recurring Vendor", "amount": "20.50", "signed_amount": "-20.50", "currency": "CHF", "transaction_date": "2026-07-16", "classification": "charge", "disposition": "booked", "mapping": {"budget_account_id": "bacc_card"}, }) assert created["status"] == "confirmed" assert matched["origin"] == "active_recurring_pattern" assert matched["category_id"] == "cat_food" assert wrong_amount["user_state"] == "decision_needed" assert wrong_account["user_state"] == "decision_needed" assert wrong_currency["user_state"] == "decision_needed" assert accountless["user_state"] == "decision_needed" def test_ambiguous_marketplace_stays_grouped_review_without_fallback_category() -> None: csv_text = ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" "A,SYN-CARD-1,2026-07-01,10.00,CHF,Amazon Marketplace,booked\n" "B,SYN-CARD-1,2026-07-02,20.00,CHF,Amazon Marketplace,booked\n" ) preview = preview_household_import(database(), {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]}) assert all(item["category_id"] is None for item in preview["items"]) cluster = next(item for item in preview["merchant_clusters"] if item["count"] == 2) assert cluster["transaction_semantics"] == "expense" assert cluster["category_decision_allowed"] is True assert cluster["merchant_family"] == "amazon marketplace" assert preview["review_threshold"]["merchant_review_cluster_count"] == 1 def test_generic_twint_without_counterparty_cannot_receive_a_bulk_category() -> None: csv_text = ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" "A,SYN-CARD-1,2026-07-01,10.00,CHF,TWINT Terminal 998877,booked\n" "B,SYN-CARD-1,2026-07-02,20.00,CHF,TWINT Terminal 887766,booked\n" ) preview = preview_household_import(database(), {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]}) cluster = next(item for item in preview["merchant_clusters"] if item["count"] == 2) assert cluster["merchant_family"] == "twint" assert cluster["category_decision_allowed"] is False assert preview["review_threshold"]["individual_review_count"] == 2 assert preview["review_threshold"]["merchant_review_cluster_count"] == 0 def test_cluster_exposes_every_member_and_honours_an_explicit_exclusion() -> None: csv_text = ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" "A,SYN-CARD-1,2026-07-01,10.00,CHF,Synthetic Cluster Shop,booked\n" "B,SYN-CARD-1,2026-07-02,20.00,CHF,Synthetic Cluster Shop,booked\n" "C,SYN-CARD-1,2026-07-03,30.00,CHF,Synthetic Cluster Shop,booked\n" "D,SYN-CARD-1,2026-07-04,40.00,CHF,Synthetic Cluster Shop,booked\n" "E,SYN-CARD-1,2026-07-05,50.00,CHF,Synthetic Cluster Shop,booked\n" ) payload = {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]} first = preview_household_import(database(), payload) cluster = next(item for item in first["merchant_clusters"] if item["count"] == 5) assert len(cluster["examples"]) == 3 assert len(cluster["members"]) == 5 assert [member["row_token"] for member in cluster["members"]] == cluster["row_tokens"] decision_payload = payload | {"cluster_decisions": [{ "cluster_token": cluster["cluster_token"], "category_id": "cat_food", "excluded_row_tokens": [cluster["row_tokens"][1]], }]} second = preview_household_import(database(), decision_payload) assert second["preview_fingerprint"] != first["preview_fingerprint"] by_token = {item["row_token"]: item for item in second["items"]} assert by_token[cluster["row_tokens"][0]]["category_id"] == "cat_food" assert by_token[cluster["row_tokens"][2]]["category_id"] == "cat_food" assert by_token[cluster["row_tokens"][3]]["category_id"] == "cat_food" assert by_token[cluster["row_tokens"][4]]["category_id"] == "cat_food" assert by_token[cluster["row_tokens"][1]]["category_id"] is None selected_cluster = next(item for item in second["merchant_clusters"] if item["cluster_token"] == cluster["cluster_token"]) assert (selected_cluster["selected_count"], selected_cluster["excluded_count"]) == (4, 1) def test_technical_confirmability_and_business_readiness_are_separate_and_read_only() -> None: conn = database() payload = visa_payload(description="Unknown Individual Merchant") before = "\n".join(conn.iterdump()) preview = preview_household_import(conn, payload) assert preview["technically_confirmable"] is True assert preview["confirmable"] is True assert preview["business_ready_for_confirm"] is False assert preview["review_threshold"]["individual_review_count"] == 1 assert "\n".join(conn.iterdump()) == before with pytest.raises(HTTPException) as blocked: confirm_household_import(conn, confirm_payload(payload, preview)) assert blocked.value.status_code == 409 assert "\n".join(conn.iterdump()) == before ready = preview_household_import(database(), visa_payload(description="Migros Filiale 4711")) assert ready["technically_confirmable"] is True assert ready["business_ready_for_confirm"] is True def test_concept_v2_90_10_gate_is_authoritative_while_strict_indicator_remains_visible() -> None: rows = ["TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType"] for index in range(1, 10): rows.append( f"KNOWN-{index},SYN-CARD-1,2026-07-{index:02d},10.00,CHF,Migros Filiale 4711,booked" ) rows.append("UNKNOWN,SYN-CARD-1,2026-07-10,10.00,CHF,Unknown Individual Merchant,booked") conn = database() payload = {"files": [{"profile": "visa_credit_card", "csv_text": "\n".join(rows) + "\n"}]} preview = preview_household_import(conn, payload) assert preview["concept_v2_readiness"] == { "ordinary_monetary_count": 10, "plausible_proposal_count": 9, "plausible_proposal_ratio": "0.9000", "minimum_plausible_proposal_ratio": "0.9000", "manual_decision_count": 1, "manual_decision_ratio": "0.1000", "maximum_manual_decision_ratio": "0.1000", "coverage_target_met": True, "manual_decision_target_met": True, "gate_met": True, } assert preview["strict_quality_indicator"]["coverage_target_met"] is False assert preview["strict_quality_indicator"]["indicator_met"] is False assert preview["technically_confirmable"] is True assert preview["business_ready_for_confirm"] is True assert confirm_household_import(conn, confirm_payload(payload, preview))["idempotent"] is False def test_concept_v2_gate_blocks_below_90_percent_coverage() -> None: rows = ["TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType"] for index in range(1, 9): rows.append( f"KNOWN-{index},SYN-CARD-1,2026-07-{index:02d},10.00,CHF,Migros Filiale 4711,booked" ) rows.extend([ "UNKNOWN-A,SYN-CARD-1,2026-07-09,10.00,CHF,Unknown Merchant A,booked", "UNKNOWN-B,SYN-CARD-1,2026-07-10,10.00,CHF,Unknown Merchant B,booked", ]) preview = preview_household_import( database(), {"files": [{"profile": "visa_credit_card", "csv_text": "\n".join(rows) + "\n"}]}, ) assert preview["concept_v2_readiness"]["plausible_proposal_ratio"] == "0.8000" assert preview["concept_v2_readiness"]["manual_decision_ratio"] == "0.2000" assert preview["concept_v2_readiness"]["gate_met"] is False assert preview["business_ready_for_confirm"] is False def test_exact_owner_confirmed_bank_cluster_can_be_bound_as_budget_neutral( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("JARVIS_FINANCE_OPERATOR_APPROVAL_KEY", "synthetic-operator-key") csv_text = ( "IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\n" "SYN-BANK-1;2026-07-01;Household Counterparty;-20.00;980.00;2026-07-01\n" "SYN-BANK-1;2026-07-02;Household Counterparty;-30.00;950.00;2026-07-02\n" ) payload = {"files": [{"profile": "raiffeisen_bank", "csv_text": csv_text}]} conn = database() first = preview_household_import(conn, payload) cluster = next(item for item in first["merchant_clusters"] if item["count"] == 2) decision = authorize_owner_neutral_cluster(conn, { "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"], }, "synthetic-operator-key") second = preview_household_import(conn, payload | {"cluster_decisions": [decision]}) selected = [item for item in second["items"] if item["row_token"] in cluster["row_tokens"]] assert second["preview_fingerprint"] != first["preview_fingerprint"] assert {item["transaction_semantics"] for item in selected} == {"user_confirmed_unmatched_transfer"} assert all(item["selected_user_decision"] == "user_confirmed_unmatched_transfer" for item in selected) assert second["expected_budget_effect"]["neutral_transfer_chf"] == "50.00" assert second["expected_budget_effect"]["expense_chf"] == "0.00" assert second["expected_budget_effect"]["income_chf"] == "0.00" applied = next(item for item in second["merchant_clusters"] if item["cluster_token"] == cluster["cluster_token"]) assert applied["decision_type"] == "user_confirmed_unmatched_transfer" assert applied["selected_count"] == 2 def test_neutral_cluster_requires_server_bound_owner_approval() -> None: csv_text = ( "IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\n" "SYN-BANK-1;2026-07-01;Synthetic Counterparty;-20.00;980.00;2026-07-01\n" "SYN-BANK-1;2026-07-02;Synthetic Counterparty;-30.00;950.00;2026-07-02\n" ) payload = {"files": [{"profile": "raiffeisen_bank", "csv_text": csv_text}]} first = preview_household_import(database(), payload) cluster = next(item for item in first["merchant_clusters"] if item["count"] == 2) with pytest.raises(HTTPException) as blocked: preview_household_import(database(), payload | {"cluster_decisions": [{ "cluster_token": cluster["cluster_token"], "decision_type": "user_confirmed_unmatched_transfer", "owner_confirmed": True, "approval_evidence": "owner_attested_known_household_counterparty_v1", "approved_row_tokens": cluster["row_tokens"], "approval_token": "forged", "excluded_row_tokens": [], }]}) assert blocked.value.status_code == 422 def test_owner_attested_neutral_cluster_approval_is_baseline_bound_and_expires( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("JARVIS_FINANCE_OPERATOR_APPROVAL_KEY", "synthetic-operator-key") monkeypatch.setattr(household_import_service.time, "time", lambda: 1_000_000) csv_text = ( "IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\n" "SYN-BANK-1;2026-07-01;Synthetic Household Counterparty;-20.00;980.00;2026-07-01\n" "SYN-BANK-1;2026-07-02;Synthetic Household Counterparty;-30.00;950.00;2026-07-02\n" ) payload = {"files": [{"profile": "raiffeisen_bank", "csv_text": csv_text}]} stale_conn = database() stale_base = preview_household_import(stale_conn, payload) stale_cluster = next(item for item in stale_base["merchant_clusters"] if item["count"] == 2) stale_decision = authorize_owner_neutral_cluster(stale_conn, { "confirm_owner_attestation": True, "preview_request": payload, "approval_evidence": "owner_attested_known_household_counterparty_v1", "cluster_token": stale_cluster["cluster_token"], "approved_row_tokens": stale_cluster["row_tokens"], }, "synthetic-operator-key") stale_conn.execute("UPDATE budget_categories SET updated_at='baseline-changed' WHERE category_id='cat_food'") with pytest.raises(HTTPException, match="approval is stale"): preview_household_import(stale_conn, payload | {"cluster_decisions": [stale_decision]}) expired_conn = database() expired_base = preview_household_import(expired_conn, payload) expired_cluster = next(item for item in expired_base["merchant_clusters"] if item["count"] == 2) expired_decision = authorize_owner_neutral_cluster(expired_conn, { "confirm_owner_attestation": True, "preview_request": payload, "approval_evidence": "owner_attested_known_household_counterparty_v1", "cluster_token": expired_cluster["cluster_token"], "approved_row_tokens": expired_cluster["row_tokens"], }, "synthetic-operator-key") monkeypatch.setattr(household_import_service.time, "time", lambda: 1_000_901) with pytest.raises(HTTPException, match="approval has expired"): preview_household_import(expired_conn, payload | {"cluster_decisions": [expired_decision]}) def test_safe_transfer_pairs_do_not_dilute_ordinary_business_readiness() -> None: raiffeisen_rows = ["IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date"] akb_rows = ["Buchung;Valuta;Buchungstext;Belastung;Gutschrift"] for index in range(1, 21): raiffeisen_rows.append( f"SYN-BANK-1;2026-07-{index:02d};Transfer eigenes Konto;-{index}.00;900.00;2026-07-{index:02d}" ) akb_rows.append( f"2026-07-{index:02d};2026-07-{index:02d};Transfer eigenes Konto;;{index}.00" ) payload = {"files": [ {"profile": "raiffeisen_bank", "csv_text": "\n".join(raiffeisen_rows) + "\n"}, {"profile": "akb_bank", "source_reference": "SYN-BANK-2", "csv_text": "\n".join(akb_rows) + "\n"}, {"profile": "visa_credit_card", "csv_text": ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" "UNKNOWN,SYN-CARD-1,2026-07-28,99.00,CHF,Unknown Individual Merchant,booked\n" )}, ]} preview = preview_household_import(database(), payload) assert preview["counts"]["safe_transfer_pairs"] == 20 assert preview["review_threshold"]["ordinary_monetary_count"] == 1 assert preview["review_threshold"]["covered_count"] == 0 assert preview["review_threshold"]["coverage_ratio"] == "0.0000" assert preview["business_ready_for_confirm"] is False def test_expense_rows_and_clusters_reject_active_income_categories() -> None: conn = database() csv_text = ( "TransactionId,CardId,Date,Amount,Currency,MerchantName,StateType\n" "TYPE-A,SYN-CARD-1,2026-07-01,10.00,CHF,Type Safe Cluster,booked\n" "TYPE-B,SYN-CARD-1,2026-07-02,20.00,CHF,Type Safe Cluster,booked\n" ) payload = {"files": [{"profile": "visa_credit_card", "csv_text": csv_text}]} initial = preview_household_import(conn, payload) with pytest.raises(HTTPException) as row_mismatch: preview_household_import(conn, payload | { "category_overrides": { initial["items"][0]["row_token"]: "cat_other_income" } }) assert row_mismatch.value.status_code == 422 cluster = next(item for item in initial["merchant_clusters"] if item["count"] == 2) with pytest.raises(HTTPException) as cluster_mismatch: preview_household_import(conn, payload | {"cluster_decisions": [{ "cluster_token": cluster["cluster_token"], "category_id": "cat_other_income", "excluded_row_tokens": [], }]}) assert cluster_mismatch.value.status_code == 422 def test_unmatched_transfer_decision_requires_server_derived_transfer_evidence() -> None: conn = database() ordinary = { "files": [{ "profile": "raiffeisen_bank", "csv_text": ( "IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\n" "SYN-BANK-1;2026-07-01;Zahlung an Opaque Person;-19.00;900.00;2026-07-01\n" "SYN-BANK-1;2026-07-02;Internal transfer fee;-5.00;895.00;2026-07-02\n" "SYN-BANK-1;2026-07-03;Wallet top up fee;-6.00;889.00;2026-07-03\n" "SYN-BANK-1;2026-07-04;Own account transfer fee;-7.00;882.00;2026-07-04\n" ), }] } initial = preview_household_import(conn, ordinary) assert len(initial["items"]) == 4 assert all(item["decision_actions"] == [] for item in initial["items"]) token = initial["items"][0]["row_token"] with pytest.raises(HTTPException) as blocked: preview_household_import(conn, ordinary | { "user_decisions": {token: "user_confirmed_unmatched_transfer"} }) assert blocked.value.status_code == 422 @pytest.mark.parametrize( ("merchant", "category_name"), [ ("C&A Mode GmbH", "Kleider und Schuhe"), ("JustEat Zürich", "Essen & Haushalt"), ("PUREGYM AG", "Hobbys & Selfcare"), ("DR MED DENT Beispielpraxis", "Gesundheit"), ("2CO COMOBITDEFENDER", "Elektronische Medien"), ("Restaurants und Verpflegung", "Essen & Haushalt"), ("Tanken, Parkieren und Fahrzeug", "Auto Melanie"), ], ) def test_compact_public_merchant_markers_resolve_without_private_names( merchant: str, category_name: str ) -> None: conn = database() if not conn.execute( "SELECT 1 FROM budget_categories WHERE name=? AND category_type='expense'", (category_name,), ).fetchone(): conn.execute( "INSERT INTO budget_categories(category_id,name,category_type,is_active,sort_order,created_at) " "VALUES (?,?, 'expense',1,99,'2026-01-01')", (f"cat_compact_{abs(hash(category_name))}", category_name), ) result = classify_household_row( conn, { "merchant": merchant, "description": merchant, "signed_amount": "-10.00", "source_type": "visa_credit_card", "transaction_date": "2026-06-01", "currency": "CHF", }, ) assert result["category_name"] == category_name assert result["user_state"] == "proposal_ready" @pytest.mark.parametrize( "merchant", ["Hospitality Supplies AG", "Sparkingshop GmbH", "Unfitness Coaching"], ) def test_public_merchant_markers_do_not_match_unrelated_substrings(merchant: str) -> None: result = classify_household_row( database(), { "merchant": merchant, "description": merchant, "signed_amount": "-10.00", "source_type": "visa_credit_card", "transaction_date": "2026-06-01", "currency": "CHF", }, ) assert result["category_id"] is None assert result["user_state"] == "decision_needed" def test_category_name_duplicates_preserve_uncached_interest_resolution_with_lookup_cache() -> None: conn = database() conn.execute( "DELETE FROM budget_categories WHERE lower(name) IN ('zinsertrag','kapitalertrag')" ) conn.execute( "INSERT INTO budget_categories(category_id,name,category_type,is_active,sort_order,created_at) " "VALUES ('cat_duplicate_interest','Sonstige Einnahmen','income',1,99,'2026-01-01')" ) sample = row( merchant="Zinsgutschrift", source="raiffeisen_bank", amount="10.00", ) uncached = classify_household_row(conn, sample) cached = classify_household_row(conn, sample, ClassificationLookupCache(conn)) assert cached == uncached assert cached["category_id"] == "cat_other_income" assert cached["user_state"] == "proposal_ready" def test_request_local_lookup_cache_preserves_classification_and_bulk_loads_history_once() -> None: conn = database() confirmed_history(conn, merchant="Known Merchant", category="cat_food") samples = [ row(merchant="Known Merchant"), row(merchant="Migros"), row(merchant="Unknown Merchant"), row(merchant="Salary payment", source="raiffeisen_bank", amount="100.00"), ] expected = [classify_household_row(conn, item) for item in samples] statements: list[str] = [] conn.set_trace_callback(statements.append) cache = ClassificationLookupCache(conn) actual = [classify_household_row(conn, item, cache) for item in samples * 5] conn.set_trace_callback(None) assert actual == expected * 5 history_queries = [ statement for statement in statements if "FROM budget_transactions t" in statement and "JOIN budget_transaction_candidates" in statement ] assert len(history_queries) == 1 assert sum("FROM budget_categories WHERE is_active=1" in statement for statement in statements) == 1