"""Regression tests for Issue #708 wiring: detection must reach a gate, not just exist. The prior #708 slice added a pure decision function with no call site, so a session whose namespaces were Connected-but-unattached still passed every mutation gate. These tests pin the parts that make the detection load-bearing: * the typed condition is distinct from #672 / #584 / #685, * the session store records a per-namespace attachment verdict, * review/merge mutations fail closed while a required namespace is unattached, * recovery is reconnect-only and never suggests an unsafe fallback, * startup ordering races and discovery-cache telemetry are reported, * a healthy final report cannot be produced without attachment proof. """ import mcp_namespace_health REQUIRED = ["gitea-author", "gitea-reviewer", "gitea-merger", "gitea-tools"] def _connected_but_unattached(): """The #708 signature: host says Connected, session tool surface is empty.""" return mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=[], required_namespaces=REQUIRED, ) # --- AC1: distinct typed detection ----------------------------------------- def test_connected_with_empty_session_tool_list_is_typed_distinctly(): res = _connected_but_unattached() assert res["attachment_healthy"] is False assert res["discovery_status"] == "connected_but_namespaces_missing" assert res["error_type"] == "mcp_connected_namespaces_missing" assert sorted(res["missing_namespaces"]) == sorted(REQUIRED) # Not misclassified as config drift (#672), transport-closed (#584), or EOF (#685). assert res["error_type"] not in { "mcp_config_drift", "transport_closed", "mcp_client_eof", } def test_proof_carries_connected_and_attached_per_namespace(): res = _connected_but_unattached() proof = res["proof_of_connected_vs_attached"] for ns in REQUIRED: assert proof[ns] == {"connected": True, "attached": False} # --- AC2: recovery is auto-attach or reconnect-only ------------------------- def test_exact_next_action_is_reconnect_only(): res = _connected_but_unattached() assert res["reconnect_required"] is True action = res["exact_next_action"] assert "Reconnect the IDE/client MCP session" in action for forbidden in ("pkill", "chmod", "curl", ".env", "sys.path"): assert forbidden not in action def test_sanctioned_recovery_tool_is_named(): res = _connected_but_unattached() assert res["sanctioned_recovery_tool"] == "gitea_request_mcp_reconnect" assert "gitea_request_mcp_reconnect" in " ".join(res["remediation"]) def test_successful_auto_attach_reports_recovered_without_operator(): res = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=REQUIRED, required_namespaces=REQUIRED, auto_attach_attempted=True, auto_attach_succeeded=True, ) assert res["attachment_healthy"] is True assert res["auto_recovered"] is True assert res["reconnect_required"] is False def test_failed_auto_attach_still_requires_reconnect(): res = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=[], required_namespaces=REQUIRED, auto_attach_attempted=True, auto_attach_succeeded=False, ) assert res["auto_recovered"] is False assert res["reconnect_required"] is True assert any("did not succeed" in r for r in res["reasons"]) def test_reconnect_rediscovery_clears_the_condition(): """Attach state after a reconnect is healthy without any other change.""" before = _connected_but_unattached() after = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=REQUIRED, required_namespaces=REQUIRED, discovery_cache_hit=False, discovery_cache_age_seconds=0.0, ) assert before["attachment_healthy"] is False assert after["attachment_healthy"] is True assert after["discovery_status"] == "namespaces_attached" assert after["missing_namespaces"] == [] # --- AC4: startup ordering across multiple role servers --------------------- def test_multi_role_startup_ordering_race_is_reported(): res = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=["gitea-author"], required_namespaces=REQUIRED, session_tool_snapshot_at=1000.0, namespace_connected_at={ "gitea-author": 990.0, "gitea-reviewer": 1005.0, "gitea-merger": 1007.0, }, ) assert res["startup_ordering_race"] is True assert res["late_attaching_namespaces"] == ["gitea-merger", "gitea-reviewer"] assert res["telemetry"]["startup_ordering_race"] is True def test_no_ordering_race_when_snapshot_follows_connect(): res = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=REQUIRED, required_namespaces=REQUIRED, session_tool_snapshot_at=2000.0, namespace_connected_at={ns: 1000.0 for ns in REQUIRED}, ) assert res["startup_ordering_race"] is False assert res["late_attaching_namespaces"] == [] # --- AC5: telemetry --------------------------------------------------------- def test_telemetry_reports_cache_and_recovery_signals(): res = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=["gitea-author"], required_namespaces=REQUIRED, discovery_cache_hit=True, discovery_cache_age_seconds=42.5, ) tel = res["telemetry"] assert tel["connected_count"] == 4 assert tel["attached_count"] == 1 assert tel["missing_count"] == 3 assert tel["discovery_cache_hit"] is True assert tel["discovery_cache_age_seconds"] == 42.5 assert tel["reconnect_required"] is True assert tel["error_type"] == "mcp_connected_namespaces_missing" def test_telemetry_leaks_no_secrets(): res = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=[], required_namespaces=REQUIRED, ) blob = repr(res["telemetry"]).lower() for leak in ("token", "authorization", "password", "secret", "/users/", "http"): assert leak not in blob # --- AC2/AC6: fail-closed mutation gate ------------------------------------- def _session_store_from(assessment): """Mirror the server-side recorder without importing the MCP server module.""" store = {} missing = set(assessment["missing_namespaces"]) for ns, state in assessment["proof_of_connected_vs_attached"].items(): attached = bool(state["attached"]) store[ns] = { "namespace": ns, "connected": bool(state["connected"]), "attached": attached, "attachment_healthy": attached and ns not in missing, "error_type": None if attached else assessment["error_type"], } return store def test_review_and_merge_fail_closed_while_unattached(): store = _session_store_from(_connected_but_unattached()) for task in ("review_pr", "submit_review", "merge_pr", "create_pr", "work_issue"): reasons = mcp_namespace_health.attachment_gate_from_session(task, store) assert reasons, f"{task} must fail closed while its namespace is unattached" assert "fail closed, #708" in reasons[0] def test_gate_offers_only_sanctioned_recovery(): store = _session_store_from(_connected_but_unattached()) reasons = mcp_namespace_health.attachment_gate_from_session("merge_pr", store) joined = " ".join(reasons) assert "reconnect" in joined.lower() assert "Workflow Safety Hard Stop (#708)" in joined # Unsafe fallbacks appear only inside the prohibition, never as advice. assert "NEVER use" in joined def test_gate_passes_once_namespaces_are_attached(): healthy = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=REQUIRED, required_namespaces=REQUIRED, ) store = _session_store_from(healthy) for task in ("review_pr", "submit_review", "merge_pr", "create_pr", "work_issue"): assert mcp_namespace_health.attachment_gate_from_session(task, store) == [] def test_unassessed_session_does_not_gate(): """No recorded assessment must not fabricate a block (matches #543 semantics).""" assert mcp_namespace_health.attachment_gate_from_session("merge_pr", {}) == [] assert mcp_namespace_health.attachment_gate_from_session("merge_pr", None) == [] def test_unmapped_task_is_not_gated(): store = _session_store_from(_connected_but_unattached()) assert mcp_namespace_health.attachment_gate_from_session("gitea_read", store) == [] def test_partial_attachment_gates_only_the_affected_role(): """Author attached, reviewer not: author work proceeds, review fails closed.""" res = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=["gitea-author", "gitea-tools"], required_namespaces=REQUIRED, ) store = _session_store_from(res) assert mcp_namespace_health.attachment_gate_from_session("work_issue", store) == [] assert mcp_namespace_health.attachment_gate_from_session("review_pr", store) assert mcp_namespace_health.attachment_gate_from_session("merge_pr", store) # --- AC4: no false healthy report without attachment proof ------------------ def test_no_false_healthy_without_attachment_proof(): """Connected alone never yields a healthy verdict.""" res = mcp_namespace_health.assess_connected_namespace_attachment( connected_servers=REQUIRED, attached_session_namespaces=None, required_namespaces=REQUIRED, ) assert res["success"] is False assert res["attachment_healthy"] is False assert res["telemetry"]["attached_count"] == 0 def test_attachment_gate_maps_each_role_namespace(): assert mcp_namespace_health.required_namespace_for_attachment("review_pr") == "gitea-reviewer" assert mcp_namespace_health.required_namespace_for_attachment("merge_pr") == "gitea-merger" assert mcp_namespace_health.required_namespace_for_attachment("create_pr") == "gitea-author" assert mcp_namespace_health.required_namespace_for_attachment("nope") is None