# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : test_syscalls.py # Description : Tests — agent syscalls, closed periods, and the hash-chained audit log. # ============================================================================= """Syscall/kernel tests. Guarantees under test: - agents get books access ONLY through syscalls, and every call (ok or error) is one hash-chained audit record, in order; - posting is idempotent and refuses closed periods; - Reverse appends a contra entry, never edits; - audit tampering is detected. """ from __future__ import annotations import json from datetime import datetime, timezone from pathlib import Path import pytest from kernel.audit import AuditLog from kernel.workspace import Workspace from sdk.cli import main from sdk.syscalls import AirKernel, SyscallError ROOT = Path(__file__).resolve().parents[1] POLICY = str(ROOT / "alsl" / "policies" / "ca-qc-2026.yaml") SALE = { "id": "evt_sc_sale", "type": "Sale", "date": "2026-02-10", "amount": {"amount": "1000.00", "currency": "CAD"}, "tax": {"jurisdiction": "CA-QC"}, } def _kernel(tmp_path: Path, actor: str = "agent:test") -> AirKernel: home = tmp_path / "books" if not (home / "meta.json").exists(): Workspace.init(home, policies=POLICY) tick = iter(range(10_000)) return AirKernel(home, actor=actor, clock=lambda: datetime(2026, 8, 5, 12, 0, next(tick), tzinfo=timezone.utc)) # --- happy path ----------------------------------------------------------------- def test_full_agent_lifecycle_is_audited(tmp_path: Path) -> None: kernel = _kernel(tmp_path) event_id = kernel.create_economic_event(SALE) assert event_id == "evt_sc_sale" diagnostics = kernel.validate() assert not any(d.severity.value == "error" for d in diagnostics) journal = kernel.compile() assert len(journal.entries) == 1 receipt = kernel.post() assert receipt["entries"] == 1 and receipt["appended"] == 1 report = kernel.generate_report("trial-balance") assert "1149.75" in report calls = [(r["syscall"], r["status"]) for r in kernel.audit.records] assert calls == [ ("CreateEconomicEvent", "ok"), ("Validate", "ok"), ("Compile", "ok"), ("Post", "ok"), ("GenerateReport", "ok"), ] assert kernel.audit.verify_chain() def test_post_is_idempotent_and_archives_drafts(tmp_path: Path) -> None: kernel = _kernel(tmp_path) kernel.create_economic_event(SALE) receipt = kernel.post() # drafts staged -> archived on post; nothing left to post with pytest.raises(SyscallError, match="no draft events"): kernel.post() archived = list((kernel.workspace.root / "documents").glob("posted_*")) assert len(archived) == 1 assert receipt["entry_ids"] == ["je_evt_sc_sale"] def test_reverse_appends_contra(tmp_path: Path) -> None: kernel = _kernel(tmp_path) kernel.create_economic_event(SALE) kernel.post() contra_id = kernel.reverse("je_evt_sc_sale") assert contra_id == "rev_je_evt_sc_sale" # books net to zero after the reversal report = kernel.generate_report("trial-balance", "csv") total = [r for r in report.splitlines() if r.startswith("TOTAL")][0] assert total.split(",")[-1] == total.split(",")[-2] assert kernel.audit.verify_chain() # --- refusals (audited errors) ------------------------------------------------------ def test_invalid_event_is_refused_and_audited(tmp_path: Path) -> None: kernel = _kernel(tmp_path) with pytest.raises(Exception): kernel.create_economic_event({ "id": "evt_bad", "type": "Sale", "date": "2026-02-10", "amount": {"amount": 19.99, "currency": "CAD"}, # float: forbidden }) (record,) = kernel.audit.records assert record["syscall"] == "CreateEconomicEvent" assert record["status"] == "error" assert kernel.audit.verify_chain() def test_closed_period_refuses_posting(tmp_path: Path) -> None: kernel = _kernel(tmp_path) kernel.close_period("2026-02") kernel.create_economic_event(SALE) # dated 2026-02-10 with pytest.raises(SyscallError, match="AIR-E700"): kernel.post() statuses = {r["syscall"]: r["status"] for r in kernel.audit.records} assert statuses["ClosePeriod"] == "ok" assert statuses["Post"] == "error" # the drafts survive: the agent can restage into an open period assert len(list((kernel.workspace.root / "drafts").glob("*.json"))) == 1 def test_reverse_unknown_entry_is_refused(tmp_path: Path) -> None: kernel = _kernel(tmp_path) with pytest.raises(SyscallError): kernel.reverse("je_nonexistent") assert kernel.audit.records[-1]["status"] == "error" def test_unknown_report_is_refused(tmp_path: Path) -> None: kernel = _kernel(tmp_path) with pytest.raises(SyscallError, match="unknown report"): kernel.generate_report("profit-magic") # --- audit log integrity --------------------------------------------------------------- def test_audit_log_tampering_is_detected(tmp_path: Path) -> None: kernel = _kernel(tmp_path) kernel.create_economic_event(SALE) kernel.post() audit_path = kernel.workspace.root / "audit.jsonl" lines = audit_path.read_text().splitlines() record = json.loads(lines[0]) record["actor"] = "agent:impostor" # rewrite history lines[0] = json.dumps(record, sort_keys=True, separators=(",", ":")) audit_path.write_text("\n".join(lines) + "\n") assert not AuditLog(path=audit_path).verify_chain() def test_audit_log_persists_across_sessions(tmp_path: Path) -> None: kernel = _kernel(tmp_path) kernel.create_economic_event(SALE) kernel.post() # a second agent session on the same home continues the same chain kernel2 = AirKernel(kernel.workspace.root, actor="agent:second") kernel2.generate_report("balance-sheet") assert len(kernel2.audit.records) == 3 assert {r["actor"] for r in kernel2.audit.records} == \ {"agent:test", "agent:second"} assert kernel2.audit.verify_chain() # --- demo agent + CLI ----------------------------------------------------------------- def test_demo_agent_end_to_end(tmp_path: Path, capsys) -> None: from sdk.demo_agent import run home = tmp_path / "books" Workspace.init(home, policies=POLICY) assert run(str(home), actor="agent:demo") == 0 out = capsys.readouterr().out assert "refused as expected" in out # closed-period enforcement assert "chain VALID" in out def test_cli_audit_lists_and_verifies(tmp_path: Path, capsys) -> None: home = tmp_path / "books" Workspace.init(home, policies=POLICY) kernel = AirKernel(home, actor="agent:test") kernel.create_economic_event(SALE) kernel.post() assert main(["audit", "--home", str(home)]) == 0 out = capsys.readouterr().out assert "CreateEconomicEvent" in out and "hash chain VALID" in out