net-kingdom/tools/posture-feedback/tests/test_execution_feedback.py

"""Execution findings from bounded-executor receipts (posture-feedback v0.1 section 4a, NK-WP-0056-T07)."""
from __future__ import annotations

import copy
import datetime as dt
import importlib.util
import json
import pathlib
import sys
import tempfile
import unittest
import uuid

HERE = pathlib.Path(__file__).resolve().parent
ROOT = HERE.parents[2]
FIXTURES = ROOT / "tools/execution-attribution-receipt/fixtures/valid"
spec = importlib.util.spec_from_file_location("posture_feedback_exec", HERE.parent / "posture_feedback.py")
PF = importlib.util.module_from_spec(spec)
sys.modules["posture_feedback_exec"] = PF
spec.loader.exec_module(PF)
AS_OF = dt.datetime(2026, 11, 4, 12, 0, tzinfo=dt.UTC)


def events(name):
    return [json.loads((FIXTURES / f"{name}-{k}.json").read_text()) for k in ("started", "receipt")]


def renamed(evs, admission):
    """The same run shape as another run of another admission (fresh run id)."""
    rid = str(uuid.uuid4())
    out = []
    for e in copy.deepcopy(evs):
        kind = "started" if e["type"] == "execution.run.started" else "receipt"
        e.update(id=f"run:{rid}:{kind}", correlation_id=rid, subject=f"admission:{admission}")
        e["data"]["run_id"] = rid
        if kind == "started":
            e["data"]["admission"] = admission
        else:
            e["data"]["target"] = f"admission:{admission}"
            e["data"]["decision_refs"][0]["ref"] = admission
        out.append(e)
    return out


class ExecutionFeedback(unittest.TestCase):
    def setUp(self):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        self.dir = pathlib.Path(tmp.name)

    def report(self, *runs, owners=None, fail_on="none"):
        paths = []
        for i, evs in enumerate(runs):
            path = self.dir / f"run-{i}.json"
            path.write_text(json.dumps(evs))
            paths.append(path)
        return PF.build_report([], as_of=AS_OF, horizon_days=14, fail_on=fail_on, receipts=paths, owners=owners)

    def classes(self, report):
        return sorted((f["class"], f["severity"]) for f in report["findings"])

    def test_a_clean_run_raises_nothing(self):
        report = self.report(events("completed"))
        self.assertEqual((report["findings"], report["validation_errors"], report["ok"]), ([], [], True))

    def test_reverted_and_retry_findings_are_routed_to_the_declared_owner(self):
        report = self.report(events("reverted"), events("completed-on-retry"))
        self.assertEqual(self.classes(report), [("execution-budget-pressure", "low"), ("execution-reverted", "medium")])
        self.assertEqual({f["owner"] for f in report["findings"]}, {"net-kingdom"})
        self.assertEqual({f["service"] for f in report["findings"]}, {"service:nk-bind-executor"})
        self.assertEqual(report["automation"]["mode"], "proposal-only")

    def test_an_unmapped_executor_is_unknown_never_inferred(self):
        report = self.report(events("reverted"), owners={})
        self.assertEqual({f["owner"] for f in report["findings"]}, {"unknown"})
        self.assertEqual(report["summary"]["unknown_owner"], 1)

    def test_not_completed_is_high(self):
        evs = events("reverted")
        evs[1]["data"].update(outcome="unknown", reverted=True)
        report = self.report(evs, fail_on="high")
        self.assertIn(("execution-not-completed", "high"), self.classes(report))
        self.assertFalse(report["ok"])

    def test_a_start_without_receipt_is_high(self):
        report = self.report(events("completed")[:1])
        self.assertEqual(self.classes(report), [("execution-receipt-missing", "high")])

    def test_recurring_reverts_are_reviewed_in_aggregate(self):
        base = events("reverted")
        report = self.report(base, renamed(base, "BEA-2026-0002"))
        aggregate = [f for f in report["findings"] if f["class"] == "execution-recurring-revert"]
        self.assertEqual(len(aggregate), 1)
        self.assertEqual((aggregate[0]["severity"], aggregate[0]["control"]), ("high", "consumers:authelia"))
        self.assertIn("BEA-2026-0001", aggregate[0]["reason"])
        self.assertIn("BEA-2026-0002", aggregate[0]["reason"])

    def test_invalid_events_are_validation_errors_not_findings(self):
        evs = events("completed")
        evs[1]["data"]["bind_password"] = "synthetic-value"
        report = self.report(evs)
        self.assertTrue(report["validation_errors"])
        self.assertEqual(report["findings"], [])
        self.assertNotIn("synthetic-value", json.dumps(report))

    def test_a_mismatched_pair_is_a_validation_error(self):
        started, receipt = events("completed")
        started["data"]["approval"] = "another-approval"
        report = self.report([started, receipt])
        self.assertTrue(any("EAR-10" in e for e in report["validation_errors"]))

    def test_deterministic(self):
        first = self.report(events("reverted"), events("completed-on-retry"))
        second = self.report(events("reverted"), events("completed-on-retry"))
        self.assertEqual(first["report_digest"], second["report_digest"])

    def test_cli_receipts_only(self):
        path = self.dir / "r.json"
        path.write_text(json.dumps(events("reverted")))
        self.assertEqual(PF.main(["--as-of", "2026-11-04T12:00:00Z", "--fail-on", "medium", "--receipts", str(path)]), 1)
        with self.assertRaises(SystemExit):
            PF.main(["--as-of", "2026-11-04T12:00:00Z"])


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