"""Sanctioned restart / graceful reload control tests (#642).

Acceptance criteria under test:

1. The sanctioned restart path is implemented behind gates (capability,
   confirmation, operator authorization, host hook).
2. Manual ``pkill`` stays forbidden and is classified as contamination.
3. Post-restart mutations require clean health/session proof.
4. Authorized restart preview, unauthorized deny, contamination classification.
5. No entry point exposes a raw kill.
"""

import json
import os
import tempfile
import unittest

import mcp_namespace_health
import runtime_recovery_guard
from task_capability_map import TASK_CAPABILITY_MAP
from webui import console_audit, console_authz, gated_actions, sanctioned_restart

NAMESPACE = "gitea-author"

# An operator-authorized, hook-configured host. Passed explicitly so no test
# depends on (or mutates) the real process environment.
READY_ENV = {
    sanctioned_restart.RESTART_HOOK_ENV: "launchd:cc.prgs.gitea-author",
    runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV: "ops-ticket-4821",
}


def admin(subject: str = "admin@example.test") -> console_authz.Principal:
    return console_authz.Principal(
        subject=subject,
        role=console_authz.ADMIN,
        identity_source=console_authz.IDENTITY_ACCESS_PROXY,
        authenticated=True,
    )


def viewer() -> console_authz.Principal:
    return console_authz.Principal(
        subject="viewer@example.test",
        role=console_authz.VIEWER,
        identity_source=console_authz.IDENTITY_ACCESS_PROXY,
        authenticated=True,
    )


class TestCapabilityWiring(unittest.TestCase):
    """AC1: authority is declared, not invented by the console."""

    def test_actions_resolve_through_the_capability_map(self):
        for action_id in (
            sanctioned_restart.ACTION_RESTART_NAMESPACE,
            sanctioned_restart.ACTION_RELOAD_NAMESPACE,
        ):
            with self.subTest(action=action_id):
                action = console_authz.get_action(action_id)
                self.assertIsNotNone(action)
                self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
                self.assertEqual(
                    action.mcp_permission,
                    TASK_CAPABILITY_MAP[action.task_key]["permission"],
                )

    def test_restart_permission_is_not_a_gitea_operation(self):
        """No configured Gitea profile should satisfy a host restart."""
        permission = TASK_CAPABILITY_MAP["restart_namespace"]["permission"]
        self.assertFalse(permission.startswith("gitea."))

    def test_restart_is_destructive_dual_control_break_glass(self):
        action = console_authz.get_action(
            sanctioned_restart.ACTION_RESTART_NAMESPACE
        )
        self.assertEqual(action.action_class, console_authz.CLASS_DESTRUCTIVE)
        self.assertEqual(action.minimum_role, console_authz.ADMIN)
        self.assertTrue(action.dual_control)
        self.assertTrue(action.break_glass)
        self.assertTrue(action.requires_confirmation)

    def test_reload_is_privileged_but_not_destructive(self):
        action = console_authz.get_action(
            sanctioned_restart.ACTION_RELOAD_NAMESPACE
        )
        self.assertEqual(action.action_class, console_authz.CLASS_PRIVILEGED)
        self.assertTrue(action.requires_confirmation)


class TestPreview(unittest.TestCase):
    """AC4: an authorized preview renders the plan without executing it."""

    def test_preview_lists_the_mutation_ledger(self):
        preview = sanctioned_restart.build_restart_preview(
            NAMESPACE, principal=admin(), env=READY_ENV
        )
        steps = [entry["step"] for entry in preview["mutation_ledger"]]
        self.assertEqual(
            steps, ["quiesce", "host_restart_hook", "health_recheck", "audit"]
        )
        self.assertTrue(preview["scope_valid"])
        self.assertTrue(preview["post_restart_verification_required"])

    def test_reload_preview_drains_instead_of_restarting(self):
        preview = sanctioned_restart.build_restart_preview(
            NAMESPACE, sanctioned_restart.MODE_RELOAD,
            principal=admin(), env=READY_ENV,
        )
        steps = [entry["step"] for entry in preview["mutation_ledger"]]
        self.assertIn("host_graceful_reload", steps)
        self.assertNotIn("host_restart_hook", steps)

    def test_preview_never_enables_execution(self):
        preview = sanctioned_restart.build_restart_preview(
            NAMESPACE, principal=admin(), env=READY_ENV
        )
        self.assertFalse(preview["execution_enabled"])
        self.assertFalse(preview["authorization"]["execution_enabled"])

    def test_confirmation_phrase_binds_the_namespace(self):
        self.assertTrue(
            sanctioned_restart.confirmation_matches(
                NAMESPACE, sanctioned_restart.MODE_RESTART,
                "restart gitea-author",
            )
        )
        # A phrase typed for one namespace must not authorize another.
        self.assertFalse(
            sanctioned_restart.confirmation_matches(
                "gitea-merger", sanctioned_restart.MODE_RESTART,
                "restart gitea-author",
            )
        )


class TestGates(unittest.TestCase):
    """AC1/AC4: every gate denies with a stable reason code."""

    def _assess(self, **kwargs):
        params = {
            "principal": admin(),
            "confirmation": f"restart {NAMESPACE}",
            "env": READY_ENV,
        }
        params.update(kwargs)
        namespace = params.pop("namespace", NAMESPACE)
        mode = params.pop("mode", sanctioned_restart.MODE_RESTART)
        return sanctioned_restart.assess_restart_request(
            namespace, mode, **params
        )

    def test_authorized_confirmed_request_passes_every_gate(self):
        result = self._assess()
        self.assertTrue(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.ALLOW_HOST_ACTION_REQUIRED
        )

    def test_passing_every_gate_is_not_an_execution_grant(self):
        """An allowed request still never lets the console touch the process."""
        result = self._assess()
        self.assertTrue(result["allowed"])
        self.assertFalse(result["execution_enabled"])
        self.assertFalse(result["console_executes"])

    def test_unauthorized_principal_is_denied(self):
        result = self._assess(principal=viewer())
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
        )

    def test_anonymous_principal_is_denied(self):
        result = self._assess(principal=None)
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
        )

    def test_missing_confirmation_is_denied(self):
        result = self._assess(confirmation=None)
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.DENY_CONFIRMATION_MISSING
        )

    def test_confirmation_for_another_namespace_is_denied(self):
        result = self._assess(confirmation="restart gitea-merger")
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.DENY_CONFIRMATION_MISMATCH
        )

    def test_missing_operator_authorization_is_denied(self):
        env = {sanctioned_restart.RESTART_HOOK_ENV: "launchd:cc.prgs.author"}
        result = self._assess(env=env)
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"],
            sanctioned_restart.DENY_OPERATOR_AUTHORIZATION,
        )

    def test_missing_host_hook_is_denied_without_kill_fallback(self):
        env = {
            runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV: "ops-ticket-1",
        }
        result = self._assess(env=env)
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.DENY_HOOK_NOT_CONFIGURED
        )

    def test_fleet_scope_is_refused(self):
        for scope in ("all", "*", "fleet"):
            with self.subTest(scope=scope):
                result = self._assess(
                    namespace=scope, confirmation=f"restart {scope}"
                )
                self.assertFalse(result["allowed"])
                self.assertEqual(
                    result["reason_code"], sanctioned_restart.DENY_FLEET_SCOPE
                )

    def test_unknown_namespace_is_refused(self):
        result = self._assess(
            namespace="gitea-nope", confirmation="restart gitea-nope"
        )
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.DENY_UNKNOWN_NAMESPACE
        )

    def test_unknown_mode_is_refused(self):
        result = self._assess(mode="obliterate")
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.DENY_UNKNOWN_MODE
        )

    def test_live_contamination_marker_blocks_restart(self):
        marker = runtime_recovery_guard.build_contamination_record(
            reason_class=runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
            command_redacted="pkill -f mcp_server.py",
        )
        result = self._assess(contamination_marker=marker)
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["reason_code"], sanctioned_restart.DENY_CONTAMINATED_RUNTIME
        )

    def test_reconciler_cleared_marker_no_longer_blocks(self):
        marker = runtime_recovery_guard.build_contamination_record(
            reason_class=runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
            command_redacted="pkill -f mcp_server.py",
        )
        marker = dict(marker, cleared_by_reconciler=True)
        result = self._assess(contamination_marker=marker)
        self.assertTrue(result["allowed"])


class TestExecutionNeverKills(unittest.TestCase):
    """AC5: no path exposes or runs a raw process kill."""

    def test_authorized_execution_defers_to_the_host_supervisor(self):
        result = sanctioned_restart.execute_restart(
            NAMESPACE,
            principal=admin(),
            confirmation=f"restart {NAMESPACE}",
            env=READY_ENV,
        )
        self.assertTrue(result["allowed"])
        self.assertFalse(result["success"])
        self.assertFalse(result["process_kill_executed"])
        self.assertEqual(
            result["outcome"], sanctioned_restart.ALLOW_HOST_ACTION_REQUIRED
        )

    def test_denied_execution_reports_the_refusing_gate(self):
        result = sanctioned_restart.execute_restart(
            NAMESPACE, principal=viewer(), confirmation=f"restart {NAMESPACE}",
            env=READY_ENV,
        )
        self.assertFalse(result["allowed"])
        self.assertEqual(
            result["outcome"], sanctioned_restart.DENY_UNAUTHORIZED
        )
        self.assertFalse(result["process_kill_executed"])

    def test_module_never_spawns_a_process(self):
        path = os.path.join(
            os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
            "webui", "sanctioned_restart.py",
        )
        with open(path, encoding="utf-8") as handle:
            source = handle.read()
        for forbidden in (
            "import subprocess", "import signal", "os.kill", "os.system",
            "popen",
        ):
            with self.subTest(forbidden=forbidden):
                self.assertNotIn(forbidden, source.lower())

    def test_no_surface_returns_a_kill_command(self):
        payloads = [
            sanctioned_restart.build_restart_preview(
                NAMESPACE, principal=admin(), env=READY_ENV
            ),
            sanctioned_restart.restart_policy(),
            sanctioned_restart.execute_restart(
                NAMESPACE, principal=admin(),
                confirmation=f"restart {NAMESPACE}", env=READY_ENV,
            ),
        ]
        for payload in payloads:
            rendered = json.dumps(payload, default=str).lower()
            self.assertNotIn("kill -9", rendered)
            self.assertNotIn("pkill -f", rendered)

    def test_policy_declares_no_raw_kill_and_no_silent_restart(self):
        policy = sanctioned_restart.restart_policy()
        self.assertFalse(policy["raw_kill_exposed"])
        self.assertFalse(policy["console_executes_process_kill"])
        self.assertFalse(policy["fleet_scope_permitted"])
        self.assertFalse(policy["silent_auto_restart_permitted"])
        self.assertTrue(policy["audit_required"])


class TestContaminationClassification(unittest.TestCase):
    """AC2: manual pkill is contamination, and it blocks clean claims."""

    def test_manual_daemon_pkill_is_contamination(self):
        result = sanctioned_restart.classify_restart_command(
            "pkill -f mcp_server.py"
        )
        self.assertTrue(result["contamination"])
        self.assertFalse(result["clean_claim_allowed"])
        self.assertIsNotNone(result["contamination_marker"])
        self.assertEqual(
            result["sanctioned_alternative"],
            sanctioned_restart.ACTION_RESTART_NAMESPACE,
        )

    def test_broad_process_kill_is_contamination(self):
        result = sanctioned_restart.classify_restart_command("killall -9 Python")
        self.assertTrue(result["contamination"])
        self.assertFalse(result["clean_claim_allowed"])

    def test_marker_names_the_sanctioned_alternative(self):
        result = sanctioned_restart.classify_restart_command(
            "pkill -f mcp_server.py"
        )
        marker = result["contamination_marker"]
        self.assertIn(
            sanctioned_restart.ACTION_RESTART_NAMESPACE, marker["detail"]
        )

    def test_benign_command_is_not_contamination(self):
        result = sanctioned_restart.classify_restart_command("git status")
        self.assertFalse(result["contamination"])
        self.assertTrue(result["clean_claim_allowed"])

    def test_no_command_is_not_contamination(self):
        result = sanctioned_restart.classify_restart_command(None)
        self.assertFalse(result["contamination"])
        self.assertTrue(result["clean_claim_allowed"])


class TestPostRestartHealth(unittest.TestCase):
    """AC3: a clean post-restart claim needs live client-namespace proof."""

    def test_live_client_probe_clears_the_session(self):
        result = sanctioned_restart.verify_post_restart_health(
            NAMESPACE,
            probe_result={"success": True},
            probe_source=mcp_namespace_health.PROBE_SOURCE_CLIENT,
            registered_tools=["gitea_whoami"],
            required_tool="gitea_whoami",
        )
        self.assertEqual(result["status"], sanctioned_restart.HEALTH_CLEAN)
        self.assertTrue(result["clean_claim_allowed"])
        self.assertTrue(result["mutations_allowed"])

    def test_offline_probe_does_not_clear_the_session(self):
        result = sanctioned_restart.verify_post_restart_health(
            NAMESPACE,
            probe_result={"success": True},
            probe_source=mcp_namespace_health.PROBE_SOURCE_OFFLINE,
            registered_tools=["gitea_whoami"],
            required_tool="gitea_whoami",
        )
        self.assertFalse(result["clean_claim_allowed"])
        self.assertFalse(result["mutations_allowed"])

    def test_failed_probe_is_unhealthy(self):
        result = sanctioned_restart.verify_post_restart_health(
            NAMESPACE,
            probe_result={"success": False, "error": "client is closing: EOF"},
            probe_source=mcp_namespace_health.PROBE_SOURCE_CLIENT,
            registered_tools=["gitea_whoami"],
            required_tool="gitea_whoami",
        )
        self.assertEqual(result["status"], sanctioned_restart.HEALTH_UNHEALTHY)
        self.assertFalse(result["clean_claim_allowed"])

    def test_static_registration_alone_never_clears_the_session(self):
        result = sanctioned_restart.verify_post_restart_health(
            NAMESPACE,
            registered_tools=["gitea_whoami"],
            required_tool="gitea_whoami",
        )
        self.assertFalse(result["clean_claim_allowed"])


class TestAuditEmission(unittest.TestCase):
    """Every restart attempt is audited with actor, target, and result."""

    def _run(self, principal, sink):
        prior = os.environ.get(console_audit.AUDIT_LOG_ENV)
        os.environ[console_audit.AUDIT_LOG_ENV] = sink
        try:
            return sanctioned_restart.execute_restart(
                NAMESPACE,
                principal=principal,
                confirmation=f"restart {NAMESPACE}",
                env=READY_ENV,
                request_id="req-642",
            )
        finally:
            if prior is None:
                os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
            else:
                os.environ[console_audit.AUDIT_LOG_ENV] = prior

    def test_allowed_attempt_is_written_with_actor_and_target(self):
        with tempfile.TemporaryDirectory() as tmp:
            sink = os.path.join(tmp, "audit.jsonl")
            result = self._run(admin(), sink)
            self.assertTrue(result["audit"]["written"])
            with open(sink, encoding="utf-8") as handle:
                record = json.loads(handle.read().strip())
        self.assertEqual(
            record["action"], sanctioned_restart.ACTION_RESTART_NAMESPACE
        )
        self.assertEqual(record["target"]["namespace"], NAMESPACE)
        self.assertEqual(record["target"]["mode"], "restart")
        self.assertEqual(record["result"], console_audit.RESULT_ALLOWED)
        self.assertEqual(record["actor"]["subject"], "admin@example.test")
        self.assertFalse(record["metadata"]["process_kill_executed"])

    def test_denied_attempt_is_audited_too(self):
        with tempfile.TemporaryDirectory() as tmp:
            sink = os.path.join(tmp, "audit.jsonl")
            self._run(viewer(), sink)
            with open(sink, encoding="utf-8") as handle:
                record = json.loads(handle.read().strip())
        self.assertEqual(record["result"], console_audit.RESULT_DENIED)
        self.assertEqual(
            record["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
        )

    def test_restart_audit_uses_break_glass_retention(self):
        action = console_authz.get_action(
            sanctioned_restart.ACTION_RESTART_NAMESPACE
        )
        self.assertEqual(
            console_audit.retention_class_for(action),
            console_audit.RETENTION_BREAK_GLASS,
        )


class TestRegistrySurface(unittest.TestCase):
    """AC5: the console surfaces the control, still disabled, with no kill."""

    def test_registry_exposes_both_actions_disabled(self):
        registry = gated_actions.load_action_registry()
        for action_id in (
            sanctioned_restart.ACTION_RESTART_NAMESPACE,
            sanctioned_restart.ACTION_RELOAD_NAMESPACE,
        ):
            with self.subTest(action=action_id):
                action = registry.get(action_id)
                self.assertIsNotNone(action)
                self.assertFalse(action.enabled)

    def test_registry_preview_names_the_namespace_target(self):
        preview = gated_actions.preview_action(
            sanctioned_restart.ACTION_RESTART_NAMESPACE, namespace=NAMESPACE
        )
        target = preview["mutation_ledger"][0]["target"]
        self.assertIn(NAMESPACE, target)
        self.assertFalse(preview["enabled"])

    def test_registry_attempt_fails_closed(self):
        result = gated_actions.attempt_action(
            sanctioned_restart.ACTION_RESTART_NAMESPACE, namespace=NAMESPACE
        )
        self.assertFalse(result["success"])


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