railiance-platform/tests/test_wp0024_owner_review.py

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_required_owners_and_only_known_checks(self) -> None:
        self.assertEqual(
            {
                "audit-core",
                "rapp-postgres",
                "railiance-cluster",
                "railiance-infra",
                "railiance-platform",
            },
            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(
            {
                "RAILIANCE-WP-0024-T02": [
                    "audit-core",
                    "rapp-postgres",
                    "railiance-platform",
                ],
                "RAILIANCE-WP-0024-T03": [
                    "audit-core",
                    "rapp-postgres",
                    "railiance-cluster",
                    "railiance-infra",
                    "railiance-platform",
                ],
            },
            self.contract["task_owners"],
        )
        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["tasks"]["RAILIANCE-WP-0024-T02"]["all_approved"])
        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()