net-kingdom/tools/execution-attribution-receipt/tests/test_validate.py

"""Execution Attribution Receipt validator (NK-WP-0056-T15). Fixtures come from the real executor."""
from __future__ import annotations

import copy
import importlib.util
import json
import pathlib
import sys
import unittest

HERE = pathlib.Path(__file__).resolve().parent
TOOL = HERE.parent


def load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


V = load("ear_validate", TOOL / "validate.py")
G = load("ear_generate", TOOL / "generate_fixtures.py")
VALID = TOOL / "fixtures/valid"
SCENARIOS = ("completed", "reverted", "completed-on-retry")


def fixture(name):
    return json.loads((VALID / name).read_text())


class Fixtures(unittest.TestCase):
    def test_committed_fixtures_are_what_the_executor_emits(self):
        self.assertEqual(G.main(["--check"]), 0)

    def test_every_valid_fixture_and_pair_passes(self):
        for s in SCENARIOS:
            with self.subTest(scenario=s):
                started, receipt = fixture(f"{s}-started.json"), fixture(f"{s}-receipt.json")
                for event in (started, receipt):
                    self.assertEqual(V.evaluate(event)["findings"], [])
                self.assertEqual(V.evaluate_pair(started, receipt), {"valid": True, "findings": []})

    def test_started_carries_the_window_audit_core_asked_for(self):
        data = fixture("completed-started.json")["data"]
        self.assertEqual(data["window"], {"not_after": "2026-11-03T11:00:00Z"})


class Rules(unittest.TestCase):
    def assert_rule(self, event, rule, at=None):
        report = V.evaluate(event)
        self.assertFalse(report["valid"])
        self.assertIn(rule, {f["rule"] for f in report["findings"]}, report)
        if at:
            self.assertIn(at, {f["at"] for f in report["findings"]})

    def receipt(self, name="completed"):
        return copy.deepcopy(fixture(f"{name}-receipt.json"))

    def test_schema_refusals(self):
        cases = {
            "extra field (a value could hide there)": lambda e: e["data"].update(bind_password="x"),
            "extra step field": lambda e: e["data"]["steps"][0].update(value="x"),
            "wrong source": lambda e: e.update(source="someone-else"),
            "naive time": lambda e: e.update(occurred_at="2026-11-03T09:31:00"),
            "unknown outcome": lambda e: e["data"].update(outcome="fine"),
            "issuer not https": lambda e: e["data"]["initiator"].update(iss="http://kc.local"),
            "error message instead of type": lambda e: e["data"]["steps"][0].update(error_type="bind failed: x y"),
            "missing receipt_version": lambda e: e["data"].pop("receipt_version"),
        }
        for name, mutate in cases.items():
            with self.subTest(case=name):
                event = self.receipt()
                mutate(event)
                self.assert_rule(event, "EAR-1")

    def test_cross_field_rules(self):
        cases = {
            "EAR-2": lambda e: e.update(id="run:" + "0" * 8 + "-0000-4000-8000-" + "0" * 12 + ":receipt"),
            "EAR-3": lambda e: e.update(subject="admission:BEA-2026-0002"),
            "EAR-4": lambda e: e.update(tenant="tenant:platform"),
            "EAR-5": lambda e: e["data"]["decision_refs"][0].update(ref="BEA-2026-0002"),
            "EAR-6": lambda e: e.update(occurred_at="2026-11-03T09:00:00Z"),
            "EAR-7": lambda e: e["data"].update(reverted=True),
            "EAR-8": lambda e: e["data"]["budget"]["used"].update(attempts=2),
            "EAR-9": lambda e: e["data"]["steps"][-1].update(outcome="check_failed"),
        }
        for rule, mutate in cases.items():
            with self.subTest(rule=rule):
                event = self.receipt()
                mutate(event)
                self.assert_rule(event, rule)

    def test_reverted_needs_a_clean_revert(self):
        event = self.receipt("reverted")
        next(s for s in event["data"]["steps"] if s["kind"] == "revert").update(outcome="check_failed")
        self.assert_rule(event, "EAR-9")
        event = self.receipt("completed")
        event["data"]["outcome"] = "reverted"
        self.assert_rule(event, "EAR-9")

    def test_completed_after_revert_needs_a_retry(self):
        event = self.receipt("completed-on-retry")
        event["data"]["attempts"] = event["data"]["budget"]["used"]["attempts"] = 1
        self.assert_rule(event, "EAR-9")

    def test_pair_rules(self):
        started, receipt = fixture("completed-started.json"), fixture("completed-receipt.json")
        cases = {
            "window": lambda s, r: s["data"]["window"].update(not_after="2026-11-03T09:30:10Z"),
            "approval": lambda s, r: s["data"].update(approval="another-approval"),
            "admission_digest": lambda s, r: s["data"].update(admission_digest="sha256:" + "0" * 64),
            "order": lambda s, r: s.update(occurred_at="2026-11-03T10:59:00Z"),
        }
        for at, mutate in cases.items():
            with self.subTest(at=at):
                s, r = copy.deepcopy(started), copy.deepcopy(receipt)
                mutate(s, r)
                report = V.evaluate_pair(s, r)
                self.assertIn({"rule": "EAR-10", "at": at}, report["findings"])
        report = V.evaluate_pair(receipt, started)
        self.assertFalse(report["valid"])

    def test_findings_never_carry_values(self):
        event = self.receipt()
        event["data"]["bind_password"] = "synthetic-secret-123"
        self.assertNotIn("synthetic-secret-123", json.dumps(V.evaluate(event)))


if __name__ == "__main__":
    unittest.main()