import unittest import gitea_mcp_server import mcp_namespace_health import review_merge_state_machine as rmsm class TestMcpNamespaceHealth(unittest.TestCase): def setUp(self): gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear() def test_registered_tool_but_live_eof_blocks_merge(self): result = mcp_namespace_health.classify_namespace_probe( "gitea-reviewer", required_tool="gitea_whoami", registered_tools=["gitea_whoami", "gitea_list_profiles"], probe_result={ "success": False, "error": "client is closing: EOF", }, process={ "pid": 4321, "profile": "prgs-reviewer", "env": { "GITEA_MCP_PROFILE": "prgs-reviewer", "GITEA_TOKEN": "must-not-leak", }, }, config_path="/Users/jasonwalker/.gemini/config/mcp_config.json", probe_source="client_namespace", ) self.assertFalse(result["healthy"]) self.assertEqual(result["error_type"], "namespace_eof") self.assertTrue(result["required_tool_registered"]) self.assertFalse(result["required_tool_callable"]) self.assertTrue(result["blocks_merge_workflow"]) self.assertFalse(result["ide_namespace_proven"]) self.assertEqual(result["probe_source"], "client_namespace") self.assertEqual(result["diagnostics"]["process_pid"], 4321) self.assertEqual(result["diagnostics"]["profile"], "prgs-reviewer") self.assertEqual( result["diagnostics"]["env"]["GITEA_MCP_PROFILE"], "prgs-reviewer", ) self.assertNotIn("GITEA_TOKEN", result["diagnostics"]["env"]) self.assertIn("gitea-reviewer", result["reasons"][0]) self.assertIn("gitea_whoami", result["reasons"][0]) def test_successful_client_namespace_invocation_is_ide_proven(self): result = mcp_namespace_health.classify_namespace_probe( "gitea-author", registered_tools=["gitea_whoami"], probe_result={"success": True, "result": {"authenticated": True}}, process={"pid": 1234, "profile": "prgs-author"}, config_path="/tmp/mcp_config.json", probe_source="client_namespace", ) self.assertTrue(result["healthy"]) self.assertTrue(result["required_tool_registered"]) self.assertTrue(result["required_tool_callable"]) self.assertTrue(result["ide_namespace_proven"]) self.assertFalse(result["blocks_merge_workflow"]) self.assertIsNone(result["error_type"]) def test_offline_spawn_success_is_not_ide_proof(self): result = mcp_namespace_health.classify_namespace_probe( "gitea-merger", registered_tools=["gitea_whoami"], probe_result={"success": True, "result": {}}, process={"pid": 99, "profile": "prgs-merger"}, probe_source="offline_spawn", ) self.assertTrue(result["healthy"]) self.assertFalse(result["ide_namespace_proven"]) self.assertFalse(result["blocks_merge_workflow"]) self.assertTrue( any("offline_spawn" in r for r in result["reasons"]) ) def test_registered_missing_required_tool_fails_before_probe_success(self): result = mcp_namespace_health.classify_namespace_probe( "gitea-tools", registered_tools=["gitea_whoami"], probe_result={"success": True, "result": {}}, probe_source="client_namespace", ) self.assertFalse(result["healthy"]) self.assertEqual(result["required_tool"], "gitea_list_profiles") self.assertFalse(result["required_tool_registered"]) self.assertEqual(result["error_type"], "tool_missing") def test_server_tool_exposes_same_assessment_and_records_session(self): result = gitea_mcp_server.gitea_assess_mcp_namespace_health( "gitea-merger", registered_tools=["gitea_whoami"], probe_result={"success": False, "error": "transport closed"}, process={"pid": 9876, "profile": "prgs-merger"}, probe_source="client_namespace", ) self.assertFalse(result["success"]) self.assertEqual(result["error_type"], "namespace_eof") self.assertEqual(result["namespace"], "gitea-merger") self.assertEqual(result["diagnostics"]["process_pid"], 9876) self.assertIn("gitea-merger", gitea_mcp_server._LIVE_NAMESPACE_HEALTH) gate = gitea_mcp_server._live_namespace_health_gate("merge_pr") self.assertTrue(gate) self.assertTrue(any("gitea-merger" in r for r in gate)) class TestLiveNamespaceBlocksMerge(unittest.TestCase): """AC5: a broken live namespace must hard-block the review/merge state machine.""" def setUp(self): gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear() def _merge_ready_completion(self): # Completion through PRE_MERGE_RECHECK is the state where a clean merge # is otherwise allowed (see test_merge_allowed_with_pre_merge_gates). idx = rmsm.REVIEW_MERGE_STATES.index("PRE_MERGE_RECHECK") return {state: True for state in rmsm.REVIEW_MERGE_STATES[: idx + 1]} def _all_gates(self): return {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES} def test_assess_workflow_blockers_flags_live_namespace_broken(self): clean = rmsm.assess_workflow_blockers() self.assertFalse(clean["block"]) broken = rmsm.assess_workflow_blockers(live_namespace_broken=True) self.assertTrue(broken["block"]) self.assertTrue(any("namespace" in r.lower() for r in broken["reasons"])) def test_can_merge_blocks_even_when_all_gates_pass(self): # Without the namespace blocker a fully-complete workflow can merge... allowed = rmsm.can_merge( self._merge_ready_completion(), pre_merge_gates=self._all_gates() ) self.assertTrue(allowed["allowed"]) # ...but a broken live namespace overrides every satisfied gate. blocked = rmsm.can_merge( self._merge_ready_completion(), pre_merge_gates=self._all_gates(), live_namespace_broken=True, ) self.assertTrue(blocked["block"]) self.assertFalse(blocked["allowed"]) def test_workflow_status_forwards_namespace_blocker(self): status = rmsm.workflow_status( self._merge_ready_completion(), live_namespace_broken=True ) self.assertFalse(status["merge_allowed"]) self.assertFalse(status["approve_allowed"]) def test_server_tool_blocks_merge_on_live_namespace_broken(self): result = gitea_mcp_server.gitea_assess_review_merge_state_machine( state_completion=self._merge_ready_completion(), pre_merge_gates=self._all_gates(), live_namespace_broken=True, ) self.assertTrue(result["blockers"]["block"]) self.assertFalse(result["merge"]["allowed"]) def test_classify_verdict_bridges_into_merge_block(self): # The classify verdict is the intended feed for live_namespace_broken. verdict = mcp_namespace_health.classify_namespace_probe( "gitea-merger", registered_tools=["gitea_whoami", "gitea_adopt_merger_pr_lease"], probe_result={"success": False, "error": "client is closing: EOF"}, process={"pid": 555, "profile": "prgs-merger"}, probe_source="client_namespace", ) self.assertTrue(verdict["blocks_merge_workflow"]) blocked = rmsm.can_merge( self._merge_ready_completion(), pre_merge_gates=self._all_gates(), live_namespace_broken=verdict["blocks_merge_workflow"], ) self.assertFalse(blocked["allowed"]) def test_offline_spawn_does_not_authorize_mutation_gate(self): gitea_mcp_server.gitea_assess_mcp_namespace_health( "gitea-merger", registered_tools=["gitea_whoami"], probe_result={"success": True, "result": {}}, probe_source="offline_spawn", ) gate = gitea_mcp_server._live_namespace_health_gate("merge_pr") self.assertTrue(gate) self.assertTrue(any("offline_spawn" in r or "client_namespace" in r for r in gate)) def test_client_namespace_healthy_clears_mutation_gate(self): gitea_mcp_server.gitea_assess_mcp_namespace_health( "gitea-merger", registered_tools=["gitea_whoami"], probe_result={"success": True, "result": {}}, probe_source="client_namespace", ) self.assertEqual(gitea_mcp_server._live_namespace_health_gate("merge_pr"), []) if __name__ == "__main__": unittest.main()