# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : test_core.py # Description : Unit tests — money, provenance, ALSL strictness, diagnostics, ledger, backends. # ============================================================================= """Unit tests for the non-golden guarantees: float rejection, SSA provenance, ALSL loader strictness, clang-style diagnostics, the hash-chained ledger, reversal semantics, and the CSV/native backends.""" from __future__ import annotations import json from datetime import date from decimal import Decimal from pathlib import Path import pytest from aic.compiler import compile_document from aic.diagnostics import CompilationError from alsl.loader import AlslLoadError, load_policy_set from backends.generic_csv.backend import GenericCsvBackend from backends.native.backend import NativeLedgerBackend from core.events import AirDocument from core.money import CurrencyMismatchError, Money, RoundingMode, money from kernel.ledger import Ledger from kernel.reporting import balance_sheet, income_statement, trial_balance ROOT = Path(__file__).resolve().parents[1] POLICY_FILE = ROOT / "alsl" / "policies" / "ca-qc-2026.yaml" # --- Money ------------------------------------------------------------------- def test_money_rejects_floats() -> None: with pytest.raises(TypeError): Money(19.99, "CAD") # type: ignore[arg-type] with pytest.raises(TypeError): money("10.00", "CAD").multiply(1.05) # type: ignore[arg-type] def test_money_currency_mismatch() -> None: with pytest.raises(CurrencyMismatchError): money("1", "CAD") + money("1", "USD") def test_rounding_modes_differ_on_ties() -> None: half = money("0.125", "CAD") assert half.quantized(RoundingMode.HALF_UP).amount == Decimal("0.13") assert half.quantized(RoundingMode.HALF_EVEN).amount == Decimal("0.12") def test_air_schema_rejects_float_amounts() -> None: with pytest.raises(Exception): AirDocument.model_validate({ "events": [{ "id": "evt_f", "type": "Sale", "date": date(2026, 1, 1), "amount": {"amount": 19.99, "currency": "CAD"}, }] }) # --- ALSL strictness ----------------------------------------------------------- def test_alsl_rejects_uncited_tax_policy(tmp_path: Path) -> None: bad = tmp_path / "bad.yaml" bad.write_text( "alsl_version: \"0.1\"\npolicy_set: bad\nversion: \"1\"\n" "policies:\n - name: mystery_tax\n kind: tax\n" " when: {jurisdiction: CA-QC}\n" " apply:\n - {code: GST, rate: \"0.05\"}\n", encoding="utf-8", ) with pytest.raises(AlslLoadError, match="cite a source"): load_policy_set(bad) def test_alsl_rejects_float_rates(tmp_path: Path) -> None: bad = tmp_path / "bad.yaml" bad.write_text( "alsl_version: \"0.1\"\npolicy_set: bad\nversion: \"1\"\n" "policies:\n - name: float_tax\n kind: tax\n" " source: docs/research/canada-gst-qst.md\n" " when: {jurisdiction: CA-QC}\n" " apply:\n - {code: GST, rate: 0.05}\n", encoding="utf-8", ) with pytest.raises(AlslLoadError, match="float"): load_policy_set(bad) # --- Diagnostics ----------------------------------------------------------------- def test_unknown_jurisdiction_is_a_precise_error() -> None: policies = load_policy_set(POLICY_FILE) document = AirDocument.model_validate({ "events": [{ "id": "evt_bc", "type": "Sale", "date": date(2026, 1, 1), "amount": {"amount": "100.00", "currency": "CAD"}, "tax": {"jurisdiction": "CA-BC"}, }] }) with pytest.raises(CompilationError) as exc: compile_document(document, policies) codes = [d.code for d in exc.value.diagnostics] assert "AIR-E400" in codes rendered = "\n".join(d.render() for d in exc.value.diagnostics) assert "CA-BC" in rendered and "help:" in rendered def test_missing_fx_rate_is_a_precise_error() -> None: policies = load_policy_set(POLICY_FILE) document = AirDocument.model_validate({ "events": [{ "id": "evt_usd", "type": "Sale", "date": date(2026, 1, 1), "amount": {"amount": "100.00", "currency": "USD"}, "tax": {"jurisdiction": "CA-QC", "exempt": True}, }] }) with pytest.raises(CompilationError) as exc: compile_document(document, policies) assert any(d.code == "AIR-E500" for d in exc.value.diagnostics) # --- helpers --------------------------------------------------------------------- def _demo_journal(): policies = load_policy_set(POLICY_FILE) document = AirDocument.model_validate({ "events": [ {"id": "evt_own", "type": "OwnerContribution", "date": date(2026, 1, 1), "amount": {"amount": "10000.00", "currency": "CAD"}}, {"id": "evt_sale", "type": "Sale", "date": date(2026, 1, 10), "amount": {"amount": "1000.00", "currency": "CAD"}, "tax": {"jurisdiction": "CA-QC"}}, {"id": "evt_buy", "type": "Purchase", "date": date(2026, 1, 12), "amount": {"amount": "400.00", "currency": "CAD"}, "tax": {"jurisdiction": "CA-QC"}}, ] }) journal, _ = compile_document(document, policies) return journal # --- Ledger (native, hash-chained) -------------------------------------------------- def test_ledger_hash_chain_and_idempotency(tmp_path: Path) -> None: journal = _demo_journal() ledger = Ledger(path=tmp_path / "ledger.jsonl") assert ledger.post_journal(journal, "key-1") == 3 assert ledger.post_journal(journal, "key-1") == 0 # idempotent replay assert ledger.verify_chain() # reload from disk: same entries, chain still valid reloaded = Ledger(path=tmp_path / "ledger.jsonl") assert len(reloaded.entries) == 3 assert reloaded.verify_chain() # tampering is detected lines = (tmp_path / "ledger.jsonl").read_text().splitlines() record = json.loads(lines[0]) record["lines"][0]["amount"] = "9999.99" lines[0] = json.dumps(record, sort_keys=True, separators=(",", ":")) (tmp_path / "ledger.jsonl").write_text("\n".join(lines) + "\n") assert not Ledger(path=tmp_path / "ledger.jsonl").verify_chain() def test_ledger_reversal_nets_to_zero() -> None: journal = _demo_journal() ledger = Ledger() ledger.post_journal(journal, "key-1") for entry in list(ledger.entries): ledger.reverse_entry(entry.id, f"rev:{entry.id}") for per_ccy in ledger.balances().values(): for balance in per_ccy.values(): assert balance == 0 # --- Reporting ------------------------------------------------------------------- def test_reports_balance_and_render() -> None: journal = _demo_journal() ledger = Ledger() ledger.post_journal(journal, "key-1") tb = trial_balance(ledger, "csv") total_row = [r for r in tb.splitlines() if r.startswith("TOTAL")][0] cells = total_row.split(",") assert cells[-2] == cells[-1] # debits == credits inc = income_statement(ledger, "json") data = json.loads(inc) net = [r for r in data["rows"] if r["section"] == "NET INCOME"][0] assert Decimal(net["amount"]) == Decimal("600.00") # 1000 revenue - 400 expense bs = balance_sheet(ledger, "markdown") lines = [l for l in bs.splitlines() if "TOTAL" in l] assets = [l for l in lines if "TOTAL ASSETS" in l][0].split("|")[-2].strip() liabeq = [l for l in lines if "TOTAL LIAB." in l][0].split("|")[-2].strip() assert assets == liabeq # Assets = Liabilities + Equity (+ net income) # --- Backends ---------------------------------------------------------------------- def test_csv_backend_post_and_reverse(tmp_path: Path) -> None: journal = _demo_journal() backend = GenericCsvBackend(tmp_path) receipt = backend.post(backend.compile(journal), "abc123") content = Path(receipt.reference).read_text() assert "je_evt_sale" in content and "1149.75" in content reversal = backend.reverse(receipt) rev = Path(reversal.reference).read_text() assert "rev_je_evt_sale" in rev and "REVERSAL:" in rev # sides swapped: original AR debit becomes credit orig_line = [l for l in content.splitlines() if "1149.75" in l][0] rev_line = [l for l in rev.splitlines() if "1149.75" in l][0] assert "debit" in orig_line and "credit" in rev_line def test_native_backend_roundtrip(tmp_path: Path) -> None: journal = _demo_journal() backend = NativeLedgerBackend(tmp_path / "books.jsonl") receipt = backend.post(backend.compile(journal), "batch-1") assert receipt.details["entries_appended"] == "3" assert backend.ledger.verify_chain()