Pulse/scripts/release_control/control_plane_audit_test.py

import os
from pathlib import Path
import re
import tempfile
import textwrap
import shlex
import subprocess
import sys
import unittest

from control_plane import (
    REPO_ROOT,
    agent_entrypoint,
    parse_args as control_plane_parse_args,
    release_branch_for_version,
)
from control_plane_audit import audit_control_plane_payload, parse_args


VALID_PAYLOAD = {
    "version": "1",
    "system": "pulse-release-control",
    "execution_model": "direct-repo-sessions",
    "agent_values_doc": "docs/release-control/internal/AGENT_VALUES.md",
    "control_plane_doc": "docs/release-control/internal/CONTROL_PLANE.md",
    "control_plane_schema": "docs/release-control/control_plane.schema.json",
    "active_profile_id": "v6",
    "active_target_id": "v6-rc-cut",
    "legacy_release_lines": [
        {
            "version_prefix": "5.1.",
            "prerelease_branch": "main",
            "stable_branch": "main",
        }
    ],
    "profiles": [
        {
            "id": "v6",
            "lifecycle": "active",
            "root": "docs/release-control/v6",
            "prerelease_branch": "main",
            "stable_branch": "main",
            "source_of_truth": "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md",
            "status": "docs/release-control/v6/internal/status.json",
            "status_schema": "docs/release-control/v6/status.schema.json",
            "development_protocol": "docs/release-control/v6/internal/CANONICAL_DEVELOPMENT_PROTOCOL.md",
            "high_risk_matrix": "docs/release-control/v6/internal/HIGH_RISK_RELEASE_VERIFICATION_MATRIX.md",
            "subsystems_dir": "docs/release-control/v6/subsystems",
            "registry": "docs/release-control/v6/internal/subsystems/registry.json",
            "registry_schema": "docs/release-control/v6/internal/subsystems/registry.schema.json",
            "subsystem_contract_template": "docs/release-control/v6/SUBSYSTEM_CONTRACT_TEMPLATE.md",
        }
    ],
    "targets": [
        {
            "id": "v6-rc-cut",
            "profile_id": "v6",
            "kind": "release",
            "status": "active",
            "summary": "Drive Pulse v6 to a governed RC cut.",
            "completion_rule": "rc_ready",
            "proof_scope": "derived",
        },
        {
            "id": "v6-ga-promotion",
            "profile_id": "v6",
            "kind": "release",
            "status": "planned",
            "summary": "Promote Pulse v6 from validated RC to governed GA.",
            "completion_rule": "release_ready",
            "proof_scope": "derived",
        }
    ],
}

VALID_PATH_KINDS = {
    "docs/release-control/internal/AGENT_VALUES.md": "file",
    "docs/release-control/internal/CONTROL_PLANE.md": "file",
    "docs/release-control/control_plane.schema.json": "file",
    "docs/release-control/v6": "dir",
    "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md": "file",
    "docs/release-control/v6/internal/status.json": "file",
    "docs/release-control/v6/status.schema.json": "file",
    "docs/release-control/v6/internal/CANONICAL_DEVELOPMENT_PROTOCOL.md": "file",
    "docs/release-control/v6/internal/HIGH_RISK_RELEASE_VERIFICATION_MATRIX.md": "file",
    "docs/release-control/v6/subsystems": "dir",
    "docs/release-control/v6/internal/subsystems/registry.json": "file",
    "docs/release-control/v6/internal/subsystems/registry.schema.json": "file",
    "docs/release-control/v6/SUBSYSTEM_CONTRACT_TEMPLATE.md": "file",
}


class ControlPlaneAuditTest(unittest.TestCase):
    def test_control_plane_parse_args_accepts_agent_entrypoint(self) -> None:
        args = control_plane_parse_args(["--agent-entrypoint", "--pretty"])
        self.assertTrue(args.agent_entrypoint)
        self.assertTrue(args.pretty)

    def test_parse_args_accepts_staged_flag(self) -> None:
        args = parse_args(["--check", "--staged"])
        self.assertTrue(args.check)
        self.assertTrue(args.staged)

    def test_agent_entrypoint_lists_startup_files_and_commands(self) -> None:
        entrypoint = agent_entrypoint()
        self.assertEqual(entrypoint["agent_values_doc"], "docs/release-control/internal/AGENT_VALUES.md")
        self.assertEqual(
            entrypoint["startup_files"],
            [
                "docs/release-control/internal/AGENT_VALUES.md",
                "docs/release-control/internal/CONTROL_PLANE.md",
                "docs/release-control/control_plane.json",
                "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md",
                "docs/release-control/v6/internal/CANONICAL_DEVELOPMENT_PROTOCOL.md",
            ],
        )
        self.assertEqual(
            entrypoint["startup_commands"],
            [
                "python3 scripts/release_control/agent_preflight.py --pretty",
                "python3 scripts/release_control/status_audit.py --pretty",
            ],
        )
        if entrypoint["active_target_id"] == "v6-product-lane-expansion":
            self.assertEqual(entrypoint["default_pick_surface"], "available_candidate_lane_queue")
            self.assertIn("candidate lanes", entrypoint["selection_rule"])
            self.assertIn(
                "python3 scripts/release_control/status_lookup.py --candidate-lane <CANDIDATE_LANE_ID> --pretty",
                entrypoint["targeted_lookup_commands"],
            )
            self.assertIn(
                "python3 scripts/release_control/status_lookup.py --coverage-gap <COVERAGE_GAP_ID> --pretty",
                entrypoint["targeted_lookup_commands"],
            )
        else:
            self.assertEqual(entrypoint["default_pick_surface"], "lanes")
            self.assertIn("lane map", entrypoint["selection_rule"])
        self.assertIn(
            "python3 scripts/release_control/status_lookup.py --lane <LANE_ID> --pretty",
            entrypoint["targeted_lookup_commands"],
        )
        self.assertIn(
            "python3 scripts/release_control/work_claim.py --kind <KIND> --id <ID> --summary <SUMMARY> --agent-id <AGENT_ID> --pretty",
            entrypoint["targeted_lookup_commands"],
        )
        self.assertIn(
            "python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...] --pretty --lean",
            entrypoint["targeted_lookup_commands"],
        )
        self.assertIn("reserve exactly one", entrypoint["claim_rule"])
        self.assertIn("record any remaining same-lane residual", entrypoint["claim_rule"])
        self.assertIn(
            "python3 scripts/release_control/worktree_base.py --base-branch <BASE_BRANCH> --pretty",
            entrypoint["targeted_lookup_commands"],
        )
        self.assertIn(
            "python3 scripts/release_control/worktree_claim.py --kind <KIND> --id <ID> --summary <SUMMARY> --agent-id <AGENT_ID> --pretty",
            entrypoint["targeted_lookup_commands"],
        )
        self.assertIn(
            "python3 scripts/release_control/worktree_finish.py --base-branch <BASE_BRANCH> --pretty",
            entrypoint["targeted_lookup_commands"],
        )
        self.assertEqual(
            entrypoint["escalation_files"],
            [
                "docs/release-control/v6/internal/status.json",
                "docs/release-control/v6/status.schema.json",
                "docs/release-control/v6/internal/subsystems/registry.json",
                "docs/release-control/v6/internal/subsystems/registry.schema.json",
            ],
        )
        self.assertEqual(entrypoint["work_claim"], "scripts/release_control/work_claim.py")
        self.assertEqual(entrypoint["worktree_base"], "scripts/release_control/worktree_base.py")
        self.assertEqual(entrypoint["worktree_claim"], "scripts/release_control/worktree_claim.py")
        self.assertEqual(entrypoint["worktree_finish"], "scripts/release_control/worktree_finish.py")

    def test_every_advertised_helper_entrypoint_executes(self) -> None:
        entrypoint = agent_entrypoint()
        commands = entrypoint["startup_commands"] + entrypoint["targeted_lookup_commands"]
        helper_paths = {
            tokens[1]
            for command in commands
            if len(tokens := shlex.split(command)) >= 2 and tokens[0] == "python3"
        }
        helper_paths.update(
            value
            for value in entrypoint.values()
            if isinstance(value, str) and value.endswith(".py")
        )

        for helper_rel in sorted(helper_paths):
            with self.subTest(helper=helper_rel):
                helper_path = REPO_ROOT / helper_rel
                self.assertTrue(helper_path.is_file(), f"advertised helper is missing: {helper_rel}")
                result = subprocess.run(
                    [sys.executable, str(helper_path), "--help"],
                    cwd=REPO_ROOT,
                    capture_output=True,
                    text=True,
                    check=False,
                )
                self.assertEqual(
                    result.returncode,
                    0,
                    f"advertised helper failed to execute: {helper_rel}\n{result.stderr}",
                )
                self.assertIn("usage:", result.stdout)

    def test_audit_accepts_valid_payload(self) -> None:
        report = audit_control_plane_payload(
            VALID_PAYLOAD,
            path_kinds=VALID_PATH_KINDS,
            status_report={"errors": [], "control_plane": {"active_target": {"completion_met": False}}},
        )

        self.assertEqual(report["errors"], [])
        self.assertEqual(report["summary"]["profile_count"], 1)
        self.assertFalse(report["summary"]["active_target_completion_met"])

    def test_release_branch_for_version_uses_profile_branch_policy(self) -> None:
        self.assertEqual(release_branch_for_version("5.1.27", control_plane=VALID_PAYLOAD), "main")
        self.assertEqual(release_branch_for_version("5.1.27-rc.3", control_plane=VALID_PAYLOAD), "main")
        self.assertEqual(release_branch_for_version("6.0.0-rc.1", control_plane=VALID_PAYLOAD), "main")
        self.assertEqual(release_branch_for_version("6.0.0-dev", control_plane=VALID_PAYLOAD), "main")
        self.assertEqual(release_branch_for_version("6.0.0", control_plane=VALID_PAYLOAD), "main")

    def test_release_branch_for_version_prefers_longest_legacy_prefix(self) -> None:
        payload = dict(VALID_PAYLOAD)
        payload["legacy_release_lines"] = [
            {
                "version_prefix": "5.",
                "prerelease_branch": "main",
                "stable_branch": "main",
            },
            {
                "version_prefix": "5.1.",
                "prerelease_branch": "pulse/release-5.1.25",
                "stable_branch": "pulse/release-5.1.25",
            },
        ]
        self.assertEqual(
            release_branch_for_version("5.1.27", control_plane=payload),
            "pulse/release-5.1.25",
        )

    def test_forward_patch_train_uses_actual_control_plane(self) -> None:
        for version in ("6.4.4-beta.1", "v6.4.4-beta.2", "6.4.4-rc.1",
                        "6.4.4", "v6.4.4+build.1", "6.4.3-rc.1", "6.4.3",
                        "6.4.5-beta.1", "v6.4.5-beta.2", "6.4.5-rc.1",
                        "6.4.5", "v6.4.5+build.1"):
            with self.subTest(version=version):
                self.assertEqual(release_branch_for_version(version), "release/v6.4")
        for version in ("6.4.1", "6.4.2", "6.4.6-beta.1", "6.4.50-beta.1", "6.4.40-beta.1",
                        "6.4.30", "6.3.20", "6.6.0-beta.1"):
            with self.subTest(version=version):
                self.assertEqual(release_branch_for_version(version), "main")
        self.assertEqual(release_branch_for_version("6.5.10-beta.1"), "release/v6.5")

    def test_forward_patch_workflow_branch_contract(self) -> None:
        # Execute the real branch-policy shell only, never dispatch a workflow.
        for workflow in ("create-release.yml", "release-dry-run.yml"):
            content = (REPO_ROOT / ".github/workflows" / workflow).read_text()
            match = re.search(
                r"(?ms)^      - name: Resolve required release branch\n"
                r".*?^        run: \|\n((?:          [^\n]*\n|\n)+)", content
            )
            self.assertIsNotNone(match)
            script = textwrap.dedent(match.group(1))
            for version in ("6.4.4-beta.1", "6.4.5-beta.1", "6.4.5-rc.1", "6.4.5"):
                for branch in ("main", "release/v6.4"):
                    with self.subTest(workflow=workflow, branch=branch, version=version), tempfile.TemporaryDirectory() as tmp:
                        output = os.path.join(tmp, "output")
                        result = subprocess.run(
                            ["bash", "-euo", "pipefail", "-c", script],
                            cwd=REPO_ROOT, capture_output=True, text=True,
                            env={**os.environ, "GITHUB_OUTPUT": output,
                                 "VERSION_INPUT": version,
                                 "WORKFLOW_OUTPUT_1": version,
                                 "WORKFLOW_OUTPUT_2": branch},
                        )
                        rejects = workflow == "create-release.yml" and branch == "main"
                        self.assertEqual(result.returncode, 1 if rejects else 0, result.stderr)
                        if rejects:
                            self.assertIn("must run from release/v6.4", result.stdout)
                        else:
                            self.assertRegex(Path(output).read_text(),
                                             r"^required_branch<<([^\n]+)\nrelease/v6.4\n\1\n$")

    def test_audit_flags_stale_active_target(self) -> None:
        report = audit_control_plane_payload(
            VALID_PAYLOAD,
            path_kinds=VALID_PATH_KINDS,
            status_report={"errors": [], "control_plane": {"active_target": {"completion_met": True}}},
        )

        self.assertTrue(report["errors"])
        self.assertIn("already satisfies its completion rule", "\n".join(report["errors"]))

    def test_audit_flags_missing_profile_path(self) -> None:
        broken_paths = dict(VALID_PATH_KINDS)
        broken_paths["docs/release-control/v6/internal/status.json"] = "missing"

        report = audit_control_plane_payload(
            VALID_PAYLOAD,
            path_kinds=broken_paths,
            status_report={"errors": [], "control_plane": {"active_target": {"completion_met": False}}},
        )

        self.assertTrue(report["errors"])
        self.assertIn("expects file", "\n".join(report["errors"]))


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