"""Unit and integration tests for Phase 2 Web Console recovery controls (#644).""" from __future__ import annotations import os import sys import types import unittest from unittest.mock import patch from starlette.testclient import TestClient import merged_cleanup_reconcile import runtime_recovery_guard import stable_branch_push_guard import stale_binding_recovery from webui import console_authz, console_recovery, system_health from webui.app import create_app class TestConsoleRecovery(unittest.TestCase): def test_diagnose_recovery_healthy(self) -> None: diag = console_recovery.diagnose_recovery() self.assertIn(diag.status, {console_recovery.STATUS_HEALTHY, console_recovery.STATUS_ACTION_REQUIRED}) self.assertIsInstance(diag.playbooks, tuple) self.assertGreaterEqual(len(diag.playbooks), 4) playbook_ids = {pb.playbook_id for pb in diag.playbooks} self.assertIn(console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, playbook_ids) self.assertIn(console_recovery.PLAYBOOK_REBIND_SESSION, playbook_ids) self.assertIn(console_recovery.PLAYBOOK_RECONCILE_CLEANUPS, playbook_ids) self.assertIn(console_recovery.PLAYBOOK_SANCTIONED_RESTART, playbook_ids) def test_confirmation_phrase_generation_and_matching(self) -> None: phrase = console_recovery.confirmation_phrase("clear_stale_binding") self.assertEqual(phrase, "confirm clear_stale_binding") self.assertTrue(console_recovery.confirmation_matches("clear_stale_binding", "confirm clear_stale_binding")) self.assertFalse(console_recovery.confirmation_matches("clear_stale_binding", "wrong phrase")) phrase_target = console_recovery.confirmation_phrase("sanctioned_restart", "gitea-author") self.assertEqual(phrase_target, "confirm sanctioned_restart gitea-author") self.assertTrue(console_recovery.confirmation_matches("sanctioned_restart", "confirm sanctioned_restart gitea-author", "gitea-author")) def test_build_recovery_preview(self) -> None: principal = console_authz.Principal("dev@example.com", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True) preview = console_recovery.build_recovery_preview( playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, target="test-worktree", principal=principal, ) self.assertEqual(preview["playbook_id"], console_recovery.PLAYBOOK_CLEAR_STALE_BINDING) self.assertEqual(preview["action_id"], console_recovery.ACTION_CLEAR_STALE_BINDING) self.assertEqual(preview["confirmation_phrase"], "confirm clear_stale_binding test-worktree") self.assertTrue(len(preview["mutation_ledger"]) >= 3) self.assertTrue(preview["authorization"]["allowed"]) def test_build_recovery_preview_unknown_playbook(self) -> None: preview = console_recovery.build_recovery_preview("unknown_playbook") self.assertFalse(preview.get("allowed")) self.assertEqual(preview.get("error"), "unknown_playbook") def test_execute_recovery_playbook_confirmation_mismatch(self) -> None: # Authorization is checked before confirmation, so the phase gate has to # pass for this test to reach the branch it is about. principal = console_authz.Principal("dev@example.com", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True) with self._phase_two_enabled(): result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, confirmation="invalid confirmation", principal=principal, ) self.assertFalse(result["success"]) self.assertFalse(result["allowed"]) self.assertEqual(result["error"], "confirmation_mismatch") def test_execute_recovery_playbook_unauthorized(self) -> None: # Anonymous principal has viewer role -> should be denied result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, confirmation="confirm clear_stale_binding", principal=console_authz.ANONYMOUS, ) self.assertFalse(result["success"]) self.assertFalse(result["allowed"]) self.assertEqual(result["error"], console_authz.DENY_UNAUTHENTICATED) def test_execute_refuses_phase_two_write_while_console_is_phase_one(self) -> None: """B1: the apply path must arm the phase gate, not skip it. ``authorize`` only applies the phase branch when ``for_execution=True``. The apply path used the default, so an operator executed a phase-2 write while ``ACTIVE_PHASE`` was 1. """ self.assertGreater( console_authz.get_action(console_recovery.ACTION_CLEAR_STALE_BINDING).phase, console_authz.ACTIVE_PHASE, "fixture assumes the recovery actions are ahead of the active phase", ) principal = console_authz.Principal( "dev@example.com", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True ) phrase = console_recovery.confirmation_phrase( console_recovery.PLAYBOOK_CLEAR_STALE_BINDING ) result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, confirmation=phrase, principal=principal, ) self.assertFalse(result["success"]) self.assertFalse(result["allowed"]) self.assertEqual(result["error"], console_authz.DENY_PHASE_NOT_ACTIVE) def test_preview_execution_enabled_matches_the_execution_decision(self) -> None: """B1: preview must not report a bare False it cannot explain.""" principal = console_authz.Principal( "dev@example.com", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True ) preview = console_recovery.build_recovery_preview( playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION, target="branches/feat-issue-644", principal=principal, ) self.assertFalse(preview["execution_enabled"]) self.assertEqual( preview["execution_blocked_reason"], console_authz.DENY_PHASE_NOT_ACTIVE ) self.assertFalse(preview["execution_authorization"]["allowed"]) # The preview (non-execution) decision still allows, by role. self.assertTrue(preview["authorization"]["allowed"]) def _phase_two_enabled(self): """Raise ACTIVE_PHASE so the execution branches are reachable in tests.""" return patch.object(console_authz, "ACTIVE_PHASE", 2) def _operator(self) -> console_authz.Principal: return console_authz.Principal( "dev@example.com", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True ) def test_rebind_mutates_the_live_environment_not_a_copy(self) -> None: """B2: the playbook must change the mapping it claims to have changed.""" live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: "branches/stale-old"} phrase = console_recovery.confirmation_phrase( console_recovery.PLAYBOOK_REBIND_SESSION, "branches/feat-issue-644" ) with self._phase_two_enabled(): result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION, confirmation=phrase, target="branches/feat-issue-644", principal=self._operator(), env=live_env, ) self.assertTrue(result["success"]) self.assertEqual( live_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV], "branches/feat-issue-644", "rebind reported success without changing the caller's environment", ) self.assertTrue(result["applied_result"]["binding_changed"]) self.assertEqual(result["applied_result"]["binding_before"], "branches/stale-old") self.assertEqual( result["applied_result"]["binding_after"], "branches/feat-issue-644" ) def test_clear_stale_binding_reports_failure_when_nothing_changed(self) -> None: """B2: a no-op recovery must never be reported as success.""" live_env: dict[str, str] = {} phrase = console_recovery.confirmation_phrase( console_recovery.PLAYBOOK_CLEAR_STALE_BINDING ) with self._phase_two_enabled(): result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, confirmation=phrase, principal=self._operator(), env=live_env, ) self.assertFalse( result["success"], "a clear that changed no binding must not report success", ) self.assertFalse(result["applied_result"]["binding_changed"]) def test_clear_stale_binding_clears_the_live_binding(self) -> None: """B2: the sanctioned clear must reach the caller's environment.""" missing = "/nonexistent/branches/deleted-worktree" live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: missing} phrase = console_recovery.confirmation_phrase( console_recovery.PLAYBOOK_CLEAR_STALE_BINDING ) with self._phase_two_enabled(): result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, confirmation=phrase, principal=self._operator(), env=live_env, ) if result["success"]: self.assertNotIn(stale_binding_recovery.ACTIVE_WORKTREE_ENV, live_env) self.assertEqual(result["applied_result"]["binding_before"], missing) self.assertIsNone(result["applied_result"]["binding_after"]) else: # Fail closed is acceptable; reporting a clear that did not happen # is not. This is the invariant the blocker was about. self.assertFalse(result["applied_result"]["binding_changed"]) self.assertEqual( live_env.get(stale_binding_recovery.ACTIVE_WORKTREE_ENV), missing ) def test_reconcile_playbook_calls_an_entry_point_that_exists(self) -> None: """B3: the previous call named a function absent from the module.""" phrase = console_recovery.confirmation_phrase( console_recovery.PLAYBOOK_RECONCILE_CLEANUPS ) fake_server = types.SimpleNamespace( gitea_reconcile_merged_cleanups=lambda **kwargs: { "success": True, "entries": [{"issue_number": 100}], } ) with self._phase_two_enabled(), patch.dict( sys.modules, {"gitea_mcp_server": fake_server} ): result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_RECONCILE_CLEANUPS, confirmation=phrase, principal=console_authz.Principal( "dev@example.com", console_authz.ADMIN, console_authz.IDENTITY_LOCAL_DEV, True, ), ) self.assertTrue(result["success"], result.get("applied_result")) self.assertNotIn("error_type", result["applied_result"]) self.assertEqual(result["applied_result"]["reconciled_count"], 1) def test_reconcile_entry_point_exists_on_the_real_module(self) -> None: """B3 regression: guard the symbol itself, not just the call shape.""" import gitea_mcp_server self.assertTrue( hasattr(gitea_mcp_server, "gitea_reconcile_merged_cleanups"), "console recovery depends on this reconciler entry point", ) self.assertFalse( hasattr(merged_cleanup_reconcile, "reconcile_merged_cleanups"), "if this module grows the orchestrator, point the playbook back at it", ) def test_contamination_gate_blocks_a_writing_playbook(self) -> None: """B4: a live marker plus a gated task key must actually block.""" marker = { "kind": "manual_daemon_kill", "reason_class": "manual_daemon_kill", "command_summary": "pkill -f gitea_mcp_server", "active": True, } phrase = console_recovery.confirmation_phrase( console_recovery.PLAYBOOK_REBIND_SESSION, "branches/feat-issue-644" ) live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: "branches/stale-old"} with self._phase_two_enabled(), patch.object( console_recovery, "load_active_contamination_marker", return_value=marker ): result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION, confirmation=phrase, target="branches/feat-issue-644", principal=self._operator(), env=live_env, ) self.assertFalse(result["success"]) self.assertEqual(result["error"], "contaminated_runtime") self.assertEqual( live_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV], "branches/stale-old", "a blocked playbook must not have mutated anything", ) def test_contamination_gate_exempts_the_reconciler_remedy(self) -> None: """B4: the designated remedy must stay reachable while contaminated.""" marker = { "kind": "manual_daemon_kill", "reason_class": "manual_daemon_kill", "command_summary": "pkill -f gitea_mcp_server", "active": True, } phrase = console_recovery.confirmation_phrase( console_recovery.PLAYBOOK_RECONCILE_CLEANUPS ) fake_server = types.SimpleNamespace( gitea_reconcile_merged_cleanups=lambda **kwargs: { "success": True, "entries": [], } ) with self._phase_two_enabled(), patch.object( console_recovery, "load_active_contamination_marker", return_value=marker ), patch.dict(sys.modules, {"gitea_mcp_server": fake_server}): result = console_recovery.execute_recovery_playbook( playbook_id=console_recovery.PLAYBOOK_RECONCILE_CLEANUPS, confirmation=phrase, principal=console_authz.Principal( "dev@example.com", console_authz.ADMIN, console_authz.IDENTITY_LOCAL_DEV, True, ), ) self.assertNotEqual(result.get("error"), "contaminated_runtime") def test_gated_task_key_is_actually_gated(self) -> None: """B4: the console action id was never a member of the gated set.""" self.assertIn( console_recovery.CONTAMINATION_GATED_TASK, stable_branch_push_guard.CONTAMINATION_GATED_TASKS, ) self.assertNotIn( console_recovery.ACTION_CLEAR_STALE_BINDING, stable_branch_push_guard.CONTAMINATION_GATED_TASKS, ) def test_diagnosis_reads_the_key_the_gate_returns(self) -> None: """B4: ``contaminated`` is a key assess_contamination_gate never returns.""" gate = runtime_recovery_guard.assess_contamination_gate( None, task=console_recovery.CONTAMINATION_GATED_TASK, actual_role="operator" ) self.assertNotIn("contaminated", gate) self.assertIn("block", gate) def test_contaminated_runtime_is_reported_unclean(self) -> None: """B4: verify_post_recovery reported contamination_clean unconditionally.""" marker = { "kind": "manual_daemon_kill", "reason_class": "manual_daemon_kill", "command_summary": "pkill -f gitea_mcp_server", "active": True, } with patch.object( console_recovery, "load_active_contamination_marker", return_value=marker ): verification = console_recovery.verify_post_recovery() diag = console_recovery.diagnose_recovery() self.assertFalse(verification["contamination_clean"]) self.assertFalse(verification["clean"]) self.assertEqual(diag.status, console_recovery.STATUS_BLOCKED_CONTAMINATION) def test_master_parity_baseline_is_not_the_head_it_is_compared_against(self) -> None: """B5: capture_startup_parity was fed the head it was then compared to.""" stale = system_health.StaleRuntime( daemon_head="a" * 40, checkout_head="b" * 40, remote_head="b" * 40, stale=True, determinable=True, mutation_safe=False, reasons=("daemon is behind the checkout",), ) with patch.object(system_health, "assess_stale_runtime", return_value=stale): diag = console_recovery.diagnose_recovery() parity = diag.master_parity self.assertEqual(parity["startup_head"], "a" * 40) self.assertEqual(parity["current_head"], "b" * 40) self.assertNotEqual(parity["startup_head"], parity["current_head"]) self.assertFalse(parity["in_parity"]) def test_master_parity_carries_the_live_remote_dimension(self) -> None: """B5: live_remote_head was never passed, dropping the #610 dimension.""" stale = system_health.StaleRuntime( daemon_head="c" * 40, checkout_head="c" * 40, remote_head="d" * 40, stale=False, determinable=True, mutation_safe=False, reasons=(), ) with patch.object(system_health, "assess_stale_runtime", return_value=stale): diag = console_recovery.diagnose_recovery() self.assertEqual(diag.master_parity.get("live_remote_head"), "d" * 40) def test_verify_post_recovery(self) -> None: verification = console_recovery.verify_post_recovery() self.assertIn("clean", verification) self.assertIn("status", verification) self.assertIn("reasons", verification) def test_unverified_inherited_binding_is_not_reported_clean(self) -> None: """B2: ``not clear_eligible`` also read clean for unproven bindings.""" binding = { "classification": stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED, "clear_eligible": False, } diag = console_recovery.diagnose_recovery() patched = console_recovery.RecoveryDiagnosis( status=diag.status, clean=diag.clean, stale_runtime=diag.stale_runtime, master_parity=diag.master_parity, stale_binding=binding, contamination=diag.contamination, worktree_anomalies=diag.worktree_anomalies, playbooks=diag.playbooks, reasons=diag.reasons, ) with patch.object(console_recovery, "diagnose_recovery", return_value=patched): verification = console_recovery.verify_post_recovery() self.assertFalse(verification["binding_clean"]) self.assertEqual( verification["binding_classification"], stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED, ) class TestConsoleRecoveryApi(unittest.TestCase): def setUp(self) -> None: self.app = create_app() self.client = TestClient(self.app) def test_api_recovery_diagnose(self) -> None: res = self.client.get("/api/v1/system/recovery/diagnose") self.assertEqual(res.status_code, 200) data = res.json() self.assertIn("status", data) self.assertIn("clean", data) self.assertIn("playbooks", data) self.assertTrue(len(data["playbooks"]) >= 4) def test_api_recovery_preview(self) -> None: res = self.client.post( "/api/v1/system/recovery/preview", json={"playbook_id": "clear_stale_binding", "target": "active"}, ) self.assertEqual(res.status_code, 200) data = res.json() self.assertEqual(data["playbook_id"], "clear_stale_binding") self.assertEqual(data["confirmation_phrase"], "confirm clear_stale_binding active") self.assertIn("mutation_ledger", data) def test_api_recovery_apply_denied_without_auth(self) -> None: res = self.client.post( "/api/v1/system/recovery/apply", json={"playbook_id": "clear_stale_binding", "confirmation": "confirm clear_stale_binding"}, ) self.assertEqual(res.status_code, 400) data = res.json() self.assertFalse(data["success"]) self.assertFalse(data["allowed"]) def test_api_recovery_apply_refuses_phase_two_write_with_dev_auth(self) -> None: """B1: this previously asserted the phase-gate bypass as intended. An authenticated operator posting a valid confirmation still must not execute a phase-2 write while the console is in phase 1. The refusal is the contract; a 200 here means the gate is not armed. """ env = { "WEBUI_AUTH_MODE": "local_dev", "WEBUI_DEV_SUBJECT": "dev@example.com", "WEBUI_DEV_ROLE": "operator", } before = os.environ.get("GITEA_ACTIVE_WORKTREE") with patch.dict(os.environ, env): res = self.client.post( "/api/v1/system/recovery/apply", json={ "playbook_id": "rebind_session_worktree", "target": "branches/feat-issue-644", "confirmation": "confirm rebind_session_worktree branches/feat-issue-644", }, ) self.assertEqual(res.status_code, 400) data = res.json() self.assertFalse(data["success"]) self.assertFalse(data["allowed"]) self.assertEqual(data["error"], console_authz.DENY_PHASE_NOT_ACTIVE) self.assertEqual( os.environ.get("GITEA_ACTIVE_WORKTREE"), before, "a refused apply must not have rebound the live process environment", ) def test_api_recovery_preview_reports_why_execution_is_disabled(self) -> None: res = self.client.post( "/api/v1/system/recovery/preview", json={"playbook_id": "rebind_session_worktree", "target": "active"}, ) self.assertEqual(res.status_code, 200) data = res.json() self.assertFalse(data["execution_enabled"]) self.assertIn("execution_authorization", data) def test_api_recovery_verify(self) -> None: res = self.client.get("/api/v1/system/recovery/verify") self.assertEqual(res.status_code, 200) data = res.json() self.assertIn("clean", data) self.assertIn("status", data) if __name__ == "__main__": unittest.main()