from __future__ import annotations

import json
from decimal import Decimal
from sqlite3 import Connection

import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient

from jarvis_finance.api.dependencies import get_db
from jarvis_finance.api.main import create_app
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 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)
    assert first == second
    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_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}]}
    preview = preview_household_import(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
    assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 0
    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"]
    matched = preview_household_import(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}]}
    preview = preview_household_import(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_preview = preview_household_import(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}]}
    preview = preview_household_import(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")}]}
    preview = preview_household_import(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"] == "12.50" and overview["expense_chf"] == "0.00"

    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_preview = preview_household_import(conn, purchase)
    confirm_household_import(conn, confirm_payload(purchase, purchase_preview))
    group = next(g for g in get_household_review(conn)["groups"] if g["title"] == "Expense Candidate")
    action = {"group_key": group["group_key"], "action": "expense"}
    action_preview = preview_household_review_action(conn, action)
    confirm_household_review_action(conn, {**action, **action_preview, "confirm": True})
    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,notes FROM budget_transactions WHERE description LIKE 'Storno %'"
    ).fetchone()
    assert result["transaction_type"] == "refund"
    assert Decimal(result["amount_original"]) == Decimal("8.25")
    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
    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"
            ),
        }]
    }
    purchase_preview = preview_household_import(conn, purchases)
    confirm_household_import(conn, confirm_payload(purchases, purchase_preview))
    group = next(g for g in get_household_review(conn)["groups"] if g["title"] == "Expense Candidate")
    action = {"group_key": group["group_key"], "action": "expense"}
    action_preview = preview_household_review_action(conn, action)
    confirm_household_review_action(conn, {**action, **action_preview, "confirm": True})

    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
    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()
    csv_text = "TransactionId,CardId,Date,Amount,Currency,MerchantName\nRV-1,SYN-CARD-001,2026-02-11,-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"
    ).fetchone()
    assert tuple(tx) == ("expense", "-8.25", "confirmed")
    filtered = list_household_transactions(
        conn, period="2026-02", account="Card A", source="Import", 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_candidate_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))
    money_id = conn.execute("SELECT transaction_candidate_id FROM budget_transaction_candidates WHERE source_type='visa_credit_card'").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,status FROM household_migros_links").fetchone()
    assert tuple(link) == (money_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] == 0

    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 tuple(second_link) == (None, None, "unmatched")


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))
    group = next(g for g in get_household_review(conn)["groups"] if g["title"] == "Expense Candidate")
    action = {"group_key": group["group_key"], "action": "expense"}
    action_preview = preview_household_review_action(conn, action)
    confirm_household_review_action(conn, {**action, **action_preview, "confirm": True})
    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 set(transactions) == {"items", "total"}
    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}