Add direct WP-0024 owner review interface

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa

tests/test_wp0024_owner_review.py

@@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import subprocess
+import tempfile
+import unittest
+from datetime import UTC, datetime
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC = importlib.util.spec_from_file_location(
+    "wp0024_owner_review", ROOT / "scripts" / "wp0024-owner-review.py"
+)
+assert SPEC and SPEC.loader
+module = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(module)
+
+
+class OwnerReviewTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.contract = module.load_contract()
+
+    def test_contract_has_four_owners_and_only_known_checks(self) -> None:
+        self.assertEqual(
+            {"audit-core", "rapp-postgres", "railiance-cluster", "railiance-infra"},
+            set(self.contract["owners"]),
+        )
+        checks = {
+            check
+            for review in self.contract["owners"].values()
+            for check in review["checks"]
+        }
+        self.assertLessEqual(checks, module.KNOWN_CHECKS)
+        self.assertEqual(64, len(module.contract_digest(self.contract)))
+
+    def test_current_artifacts_match_contract(self) -> None:
+        for owner in self.contract["owners"]:
+            state = module.artifact_state(self.contract, owner)
+            self.assertTrue(all(item["matches"] for item in state.values()))
+
+    def test_artifact_change_prevents_checks_and_approval(self) -> None:
+        contract = json.loads(json.dumps(self.contract))
+        owner = "rapp-postgres"
+        review = contract["owners"][owner]
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            for relative in review["artifacts"]:
+                target = root / relative
+                target.parent.mkdir(parents=True, exist_ok=True)
+                target.write_bytes((ROOT / relative).read_bytes())
+            (root / review["artifacts"][0]).write_text("changed", encoding="utf-8")
+            result = module.verify(
+                contract,
+                owner,
+                root=root,
+                check_runner=lambda _: self.fail("checks must not run after hash mismatch"),
+            )
+        self.assertFalse(result["passed"])
+        self.assertEqual([], result["checks"])
+        with self.assertRaises(module.ReviewError):
+            module.build_receipt(
+                contract, owner, "approve", owner, verification=result
+            )
+
+    def test_named_preflight_requires_safe_machine_result(self) -> None:
+        def runner(*args, **kwargs):
+            return subprocess.CompletedProcess(
+                args[0],
+                0,
+                stdout=json.dumps(
+                    {
+                        "automated_checks_passed": True,
+                        "ready_for_live_execution": False,
+                        "secret_values_observed": False,
+                    }
+                ),
+                stderr="",
+            )
+
+        result = module.run_named_check("database-lease-preflight", runner=runner)
+        self.assertTrue(result["passed"])
+        self.assertFalse(result["ready_for_live_execution"])
+        self.assertFalse(result["secret_values_observed"])
+
+    def test_receipt_round_trip_and_stale_digest_invalidation(self) -> None:
+        owner = "audit-core"
+        verification = {
+            "passed": True,
+            "checks": [
+                {"id": check, "passed": True}
+                for check in self.contract["owners"][owner]["checks"]
+            ],
+        }
+        receipt = module.build_receipt(
+            self.contract,
+            owner,
+            "approve",
+            owner,
+            verification=verification,
+            now=datetime(2026, 8, 22, 12, 0, tzinfo=UTC),
+        )
+        message = {
+            "from_agent": owner,
+            "subject": module.receipt_subject(receipt),
+            "body": module.canonical_json(receipt),
+        }
+        self.assertEqual(receipt, module.parse_receipt_message(message))
+        current = module.aggregate_status(self.contract, [message])
+        self.assertEqual("approve", current["owners"][owner]["decision"])
+        self.assertTrue(current["contract_artifacts_current"])
+        self.assertFalse(current["all_approved"])
+
+        stale = json.loads(json.dumps(receipt))
+        stale["contract_digest"] = "0" * 64
+        stale_message = {
+            "from_agent": owner,
+            "subject": module.receipt_subject(stale),
+            "body": module.canonical_json(stale),
+        }
+        status = module.aggregate_status(self.contract, [stale_message])
+        self.assertEqual("missing", status["owners"][owner]["decision"])
+        self.assertEqual(1, status["owners"][owner]["stale_receipt_count"])
+
+    def test_incomplete_approval_is_not_counted(self) -> None:
+        owner = "rapp-postgres"
+        verification = {
+            "passed": True,
+            "checks": [
+                {"id": check, "passed": True}
+                for check in self.contract["owners"][owner]["checks"]
+            ],
+        }
+        receipt = module.build_receipt(
+            self.contract, owner, "approve", owner, verification=verification
+        )
+        receipt["checks"].pop()
+        message = {
+            "from_agent": owner,
+            "subject": module.receipt_subject(receipt),
+            "body": module.canonical_json(receipt),
+        }
+        status = module.aggregate_status(self.contract, [message])
+        self.assertEqual("missing", status["owners"][owner]["decision"])
+        self.assertEqual(1, status["owners"][owner]["stale_receipt_count"])
+
+    def test_malformed_receipt_message_is_ignored(self) -> None:
+        message = {
+            "from_agent": "audit-core",
+            "subject": module.RECEIPT_PREFIX + "/invalid",
+            "body": "{}",
+        }
+        self.assertIsNone(module.parse_receipt_message(message))
+
+    def test_request_changes_requires_safe_note(self) -> None:
+        receipt = module.build_receipt(
+            self.contract,
+            "railiance-infra",
+            "request-changes",
+            "infra-owner",
+            note="Name the provider-console operator.",
+        )
+        self.assertEqual("request-changes", receipt["decision"])
+        with self.assertRaises(module.ReviewError):
+            module.build_receipt(
+                self.contract,
+                "railiance-infra",
+                "request-changes",
+                "infra-owner",
+                note="",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()