spb/air Public MIT
AIR — The Language of Accounting.
Python 100%
1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : test_reconcile.py6# Description : Tests — camt.053/MT940/CSV parsers and bank reconciliation matching.7# =============================================================================8"""Bank reconciliation tests (Phase 6).910Fixtures come from docs/research/bank-statement-formats.md (balance-consistent11statement: 25,000.00 - 1,150.00 + 3,449.93 = 27,299.93). The camt.053 and12MT940 fixtures describe the SAME statement, so parsing either must yield the13same transactions."""14from __future__ import annotations1516from datetime import date17from decimal import Decimal18from pathlib import Path1920import pytest2122from aic.compiler import compile_document23from alsl.loader import load_policy_set24from core.events import AirDocument25from kernel.bank_formats import (26 BankFormatError,27 parse_bank_csv,28 parse_camt053,29 parse_mt940,30 parse_statement,31)32from kernel.ledger import Ledger33from kernel.reconcile import cash_movements, reconcile34from kernel.workspace import Workspace35from sdk.cli import main36from sdk.syscalls import AirKernel3738ROOT = Path(__file__).resolve().parents[1]39POLICY = str(ROOT / "alsl" / "policies" / "ca-qc-2026.yaml")40FIXTURES = ROOT / "tests" / "fixtures"41CAMT = FIXTURES / "bank_statement.xml"42MT940 = FIXTURES / "bank_statement.mt940"43CSV = FIXTURES / "bank_statement.csv"444546# --- parsers ----------------------------------------------------------------------47def test_camt053_parses_signs_dates_and_references() -> None:48 out_txn, in_txn = parse_camt053(CAMT)49 assert out_txn.amount == Decimal("-1150.00") # DBIT -> money out50 assert out_txn.currency == "CAD"51 assert out_txn.date == date(2026, 8, 1)52 assert out_txn.reference == "INV-2026-0042" # EndToEndId wins53 assert "Acme" in out_txn.description # AddtlNtryInf54 assert in_txn.amount == Decimal("3449.93") # CRDT -> money in55 assert in_txn.date == date(2026, 8, 4)565758def test_mt940_parses_the_same_statement_identically() -> None:59 from_mt940 = parse_mt940(MT940)60 from_camt = parse_camt053(CAMT)61 assert [(t.date, t.amount, t.currency, t.reference) for t in from_mt940] == \62 [(t.date, t.amount, t.currency, t.reference) for t in from_camt]63 # :86: free text became the description64 assert "ACME" in from_mt940[0].description656667def test_mt940_balance_integrity_is_enforced(tmp_path: Path) -> None:68 tampered = MT940.read_text().replace("27299,93", "27300,00")69 bad = tmp_path / "bad.mt940"70 bad.write_text(tampered)71 with pytest.raises(BankFormatError, match="does not balance"):72 parse_mt940(bad)737475def test_csv_parser_and_dispatcher_sniffing(tmp_path: Path) -> None:76 txns = parse_bank_csv(CSV)77 assert len(txns) == 3 and txns[2].amount == Decimal("-42.00")78 # dispatcher: unknown suffix, content sniffed79 renamed = tmp_path / "statement.txt"80 renamed.write_text(MT940.read_text())81 assert len(parse_statement(renamed)) == 282 with pytest.raises(BankFormatError, match="unrecognized"):83 empty = tmp_path / "noise.txt"84 empty.write_text("hello world")85 parse_statement(empty)868788# --- matching ----------------------------------------------------------------------89def _books() -> Ledger:90 """Books whose cash activity mirrors the fixture statement, plus one91 ledger-only movement the bank never saw."""92 document = AirDocument.model_validate({"events": [93 {"id": "evt_rc_out", "type": "PaymentSent", "date": date(2026, 8, 2),94 "description": "pay Acme invoice",95 "amount": {"amount": "1150.00", "currency": "CAD"}},96 {"id": "evt_rc_in", "type": "PaymentReceived", "date": date(2026, 8, 4),97 "description": "customer settles sale 7781",98 "amount": {"amount": "3449.93", "currency": "CAD"}},99 {"id": "evt_rc_ghost", "type": "PaymentReceived", "date": date(2026, 8, 3),100 "description": "cheque recorded, not yet deposited",101 "amount": {"amount": "500.00", "currency": "CAD"}},102 ]})103 journal, _ = compile_document(document, load_policy_set(POLICY))104 ledger = Ledger()105 ledger.post_journal(journal, "rc-books")106 return ledger107108109def test_reconcile_matches_within_date_tolerance() -> None:110 result = reconcile(parse_camt053(CAMT), cash_movements(_books()))111 assert result.summary() == {"matched": 2, "unmatched_bank": 0,112 "unmatched_ledger": 1}113 # the 1150 bank debit (Aug 1) matched the books' payment dated Aug 2114 (pair,) = [(t, m) for t, m in result.matched if t.amount < 0]115 assert pair[1].entry_id == "je_evt_rc_out"116 # the undeposited cheque is flagged, not dropped117 assert result.unmatched_ledger[0].entry_id == "je_evt_rc_ghost"118 assert "DIFFERENCES FOUND" in result.render()119120121def test_reconcile_zero_tolerance_refuses_date_drift() -> None:122 result = reconcile(parse_camt053(CAMT), cash_movements(_books()),123 tolerance_days=0)124 assert result.summary()["matched"] == 1 # only the same-day one125126127def test_reconcile_csv_flags_bank_only_fee() -> None:128 result = reconcile(parse_bank_csv(CSV), cash_movements(_books()))129 fees = [t for t in result.unmatched_bank if "FEE" in t.description]130 assert len(fees) == 1 # bank fee not yet booked131132133# --- syscall + CLI ----------------------------------------------------------------------134def _home_with_books(tmp_path: Path) -> Path:135 home = tmp_path / "books"136 Workspace.init(home, policies=POLICY)137 kernel = AirKernel(home, actor="agent:reco")138 kernel.create_economic_event({139 "id": "evt_rc_out", "type": "PaymentSent", "date": "2026-08-02",140 "amount": {"amount": "1150.00", "currency": "CAD"}})141 kernel.create_economic_event({142 "id": "evt_rc_in", "type": "PaymentReceived", "date": "2026-08-04",143 "amount": {"amount": "3449.93", "currency": "CAD"}})144 kernel.post()145 return home146147148def test_reconcile_syscall_is_audited(tmp_path: Path) -> None:149 home = _home_with_books(tmp_path)150 kernel = AirKernel(home, actor="agent:reco")151 summary = kernel.reconcile(MT940)152 assert summary["matched"] == 2 and summary["clean"] is True153 assert kernel.audit.records[-1]["syscall"] == "Reconcile"154 assert kernel.audit.verify_chain()155156157def test_cli_reconcile_exit_codes(tmp_path: Path, capsys) -> None:158 home = _home_with_books(tmp_path)159 # clean statement -> 0160 assert main(["reconcile", str(MT940), "--home", str(home)]) == 0161 assert "CLEAN" in capsys.readouterr().out162 # CSV includes an unbooked bank fee -> differences -> exit 2163 assert main(["reconcile", str(CSV), "--home", str(home)]) == 2164 assert "BANK?" in capsys.readouterr().out165