# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : test_reconcile.py # Description : Tests — camt.053/MT940/CSV parsers and bank reconciliation matching. # ============================================================================= """Bank reconciliation tests (Phase 6). Fixtures come from docs/research/bank-statement-formats.md (balance-consistent statement: 25,000.00 - 1,150.00 + 3,449.93 = 27,299.93). The camt.053 and MT940 fixtures describe the SAME statement, so parsing either must yield the same transactions.""" from __future__ import annotations from datetime import date from decimal import Decimal from pathlib import Path import pytest from aic.compiler import compile_document from alsl.loader import load_policy_set from core.events import AirDocument from kernel.bank_formats import ( BankFormatError, parse_bank_csv, parse_camt053, parse_mt940, parse_statement, ) from kernel.ledger import Ledger from kernel.reconcile import cash_movements, reconcile from kernel.workspace import Workspace from sdk.cli import main from sdk.syscalls import AirKernel ROOT = Path(__file__).resolve().parents[1] POLICY = str(ROOT / "alsl" / "policies" / "ca-qc-2026.yaml") FIXTURES = ROOT / "tests" / "fixtures" CAMT = FIXTURES / "bank_statement.xml" MT940 = FIXTURES / "bank_statement.mt940" CSV = FIXTURES / "bank_statement.csv" # --- parsers ---------------------------------------------------------------------- def test_camt053_parses_signs_dates_and_references() -> None: out_txn, in_txn = parse_camt053(CAMT) assert out_txn.amount == Decimal("-1150.00") # DBIT -> money out assert out_txn.currency == "CAD" assert out_txn.date == date(2026, 8, 1) assert out_txn.reference == "INV-2026-0042" # EndToEndId wins assert "Acme" in out_txn.description # AddtlNtryInf assert in_txn.amount == Decimal("3449.93") # CRDT -> money in assert in_txn.date == date(2026, 8, 4) def test_mt940_parses_the_same_statement_identically() -> None: from_mt940 = parse_mt940(MT940) from_camt = parse_camt053(CAMT) assert [(t.date, t.amount, t.currency, t.reference) for t in from_mt940] == \ [(t.date, t.amount, t.currency, t.reference) for t in from_camt] # :86: free text became the description assert "ACME" in from_mt940[0].description def test_mt940_balance_integrity_is_enforced(tmp_path: Path) -> None: tampered = MT940.read_text().replace("27299,93", "27300,00") bad = tmp_path / "bad.mt940" bad.write_text(tampered) with pytest.raises(BankFormatError, match="does not balance"): parse_mt940(bad) def test_csv_parser_and_dispatcher_sniffing(tmp_path: Path) -> None: txns = parse_bank_csv(CSV) assert len(txns) == 3 and txns[2].amount == Decimal("-42.00") # dispatcher: unknown suffix, content sniffed renamed = tmp_path / "statement.txt" renamed.write_text(MT940.read_text()) assert len(parse_statement(renamed)) == 2 with pytest.raises(BankFormatError, match="unrecognized"): empty = tmp_path / "noise.txt" empty.write_text("hello world") parse_statement(empty) # --- matching ---------------------------------------------------------------------- def _books() -> Ledger: """Books whose cash activity mirrors the fixture statement, plus one ledger-only movement the bank never saw.""" document = AirDocument.model_validate({"events": [ {"id": "evt_rc_out", "type": "PaymentSent", "date": date(2026, 8, 2), "description": "pay Acme invoice", "amount": {"amount": "1150.00", "currency": "CAD"}}, {"id": "evt_rc_in", "type": "PaymentReceived", "date": date(2026, 8, 4), "description": "customer settles sale 7781", "amount": {"amount": "3449.93", "currency": "CAD"}}, {"id": "evt_rc_ghost", "type": "PaymentReceived", "date": date(2026, 8, 3), "description": "cheque recorded, not yet deposited", "amount": {"amount": "500.00", "currency": "CAD"}}, ]}) journal, _ = compile_document(document, load_policy_set(POLICY)) ledger = Ledger() ledger.post_journal(journal, "rc-books") return ledger def test_reconcile_matches_within_date_tolerance() -> None: result = reconcile(parse_camt053(CAMT), cash_movements(_books())) assert result.summary() == {"matched": 2, "unmatched_bank": 0, "unmatched_ledger": 1} # the 1150 bank debit (Aug 1) matched the books' payment dated Aug 2 (pair,) = [(t, m) for t, m in result.matched if t.amount < 0] assert pair[1].entry_id == "je_evt_rc_out" # the undeposited cheque is flagged, not dropped assert result.unmatched_ledger[0].entry_id == "je_evt_rc_ghost" assert "DIFFERENCES FOUND" in result.render() def test_reconcile_zero_tolerance_refuses_date_drift() -> None: result = reconcile(parse_camt053(CAMT), cash_movements(_books()), tolerance_days=0) assert result.summary()["matched"] == 1 # only the same-day one def test_reconcile_csv_flags_bank_only_fee() -> None: result = reconcile(parse_bank_csv(CSV), cash_movements(_books())) fees = [t for t in result.unmatched_bank if "FEE" in t.description] assert len(fees) == 1 # bank fee not yet booked # --- syscall + CLI ---------------------------------------------------------------------- def _home_with_books(tmp_path: Path) -> Path: home = tmp_path / "books" Workspace.init(home, policies=POLICY) kernel = AirKernel(home, actor="agent:reco") kernel.create_economic_event({ "id": "evt_rc_out", "type": "PaymentSent", "date": "2026-08-02", "amount": {"amount": "1150.00", "currency": "CAD"}}) kernel.create_economic_event({ "id": "evt_rc_in", "type": "PaymentReceived", "date": "2026-08-04", "amount": {"amount": "3449.93", "currency": "CAD"}}) kernel.post() return home def test_reconcile_syscall_is_audited(tmp_path: Path) -> None: home = _home_with_books(tmp_path) kernel = AirKernel(home, actor="agent:reco") summary = kernel.reconcile(MT940) assert summary["matched"] == 2 and summary["clean"] is True assert kernel.audit.records[-1]["syscall"] == "Reconcile" assert kernel.audit.verify_chain() def test_cli_reconcile_exit_codes(tmp_path: Path, capsys) -> None: home = _home_with_books(tmp_path) # clean statement -> 0 assert main(["reconcile", str(MT940), "--home", str(home)]) == 0 assert "CLEAN" in capsys.readouterr().out # CSV includes an unbooked bank fee -> differences -> exit 2 assert main(["reconcile", str(CSV), "--home", str(home)]) == 2 assert "BANK?" in capsys.readouterr().out