"""Tests for sanctioned Codex MCP reconnect request surface (#678).""" from __future__ import annotations import os import unittest from unittest import mock import mcp_client_reconnect as mcr class NormalizeReasonTests(unittest.TestCase): def test_stale_runtime_aliases(self): self.assertEqual(mcr.normalize_reason("stale-runtime"), mcr.REASON_STALE_RUNTIME) self.assertEqual(mcr.normalize_reason("stale_runtime"), mcr.REASON_STALE_RUNTIME) self.assertEqual(mcr.normalize_reason("STALE"), mcr.REASON_STALE_RUNTIME) def test_transport_eof_aliases(self): self.assertEqual(mcr.normalize_reason("transport_eof"), mcr.REASON_TRANSPORT_EOF) self.assertEqual(mcr.normalize_reason("EOF"), mcr.REASON_TRANSPORT_EOF) self.assertEqual( mcr.normalize_reason("client_is_closing"), mcr.REASON_TRANSPORT_EOF ) def test_missing_namespace(self): self.assertEqual( mcr.normalize_reason("missing_namespace"), mcr.REASON_MISSING_NAMESPACE ) def test_empty_is_unspecified(self): self.assertEqual(mcr.normalize_reason(None), mcr.REASON_UNSPECIFIED) self.assertEqual(mcr.normalize_reason(""), mcr.REASON_UNSPECIFIED) class BoundaryClassificationTests(unittest.TestCase): def test_clean_when_shas_match(self): self.assertEqual( mcr.classify_boundary_status( startup_sha="abc", current_master_sha="abc" ), mcr.BOUNDARY_CLEAN, ) def test_mismatch_when_shas_differ(self): self.assertEqual( mcr.classify_boundary_status( startup_sha="aaa", current_master_sha="bbb" ), mcr.BOUNDARY_MISMATCH, ) def test_stale_when_live_stale(self): self.assertEqual( mcr.classify_boundary_status( startup_sha="aaa", current_master_sha="aaa", live_stale=True, ), mcr.BOUNDARY_STALE, ) class BuildReconnectRequestTests(unittest.TestCase): def test_stale_runtime_returns_typed_blocker_with_codex_steps(self): result = mcr.build_reconnect_request( namespace="gitea-author", profile="prgs-author", pid=1234, session_id="sess-1", startup_sha="aaa111", current_master_sha="bbb222", reason="stale-runtime", client="codex", restart_required=True, stop_required=True, ) self.assertTrue(result["success"]) self.assertTrue(result["read_only"]) self.assertFalse(result["reconnect_performed"]) self.assertFalse(result["mutation_performed"]) self.assertTrue(result["reconnect_needed"]) self.assertEqual(result["namespace"], "gitea-author") self.assertEqual(result["profile"], "prgs-author") self.assertEqual(result["pid"], 1234) self.assertEqual(result["session_id"], "sess-1") self.assertEqual(result["startup_sha"], "aaa111") self.assertEqual(result["current_master_sha"], "bbb222") self.assertEqual(result["boundary_status"], mcr.BOUNDARY_MISMATCH) self.assertEqual(result["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT) self.assertIsNotNone(result["typed_blocker"]) blocker = result["typed_blocker"] self.assertEqual(blocker["namespaces"], ["gitea-author"]) self.assertEqual(blocker["why_reconnect_required"], mcr.REASON_STALE_RUNTIME) self.assertTrue(any("Codex" in s or "Reload" in s for s in blocker["operator_ui_steps"])) self.assertIn("pkill", " ".join(result["forbidden_recovery_paths"]).lower()) self.assertTrue( mcr.reasons_never_suggest_forbidden(result["exact_safe_next_action"] or "") ) # Must not recommend forbidden recovery. for step in blocker["operator_ui_steps"]: self.assertTrue(mcr.reasons_never_suggest_forbidden(step), step) def test_transport_eof_typed_blocker(self): result = mcr.build_reconnect_request( namespace="gitea-reviewer", reason="transport_eof", client="claude_code", ) self.assertTrue(result["reconnect_needed"]) self.assertEqual(result["reason"], mcr.REASON_TRANSPORT_EOF) self.assertEqual(result["client"], "claude_code") steps = " ".join(result["operator_ui_steps"]).lower() self.assertIn("/mcp", steps) def test_missing_namespace_typed_blocker(self): result = mcr.build_reconnect_request( namespace="gitea-merger", reason="missing_namespace", client="codex", ) self.assertTrue(result["reconnect_needed"]) self.assertEqual(result["reason"], mcr.REASON_MISSING_NAMESPACE) self.assertEqual( result["typed_blocker"]["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT ) def test_healthy_not_required(self): result = mcr.build_reconnect_request( namespace="gitea-tools", startup_sha="deadbeef", current_master_sha="deadbeef", reason="not_required", client="codex", in_parity=True, restart_required=False, stop_required=False, ) self.assertFalse(result["reconnect_needed"]) self.assertEqual(result["blocker_kind"], mcr.BLOCKER_NONE) self.assertIsNone(result["typed_blocker"]) self.assertFalse(result["stop_required"]) self.assertFalse(result["restart_required"]) self.assertIn("not required", (result["exact_safe_next_action"] or "").lower()) def test_successful_reconnect_report_fields_present(self): """AC2: reconnect result reports required fields (even when needed).""" result = mcr.build_reconnect_request( namespace="gitea-controller", profile="prgs-controller", pid=99, session_id="sid", startup_sha="s" * 40, current_master_sha="c" * 40, reason="stale-runtime", ) for key in ( "namespace", "profile", "pid", "session_id", "startup_sha", "current_master_sha", "boundary_status", ): self.assertIn(key, result) self.assertIsNotNone(result[key], key) class ToolSurfaceTests(unittest.TestCase): """Exercise gitea_request_mcp_reconnect with a stubbed server context.""" def test_tool_is_registered_and_side_effect_free(self): import gitea_mcp_server as srv self.assertTrue(hasattr(srv, "gitea_request_mcp_reconnect")) with mock.patch.object(srv, "_profile_operation_gate", return_value=None): with mock.patch.object( srv, "get_profile", return_value={ "profile_name": "prgs-author", "role_kind": "author", "role": "author", }, ): with mock.patch.object( srv, "_current_master_parity", return_value={ "startup_head": "a" * 40, "current_head": "a" * 40, "daemon_start_head": "a" * 40, "local_head": "a" * 40, "in_parity": True, "stale": False, "restart_required": False, "determinable": True, "live_stale": False, "live_known": True, "reasons": [], }, ): with mock.patch.object( srv.master_parity_gate, "format_parity", return_value="in parity", ): with mock.patch.object( srv.role_namespace_gate, "infer_mcp_namespace", return_value="gitea-author", ): with mock.patch.object( srv.session_ctx, "mutation_context_audit_fields", return_value={"session_profile": "prgs-author"}, ): result = srv.gitea_request_mcp_reconnect( namespace="gitea-author", reason="not_required", client="codex", remote="prgs", ) self.assertTrue(result.get("success")) self.assertFalse(result.get("reconnect_performed")) self.assertFalse(result.get("mutation_performed")) self.assertEqual(result.get("namespace"), "gitea-author") self.assertEqual(result.get("profile"), "prgs-author") self.assertEqual(result.get("pid"), os.getpid()) self.assertIn("startup_sha", result) self.assertIn("current_master_sha", result) self.assertIn("boundary_status", result) self.assertTrue( mcr.reasons_never_suggest_forbidden( result.get("exact_safe_next_action") or "" ) ) def test_tool_stale_returns_typed_blocker(self): import gitea_mcp_server as srv with mock.patch.object(srv, "_profile_operation_gate", return_value=None): with mock.patch.object( srv, "get_profile", return_value={ "profile_name": "prgs-reconciler", "role_kind": "reconciler", "role": "reconciler", }, ): with mock.patch.object( srv, "_current_master_parity", return_value={ "startup_head": "a" * 40, "current_head": "b" * 40, "daemon_start_head": "a" * 40, "local_head": "b" * 40, "in_parity": False, "stale": True, "restart_required": True, "determinable": True, "live_stale": True, "live_known": True, "reasons": ["stale"], }, ): with mock.patch.object( srv.master_parity_gate, "format_parity", return_value="stale", ): with mock.patch.object( srv.role_namespace_gate, "infer_mcp_namespace", return_value="gitea-reconciler", ): with mock.patch.object( srv.session_ctx, "mutation_context_audit_fields", return_value={}, ): result = srv.gitea_request_mcp_reconnect( reason="stale-runtime", client="codex", ) self.assertTrue(result["reconnect_needed"]) self.assertEqual( result["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT ) self.assertIsNotNone(result["typed_blocker"]) self.assertIn("gitea-reconciler", result["typed_blocker"]["namespaces"]) self.assertTrue(result["stop_required"]) self.assertTrue(result["restart_required"]) self.assertTrue( mcr.reasons_never_suggest_forbidden( result.get("exact_safe_next_action") or "" ) ) class InventoryRegistrationTests(unittest.TestCase): def test_reconnect_path_in_restart_inventory(self): import mcp_restart_paths as mrp ids = {p.path_id for p in mrp.iter_restart_paths()} self.assertIn("codex_client_reconnect_request", ids) self.assertIn("ide_client_reconnect", ids) def test_tool_name_in_documented_inventory(self): import mcp_tool_inventory as inv doc_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), inv.INVENTORY_DOC_PATH ) with open(doc_path, encoding="utf-8") as handle: documented = inv.parse_documented_inventory(handle.read()) self.assertIn("gitea_request_mcp_reconnect", documented) if __name__ == "__main__": unittest.main()