"""Tests for emergency break-glass MCP restart workflow (#664).""" from __future__ import annotations import os import unittest from unittest.mock import MagicMock, patch import gitea_mcp_server 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() def tearDown(self) -> None: self.env_patcher.stop() def test_ordinary_role_denied_fail_closed(self) -> None: """AC1: Ordinary LLM roles (author, reviewer, merger, reconciler) cannot invoke break-glass.""" with patch.object( gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-author", "allowed_operations": ["gitea.read"]} ), patch.object( gitea_mcp_server, "_profile_role_kind", return_value="author" ), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency restart needed due to deadlocked worker daemon processes", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, remote="prgs", ) self.assertFalse(res["success"]) self.assertFalse(res["break_glass_executed"]) self.assertEqual(res["blocker_kind"], "role_authorization") self.assertIn("ordinary LLM role 'author' is forbidden", res["reasons"][0]) def test_short_reason_denied(self) -> None: """AC2: Reason is required and must be at least 10 characters long.""" with patch.object( gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} ), patch.object( gitea_mcp_server, "_profile_role_kind", return_value="controller" ), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Too short", 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") self.assertIn("at least 10 characters", res["reasons"][0]) def test_confirmation_mismatch_denied(self) -> None: """AC2: Confirmation phrase must match exact required string.""" with patch.object( gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} ), patch.object( gitea_mcp_server, "_profile_role_kind", return_value="controller" ), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ): 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") self.assertIn("confirmation string mismatch", res["reasons"][0]) def test_impact_ack_required_denied(self) -> None: """AC2: impact_ack=True is mandatory.""" with patch.object( gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} ), patch.object( gitea_mcp_server, "_profile_role_kind", return_value="controller" ), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ): 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_dry_run_evaluation(self) -> None: """AC5: Dry-run evaluation returns preview without live execution or incident creation.""" with patch.object( gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} ), patch.object( gitea_mcp_server, "_profile_role_kind", return_value="controller" ), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": [{"session_id": "s1"}]} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ): 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, 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.assertTrue(res["reconciliation_required"]) self.assertEqual(res["reconciliation_tool"], "gitea_reconcile_after_restart") self.assertIn("#664", res["cross_references"]) def test_privileged_execute_creates_incident_and_audit(self) -> None: """AC3 & AC4: Execution creates incident issue, audit entry, and mandates post-restart reconcile.""" mock_api_request = MagicMock(return_value={"number": 999, "title": "[INCIDENT] Break-glass"}) mock_save_state = MagicMock(return_value={"saved": True}) with patch.object( gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]} ), patch.object( gitea_mcp_server, "_profile_role_kind", return_value="controller" ), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ), patch.object( gitea_mcp_server, "_auth", return_value={"Authorization": "token test"} ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": [{"session_id": "s1"}]} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="sysadmin" ), patch.object( gitea_mcp_server, "api_request", mock_api_request ), patch("mcp_session_state.save_state", mock_save_state): res = gitea_mcp_server.gitea_break_glass_restart( reason="Emergency break-glass restart due to unrecoverable transport deadlock", confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION", impact_ack=True, dry_run=False, create_incident_issue=True, remote="prgs", ) self.assertTrue(res["success"]) self.assertFalse(res["dry_run"]) self.assertTrue(res["break_glass_executed"]) self.assertTrue(res["reconciliation_required"]) self.assertEqual(res["reconciliation_tool"], "gitea_reconcile_after_restart") self.assertEqual(res["incident_issue"]["number"], 999) mock_save_state.assert_called_once() mock_api_request.assert_called_once() def test_env_authorization_override_for_worker_role(self) -> None: """Environment break-glass authorization enables privileged break-glass for configured sessions.""" os.environ["GITEA_BREAKGLASS_RESTART_AUTHORIZATION"] = "authorized-token" with patch.object( gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-author", "allowed_operations": ["gitea.read"]} ), patch.object( gitea_mcp_server, "_profile_role_kind", return_value="author" ), patch.object( gitea_mcp_server, "_profile_operation_gate", return_value=None ), patch.object( gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []} ), patch.object( gitea_mcp_server, "_authenticated_username", return_value="jcwalker3" ): res = gitea_mcp_server.gitea_break_glass_restart( reason="Authorized emergency break-glass restart test", 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"]) if __name__ == "__main__": unittest.main()