"""Tests for emergency break-glass MCP restart workflow (#664). Regression suite for review #641 remediation: B13, B1, B14, B6/B11, B8, and preservation of previously accepted B2/B3/B5/B7/B9/B10 corrections. """ from __future__ import annotations import os import unittest from unittest.mock import MagicMock, patch import gitea_audit import gitea_config import gitea_mcp_server def _controller_profile(**extra) -> dict: base = { "profile_name": "prgs-controller", "execution_profile": "prgs-controller", "role": "reconciler", # declared role must not be redefined by capability "allowed_operations": [ "gitea.read", "gitea.issue.create", "gitea.branch.delete", "gitea.pr.close", "gitea.pr.comment", "gitea.issue.comment", "runtime.break_glass_restart", ], "forbidden_operations": [ "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.create", "gitea.branch.push", ], } base.update(extra) return base def _gate_open_patches(): """Keep master-parity / runtime-mode blocks out of unit tests.""" return ( patch.object(gitea_mcp_server, "_master_parity_block", return_value=[]), patch.object(gitea_mcp_server, "_runtime_mode_block", return_value=[]), patch.object(gitea_mcp_server, "_try_auto_switch_for_operation", return_value=False), ) class TestBreakGlassRestart(unittest.TestCase): """Test suite for gitea_break_glass_restart tool and guardrails (#664).""" def setUp(self) -> None: self.env_patcher = patch.dict(os.environ, {}, clear=False) self.env_patcher.start() os.environ.pop("GITEA_BREAKGLASS_RESTART_AUTHORIZATION", None) os.environ.pop("GITEA_SANCTIONED_RESTART_HOOK", None) os.environ.pop("GITEA_AUDIT_LOG", None) # Reset injectable executor between tests. gitea_mcp_server._break_glass_restart_executor = None def tearDown(self) -> None: gitea_mcp_server._break_glass_restart_executor = None self.env_patcher.stop() # ── B13: operation registration / real gate ─────────────────────────── def test_normalize_operation_accepts_canonical_break_glass_op(self) -> None: """B13: production normalizer accepts exact runtime.break_glass_restart.""" self.assertEqual( gitea_config.normalize_operation( "runtime.break_glass_restart", service="runtime" ), "runtime.break_glass_restart", ) ok, reason = gitea_config.check_operation( "runtime.break_glass_restart", ["gitea.read", "runtime.break_glass_restart"], ) self.assertTrue(ok, reason) self.assertEqual(reason, "allowed") def test_normalize_unknown_and_misspelled_ops_fail_closed(self) -> None: """B13: unknown / misspelled operations fail closed (no gitea.read fallback).""" # Well-formed but misspelled runtime op normalizes, then is not allowed. ok, reason = gitea_config.check_operation( "runtime.break_glass_restar", # misspelled ["gitea.read", "runtime.break_glass_restart"], ) self.assertFalse(ok) self.assertEqual(reason, "not-allowed") ok2, reason2 = gitea_config.check_operation( "runtime.break_glass_restart", ["gitea.read"], # capability not granted ) self.assertFalse(ok2) self.assertEqual(reason2, "not-allowed") ok3, reason3 = gitea_config.check_operation( "frobnicate", ["gitea.read", "runtime.break_glass_restart"], ) self.assertFalse(ok3) self.assertEqual(reason3, "invalid-operation") # gitea.read grant alone never authorizes break-glass. ok4, reason4 = gitea_config.check_operation( "runtime.break_glass_restart", ["gitea.read"], ) self.assertFalse(ok4) self.assertNotEqual(reason4, "allowed") def test_real_profile_operation_gate_without_stubbing(self) -> None: """B13: exercise real _profile_operation_gate (not stubbed).""" allowed = _controller_profile() denied = _controller_profile( allowed_operations=["gitea.read", "gitea.issue.create"] ) p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=allowed): self.assertEqual( gitea_mcp_server._profile_operation_gate( "runtime.break_glass_restart" ), [], ) with patch.object(gitea_mcp_server, "get_profile", return_value=denied): reasons = gitea_mcp_server._profile_operation_gate( "runtime.break_glass_restart" ) self.assertTrue(reasons) self.assertTrue( any("runtime.break_glass_restart" in r or "not allowed" in r for r in reasons) ) def test_entry_point_uses_same_operation_as_gate(self) -> None: """B13: entry point enforces runtime.break_glass_restart, not gitea.read.""" # Profile has gitea.read but not the break-glass capability. prof = _controller_profile( allowed_operations=["gitea.read", "gitea.issue.create"] ) p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart required due to deadlock in worker pool", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "permission_denied") self.assertNotIn("gitea.read", " ".join(res.get("reasons") or [])) # ── B1 / B14: exact profile auth, no substring, role preservation ───── def test_trusted_prgs_controller_authorized(self) -> None: """B1: exact trusted prgs-controller with capability is authorized.""" prof = _controller_profile() p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart required due to deadlock in worker pool", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=True, remote="prgs", ) self.assertTrue(res["success"], res) self.assertFalse(res["break_glass_executed"]) def test_fabricated_controller_like_profile_names_denied(self) -> None: """B1: lookalike profile names never become authorized.""" p_parity, p_runtime, p_switch = _gate_open_patches() for name in ( "fake-controller", "controller-copy", "not-controller", "xcontrollerx", "CONTROLLER", "prgs-controller-copy", ): prof = _controller_profile(profile_name=name, execution_profile=name) with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart required due to deadlock in worker pool", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=True, remote="prgs", ) self.assertFalse(res["success"], name) self.assertEqual(res["blocker_kind"], "role_authorization", name) self.assertFalse(res["break_glass_executed"]) def test_ordinary_and_non_controller_reconciler_denied(self) -> None: """B1: ordinary roles and non-controller reconcilers are denied.""" p_parity, p_runtime, p_switch = _gate_open_patches() denied = [ {"profile_name": "prgs-author", "role": "author", "allowed_operations": ["gitea.read", "runtime.break_glass_restart"], "forbidden_operations": []}, {"profile_name": "prgs-reviewer", "role": "reviewer", "allowed_operations": ["gitea.read", "runtime.break_glass_restart"], "forbidden_operations": []}, {"profile_name": "prgs-merger", "role": "merger", "allowed_operations": ["gitea.read", "runtime.break_glass_restart"], "forbidden_operations": []}, {"profile_name": "prgs-reconciler", "role": "reconciler", "allowed_operations": [ "gitea.read", "gitea.branch.delete", "runtime.break_glass_restart" ], "forbidden_operations": []}, {"profile_name": "prgs-controller", "role": "reconciler", "allowed_operations": ["gitea.read"], "forbidden_operations": []}, # no capability ] for prof in denied: with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart required due to deadlock in worker pool", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=True, remote="prgs", ) self.assertFalse(res["success"], prof) self.assertFalse(res["break_glass_executed"], prof) self.assertIn( res["blocker_kind"], ("role_authorization", "permission_denied"), prof, ) def test_env_var_cannot_grant_authorization_or_bypass_denial(self) -> None: """B2 preserved: env var cannot grant break-glass authorization.""" os.environ["GITEA_BREAKGLASS_RESTART_AUTHORIZATION"] = "secret-bypass-token" prof = { "profile_name": "prgs-author", "role": "author", "allowed_operations": [ "gitea.read", "gitea.issue.create", "runtime.break_glass_restart" ], "forbidden_operations": [], } p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart attempting env var bypass", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "role_authorization") def test_profile_role_kind_no_substring_authority(self) -> None: """B1/B14: substring lookalikes do not become controller.""" for name in ( "fake-controller", "controller-copy", "not-controller", "xcontrollerx", "myCONTROLLER", ): prof = { "profile_name": name, "role": "author", "allowed_operations": ["gitea.read"], } self.assertEqual( gitea_mcp_server._profile_role_kind(prof), "author", name, ) # Role substrings must not promote. for role in ("not-controller", "control-plane", "xcontrollerx"): prof = {"profile_name": "other", "role": role, "allowed_operations": ["gitea.read"]} self.assertEqual( gitea_mcp_server._profile_role_kind(prof), role, role, ) def test_prgs_controller_retains_declared_reconciler_role(self) -> None: """B14: break-glass capability does not redefine global role.""" prof = _controller_profile(role="reconciler") self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler") # Declared controller still wins when declared. prof2 = _controller_profile(role="controller") self.assertEqual(gitea_mcp_server._profile_role_kind(prof2), "controller") def test_cleanup_merged_pr_branch_role_resolution_unchanged(self) -> None: """B14: prgs-controller with reconciler role still resolves for cleanup.""" # When declared reconciler, cleanup gate's role check should see reconciler. prof = _controller_profile(role="reconciler") self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler") # Fabricated controller-like names with reconciler ops do not become controller. fake = { "profile_name": "fake-controller", "role": "reconciler", "allowed_operations": [ "gitea.read", "gitea.branch.delete", "gitea.pr.close" ], } self.assertEqual(gitea_mcp_server._profile_role_kind(fake), "reconciler") def test_narrow_break_glass_capability_does_not_redefine_role(self) -> None: """B14: granting runtime.break_glass_restart does not invent controller role.""" prof = { "profile_name": "prgs-reconciler", "role": "reconciler", "allowed_operations": [ "gitea.read", "gitea.branch.delete", "runtime.break_glass_restart", ], } self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler") # ── B2-style input validation (preserved) ───────────────────────────── def test_reason_validation(self) -> None: """AC2: Reason is required and must be at least 10 characters long.""" prof = _controller_profile() p_parity, p_runtime, p_switch = _gate_open_patches() for invalid_reason in ["", " ", "too short", "123456789"]: with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof): res = gitea_mcp_server.gitea_break_glass_restart( reason=invalid_reason, confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertEqual(res["blocker_kind"], "missing_required_fields") def test_confirmation_validation(self) -> None: """AC2: Confirmation phrase must match exact required string.""" prof = _controller_profile() p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart needed due to stuck daemon processes", confirmation="wrong_confirmation_phrase", impact_ack=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertEqual(res["blocker_kind"], "confirmation_mismatch") def test_impact_ack_validation(self) -> None: """AC2: impact_ack=True is mandatory.""" prof = _controller_profile() p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart needed due to stuck daemon processes", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=False, remote="prgs", ) self.assertFalse(res["success"]) self.assertEqual(res["blocker_kind"], "impact_ack_required") def test_incident_permission_gate(self) -> None: """B3 preserved: Gate incident creation on gitea.issue.create permission.""" prof = _controller_profile( allowed_operations=["gitea.read", "runtime.break_glass_restart"] ) p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart needed due to hung worker process cohort", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, create_incident_issue=True, dry_run=False, remote="prgs", ) self.assertFalse(res["success"]) self.assertEqual(res["blocker_kind"], "permission_denied") # ── B8: redaction ───────────────────────────────────────────────────── def test_redaction_key_value_and_connection_strings(self) -> None: """B8: key/value, bearer, connection-string, and embedded secrets.""" cases = [ ("password=hunter2supersecret", ["hunter2supersecret"], "password"), ("api_key: sk-live-abcdef1234567890ab", ["sk-live-abcdef1234567890ab"], "api_key"), ("Server=db;Password=s3cretValue;Uid=sa", ["s3cretValue"], "password"), ( "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.abc.def", ["eyJhbGciOiJIUzI1NiJ9.abc.def", "Bearer eyJ"], "authorization", ), ( "token ghp_1234567890abcdef12345678 embedded", ["ghp_1234567890abcdef12345678"], "token", ), ("nested note password=letmein12345 end", ["letmein12345"], "password"), ] for raw, secrets, key in cases: out = gitea_audit._redact_str(raw) self.assertIn("[REDACTED]", out, raw) for secret in secrets: self.assertNotIn(secret, out, raw) self.assertIn(key, out.lower(), raw) nested = gitea_audit.redact({ "reason": "password=supersecret99", "items": [{"api_key": "abc123xyz"}, "token ghp_abcdefghijklmnop1234"], "error": RuntimeError("pwd=nestedSecret99"), }) # Exception objects pass through redact as non-str/non-container; ensure # string forms are covered via str conversion in _redact_str usage. self.assertEqual(nested["items"][0]["api_key"], gitea_audit.REDACTED) self.assertNotIn("supersecret99", nested["reason"]) self.assertNotIn("ghp_abcdefghijklmnop1234", nested["items"][1]) def test_redaction_preserves_benign_sec_prefix_text(self) -> None: """B8: ordinary text beginning with sec- must not be erased.""" benign = ( "Emergency restart in sec-primary-region for sector-planning " "and secondary-health checks" ) out = gitea_audit._redact_str(benign) self.assertIn("sec-primary-region", out) self.assertIn("sector-planning", out) self.assertIn("secondary-health", out) self.assertNotIn("[REDACTED]", out) def test_redaction_across_break_glass_surfaces(self) -> None: """B8: operator reason is redacted on result, incident, and audit surfaces.""" prof = _controller_profile() raw_reason = ( "Emergency restart: password=supersecret99 api_key: " "sk-live-abcdef1234567890ab and url https://user:pass@gitea.prgs.cc/api" ) mock_api_request = MagicMock(return_value={"number": 101, "title": "[INCIDENT]"}) fake_exec = { "success": True, "apply_supported": True, "apply_authorized": True, "restart_performed": True, "break_glass_executed": True, } gitea_mcp_server._break_glass_restart_executor = lambda req: fake_exec p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "api_request", mock_api_request ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ), patch.object( gitea_mcp_server, "gitea_reconcile_after_restart", return_value={"success": True}, ), patch.object( gitea_audit, "audit_enabled", return_value=True ), patch.object( gitea_audit, "write_event", return_value=True ): res = gitea_mcp_server.gitea_break_glass_restart( reason=raw_reason, confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) self.assertTrue(res["success"], res) self.assertNotIn("supersecret99", res["reason"]) self.assertNotIn("sk-live-abcdef1234567890ab", res["reason"]) self.assertNotIn("user:pass@gitea.prgs.cc", res["reason"]) posted_body = mock_api_request.call_args[0][3]["body"] self.assertNotIn("supersecret99", posted_body) self.assertNotIn("sk-live-abcdef1234567890ab", posted_body) # ── B6/B11: reachable executor / truthful flags ─────────────────────── def test_dry_run_never_executes_and_reports_false(self) -> None: """B7/B6: dry-run never executes; break_glass_executed always false.""" prof = _controller_profile() called = {"n": 0} def _should_not_run(_req): called["n"] += 1 return {"restart_performed": True, "break_glass_executed": True} gitea_mcp_server._break_glass_restart_executor = _should_not_run mock_audit = MagicMock() mock_api = MagicMock() p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": [{"session_id": "s1"}]}, ), patch.object( gitea_audit, "write_event", mock_audit ), patch.object( gitea_mcp_server, "api_request", mock_api ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart preview in dry-run mode", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=True, remote="prgs", ) self.assertTrue(res["success"]) self.assertTrue(res["dry_run"]) self.assertFalse(res["break_glass_executed"]) self.assertTrue(res["would_execute"]) self.assertEqual(called["n"], 0) mock_audit.assert_not_called() mock_api.assert_not_called() def test_unsupported_apply_truthful(self) -> None: """B6/B11: unsupported apply returns blocked result, execution false.""" prof = _controller_profile() gitea_mcp_server._break_glass_restart_executor = lambda req: { "success": False, "apply_supported": False, "apply_authorized": False, "restart_performed": False, "break_glass_executed": False, "reasons": ["no hook"], } p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ), patch.object( gitea_audit, "audit_enabled", return_value=True ), patch.object( gitea_audit, "write_event", return_value=True ), patch.object( gitea_mcp_server, "api_request", return_value={"number": 555, "title": "[INCIDENT]"}, ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Privileged restart request with unsupported apply", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["performed"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "apply_unsupported") def test_delegation_success_rejection_and_failure(self) -> None: """B6/B11: distinct terminal states for success / rejection / failure.""" prof = _controller_profile() p_parity, p_runtime, p_switch = _gate_open_patches() def _run(exec_result, recon=None): gitea_mcp_server._break_glass_restart_executor = lambda req: exec_result with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ), patch.object( gitea_mcp_server, "gitea_reconcile_after_restart", return_value=recon or {"success": True}, ), patch.object( gitea_audit, "audit_enabled", return_value=True ), patch.object( gitea_audit, "write_event", return_value=True ), patch.object( gitea_mcp_server, "api_request", return_value={"number": 555, "title": "[INCIDENT]"}, ): return gitea_mcp_server.gitea_break_glass_restart( reason="Privileged break-glass restart delegation path", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) ok = _run({ "success": True, "apply_supported": True, "apply_authorized": True, "restart_performed": True, "break_glass_executed": True, }) self.assertTrue(ok["success"], ok) self.assertTrue(ok["break_glass_executed"]) self.assertTrue(ok["performed"]) rejected = _run({ "success": False, "apply_supported": True, "apply_authorized": False, "restart_performed": False, "break_glass_executed": False, "reasons": ["class denied"], }) self.assertFalse(rejected["success"]) self.assertFalse(rejected["break_glass_executed"]) self.assertEqual(rejected["blocker_kind"], "restart_delegation_failed") failed = _run({ "success": False, "apply_supported": True, "apply_authorized": True, "restart_performed": False, "break_glass_executed": False, "reasons": ["executor error"], }) self.assertFalse(failed["success"]) self.assertFalse(failed["break_glass_executed"]) self.assertEqual(failed["blocker_kind"], "restart_delegation_failed") def test_reconciliation_success_and_failure_truthful_flags(self) -> None: """B10 preserved: recon failure keeps break_glass_executed=true.""" prof = _controller_profile() gitea_mcp_server._break_glass_restart_executor = lambda req: { "success": True, "apply_supported": True, "apply_authorized": True, "restart_performed": True, "break_glass_executed": True, } p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ), patch.object( gitea_mcp_server, "gitea_reconcile_after_restart", return_value={"success": False, "error": "lease cleanup failed"}, ), patch.object( gitea_audit, "audit_enabled", return_value=True ), patch.object( gitea_audit, "write_event", return_value=True ), patch.object( gitea_mcp_server, "api_request", return_value={"number": 555, "title": "[INCIDENT]"}, ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Privileged restart request with failing reconciliation", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertTrue(res["performed"]) self.assertTrue(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "reconciliation_failed") def test_authorization_is_not_execution(self) -> None: """B6: apply_authorized alone never sets break_glass_executed.""" # Default executor without hook → unsupported, not executed. prof = _controller_profile() p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ), patch.object( gitea_audit, "audit_enabled", return_value=True ), patch.object( gitea_audit, "write_event", return_value=True ), patch.object( gitea_mcp_server, "api_request", return_value={"number": 555, "title": "[INCIDENT]"}, ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Privileged restart without host hook configured", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "apply_unsupported") # ── preserved fail-closed pre-exec (B5/B9) ──────────────────────────── def test_audit_failure_before_execution_fails_closed(self) -> None: """B9 preserved: Audit recording failure stops execution fail-closed.""" prof = _controller_profile() called = {"n": 0} gitea_mcp_server._break_glass_restart_executor = lambda req: called.__setitem__("n", called["n"] + 1) or {} p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ), patch.object( gitea_audit, "audit_enabled", return_value=True ), patch.object( gitea_audit, "write_event", return_value=False ), patch.object( gitea_mcp_server, "api_request", MagicMock() ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart with failing audit sink", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "audit_recording_failed") self.assertEqual(called["n"], 0) def test_incident_creation_failure_before_execution_fails_closed(self) -> None: """B5 preserved: Incident creation failure stops execution fail-closed.""" prof = _controller_profile() called = {"n": 0} gitea_mcp_server._break_glass_restart_executor = lambda req: called.__setitem__("n", called["n"] + 1) or {} p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ), patch.object( gitea_audit, "audit_enabled", return_value=True ), patch.object( gitea_audit, "write_event", return_value=True ), patch.object( gitea_mcp_server, "api_request", side_effect=RuntimeError("Gitea 500 API Error"), ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart with failing incident POST", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "incident_creation_failed") self.assertEqual(called["n"], 0) def test_incident_opt_out_on_real_execution_fails_closed(self) -> None: """B5 preserved: create_incident_issue=False fails closed on real execution.""" prof = _controller_profile() p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart trying to skip incident creation", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=False, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "incident_creation_required") def test_audit_disabled_fails_closed(self) -> None: """B9 preserved: Disabling audit recording blocks execution fail-closed.""" prof = _controller_profile() p_parity, p_runtime, p_switch = _gate_open_patches() with p_parity, p_runtime, p_switch: with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object( gitea_audit, "audit_enabled", return_value=False ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}, ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart with disabled audit logging", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "audit_recording_failed") def test_capability_map_registration(self) -> None: """B12/B13: capability map registers exact runtime.break_glass_restart.""" from task_capability_map import TASK_CAPABILITY_MAP entry = TASK_CAPABILITY_MAP.get("gitea_break_glass_restart") self.assertIsNotNone(entry) self.assertEqual(entry["permission"], "runtime.break_glass_restart") self.assertEqual(entry["role"], "controller") entry2 = TASK_CAPABILITY_MAP.get("break_glass_restart") self.assertEqual(entry2["permission"], "runtime.break_glass_restart") if __name__ == "__main__": unittest.main()