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_syscalls.py6# Description : Tests — agent syscalls, closed periods, and the hash-chained audit log.7# =============================================================================8"""Syscall/kernel tests.910Guarantees under test:11- agents get books access ONLY through syscalls, and every call (ok or error)12 is one hash-chained audit record, in order;13- posting is idempotent and refuses closed periods;14- Reverse appends a contra entry, never edits;15- audit tampering is detected.16"""17from __future__ import annotations1819import json20from datetime import datetime, timezone21from pathlib import Path2223import pytest2425from kernel.audit import AuditLog26from kernel.workspace import Workspace27from sdk.cli import main28from sdk.syscalls import AirKernel, SyscallError2930ROOT = Path(__file__).resolve().parents[1]31POLICY = str(ROOT / "alsl" / "policies" / "ca-qc-2026.yaml")3233SALE = {34 "id": "evt_sc_sale", "type": "Sale", "date": "2026-02-10",35 "amount": {"amount": "1000.00", "currency": "CAD"},36 "tax": {"jurisdiction": "CA-QC"},37}383940def _kernel(tmp_path: Path, actor: str = "agent:test") -> AirKernel:41 home = tmp_path / "books"42 if not (home / "meta.json").exists():43 Workspace.init(home, policies=POLICY)44 tick = iter(range(10_000))45 return AirKernel(home, actor=actor,46 clock=lambda: datetime(2026, 8, 5, 12, 0, next(tick),47 tzinfo=timezone.utc))484950# --- happy path -----------------------------------------------------------------51def test_full_agent_lifecycle_is_audited(tmp_path: Path) -> None:52 kernel = _kernel(tmp_path)5354 event_id = kernel.create_economic_event(SALE)55 assert event_id == "evt_sc_sale"56 diagnostics = kernel.validate()57 assert not any(d.severity.value == "error" for d in diagnostics)58 journal = kernel.compile()59 assert len(journal.entries) == 16061 receipt = kernel.post()62 assert receipt["entries"] == 1 and receipt["appended"] == 16364 report = kernel.generate_report("trial-balance")65 assert "1149.75" in report6667 calls = [(r["syscall"], r["status"]) for r in kernel.audit.records]68 assert calls == [69 ("CreateEconomicEvent", "ok"), ("Validate", "ok"), ("Compile", "ok"),70 ("Post", "ok"), ("GenerateReport", "ok"),71 ]72 assert kernel.audit.verify_chain()737475def test_post_is_idempotent_and_archives_drafts(tmp_path: Path) -> None:76 kernel = _kernel(tmp_path)77 kernel.create_economic_event(SALE)78 receipt = kernel.post()79 # drafts staged -> archived on post; nothing left to post80 with pytest.raises(SyscallError, match="no draft events"):81 kernel.post()82 archived = list((kernel.workspace.root / "documents").glob("posted_*"))83 assert len(archived) == 184 assert receipt["entry_ids"] == ["je_evt_sc_sale"]858687def test_reverse_appends_contra(tmp_path: Path) -> None:88 kernel = _kernel(tmp_path)89 kernel.create_economic_event(SALE)90 kernel.post()91 contra_id = kernel.reverse("je_evt_sc_sale")92 assert contra_id == "rev_je_evt_sc_sale"93 # books net to zero after the reversal94 report = kernel.generate_report("trial-balance", "csv")95 total = [r for r in report.splitlines() if r.startswith("TOTAL")][0]96 assert total.split(",")[-1] == total.split(",")[-2]97 assert kernel.audit.verify_chain()9899100# --- refusals (audited errors) ------------------------------------------------------101def test_invalid_event_is_refused_and_audited(tmp_path: Path) -> None:102 kernel = _kernel(tmp_path)103 with pytest.raises(Exception):104 kernel.create_economic_event({105 "id": "evt_bad", "type": "Sale", "date": "2026-02-10",106 "amount": {"amount": 19.99, "currency": "CAD"}, # float: forbidden107 })108 (record,) = kernel.audit.records109 assert record["syscall"] == "CreateEconomicEvent"110 assert record["status"] == "error"111 assert kernel.audit.verify_chain()112113114def test_closed_period_refuses_posting(tmp_path: Path) -> None:115 kernel = _kernel(tmp_path)116 kernel.close_period("2026-02")117 kernel.create_economic_event(SALE) # dated 2026-02-10118 with pytest.raises(SyscallError, match="AIR-E700"):119 kernel.post()120 statuses = {r["syscall"]: r["status"] for r in kernel.audit.records}121 assert statuses["ClosePeriod"] == "ok"122 assert statuses["Post"] == "error"123 # the drafts survive: the agent can restage into an open period124 assert len(list((kernel.workspace.root / "drafts").glob("*.json"))) == 1125126127def test_reverse_unknown_entry_is_refused(tmp_path: Path) -> None:128 kernel = _kernel(tmp_path)129 with pytest.raises(SyscallError):130 kernel.reverse("je_nonexistent")131 assert kernel.audit.records[-1]["status"] == "error"132133134def test_unknown_report_is_refused(tmp_path: Path) -> None:135 kernel = _kernel(tmp_path)136 with pytest.raises(SyscallError, match="unknown report"):137 kernel.generate_report("profit-magic")138139140# --- audit log integrity ---------------------------------------------------------------141def test_audit_log_tampering_is_detected(tmp_path: Path) -> None:142 kernel = _kernel(tmp_path)143 kernel.create_economic_event(SALE)144 kernel.post()145146 audit_path = kernel.workspace.root / "audit.jsonl"147 lines = audit_path.read_text().splitlines()148 record = json.loads(lines[0])149 record["actor"] = "agent:impostor" # rewrite history150 lines[0] = json.dumps(record, sort_keys=True, separators=(",", ":"))151 audit_path.write_text("\n".join(lines) + "\n")152153 assert not AuditLog(path=audit_path).verify_chain()154155156def test_audit_log_persists_across_sessions(tmp_path: Path) -> None:157 kernel = _kernel(tmp_path)158 kernel.create_economic_event(SALE)159 kernel.post()160 # a second agent session on the same home continues the same chain161 kernel2 = AirKernel(kernel.workspace.root, actor="agent:second")162 kernel2.generate_report("balance-sheet")163 assert len(kernel2.audit.records) == 3164 assert {r["actor"] for r in kernel2.audit.records} == \165 {"agent:test", "agent:second"}166 assert kernel2.audit.verify_chain()167168169# --- demo agent + CLI -----------------------------------------------------------------170def test_demo_agent_end_to_end(tmp_path: Path, capsys) -> None:171 from sdk.demo_agent import run172 home = tmp_path / "books"173 Workspace.init(home, policies=POLICY)174 assert run(str(home), actor="agent:demo") == 0175 out = capsys.readouterr().out176 assert "refused as expected" in out # closed-period enforcement177 assert "chain VALID" in out178179180def test_cli_audit_lists_and_verifies(tmp_path: Path, capsys) -> None:181 home = tmp_path / "books"182 Workspace.init(home, policies=POLICY)183 kernel = AirKernel(home, actor="agent:test")184 kernel.create_economic_event(SALE)185 kernel.post()186 assert main(["audit", "--home", str(home)]) == 0187 out = capsys.readouterr().out188 assert "CreateEconomicEvent" in out and "hash chain VALID" in out189