140 lines
6.2 KiB
Python
140 lines
6.2 KiB
Python
"""Tests for Issue #686: Detect and reject manually launched duplicate MCP role servers."""
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
from datetime import datetime
|
|
|
|
import gitea_config
|
|
import gitea_mcp_server
|
|
import mcp_namespace_health
|
|
|
|
|
|
class TestIssue686ManualMcpProvenance(unittest.TestCase):
|
|
|
|
def test_client_managed_process_detection(self):
|
|
"""Test _is_client_managed_process correctly detects provenance markers."""
|
|
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
|
|
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
|
|
with patch.dict(os.environ, {"GITEA_MCP_CLIENT_MANAGED": "true"}, clear=True):
|
|
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
|
|
with patch.dict(os.environ, {"GITEA_SERVER_PROVENANCE": "client_managed"}, clear=True):
|
|
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
|
|
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
|
|
self.assertFalse(gitea_mcp_server._is_client_managed_process())
|
|
|
|
def test_unconsumed_gitea_env_overrides(self):
|
|
"""Test surfacing of unsupported GITEA_* env overrides (e.g. GITEA_DUMMY)."""
|
|
env = {
|
|
"GITEA_MCP_PROFILE": "prgs-author",
|
|
"GITEA_CLIENT_MANAGED": "1",
|
|
"GITEA_DUMMY": "2",
|
|
"GITEA_UNKNOWN_FLAG": "abc",
|
|
}
|
|
unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
|
|
self.assertIn("GITEA_DUMMY", unconsumed)
|
|
self.assertEqual(unconsumed["GITEA_DUMMY"], "2")
|
|
self.assertIn("GITEA_UNKNOWN_FLAG", unconsumed)
|
|
self.assertNotIn("GITEA_MCP_PROFILE", unconsumed)
|
|
self.assertNotIn("GITEA_CLIENT_MANAGED", unconsumed)
|
|
|
|
def test_manual_server_mutation_fail_closed(self):
|
|
"""AC 2: Mutating tools on a server without client-managed provenance fail closed with a typed blocker."""
|
|
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
|
|
block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
|
|
self.assertIsNotNone(block)
|
|
self.assertFalse(block["success"])
|
|
self.assertFalse(block["performed"])
|
|
self.assertEqual(block["blocker_kind"], "unsupported_manual_launch")
|
|
self.assertEqual(block["provenance"], "manual_launch")
|
|
self.assertTrue(any("mutation denied: server process was launched manually" in r for r in block["reasons"]))
|
|
self.assertIn("BLOCKED + RECONNECT", block["exact_next_action"])
|
|
|
|
def test_client_managed_server_mutation_passes_provenance_gate(self):
|
|
"""AC 3: Clean client-managed baseline passes the provenance gate."""
|
|
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
|
|
block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
|
|
self.assertIsNone(block)
|
|
|
|
@patch("subprocess.run")
|
|
@patch("os.path.getmtime")
|
|
@patch("os.path.exists")
|
|
@patch("os.getpid")
|
|
def test_manual_duplicate_does_not_mask_stale_runtime(
|
|
self, mock_getpid, mock_exists, mock_getmtime, mock_run
|
|
):
|
|
"""AC 1 & AC 3: Staleness detection ignores manual duplicates and reports stale supported runtimes."""
|
|
mock_getpid.return_value = 12345
|
|
mock_exists.return_value = True
|
|
|
|
code_time = datetime(2026, 7, 8, 14, 0, 0)
|
|
mock_getmtime.return_value = code_time.timestamp()
|
|
|
|
# PID 12345: stale client-managed process (started at 13:00)
|
|
# PID 99999: fresh manual duplicate process (started at 15:00, no GITEA_CLIENT_MANAGED)
|
|
ps_output = (
|
|
" PID LSTART COMMAND\n"
|
|
"12345 Wed Jul 8 13:00:00 2026 /path/to/python mcp_server.py\n"
|
|
"99999 Wed Jul 8 15:00:00 2026 /path/to/python mcp_server.py\n"
|
|
)
|
|
|
|
mock_run_ps = MagicMock()
|
|
mock_run_ps.stdout = ps_output
|
|
|
|
mock_env_12345 = MagicMock()
|
|
mock_env_12345.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
|
|
|
|
mock_env_99999 = MagicMock()
|
|
mock_env_99999.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_DUMMY=2"
|
|
|
|
def side_effect(args, **kwargs):
|
|
if args[0] == "ps" and "eww" in args:
|
|
pid = args[2]
|
|
if pid == "12345":
|
|
return mock_env_12345
|
|
elif pid == "99999":
|
|
return mock_env_99999
|
|
elif args[0] == "ps":
|
|
return mock_run_ps
|
|
raise ValueError(f"Unexpected args: {args}")
|
|
|
|
mock_run.side_effect = side_effect
|
|
|
|
reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
|
|
|
|
# Manual duplicate process must be flagged
|
|
self.assertTrue(any("Duplicate MCP server process(es) detected" in r for r in reasons))
|
|
# Unsupported env override (GITEA_DUMMY=2) must be flagged
|
|
self.assertTrue(any("unsupported-env: Unsupported GITEA_* environment variable override(s) detected: GITEA_DUMMY=2" in r for r in reasons))
|
|
# Stale runtime must NOT be masked by fresh manual process 99999!
|
|
self.assertTrue(any("All matching profiles for task 'create_issue' (['prgs-author']) are running but stale" in r for r in reasons))
|
|
|
|
def test_namespace_health_classification_includes_provenance(self):
|
|
"""AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env."""
|
|
process = {
|
|
"pid": 5555,
|
|
"profile": "prgs-author",
|
|
"env": {
|
|
"GITEA_MCP_PROFILE": "prgs-author",
|
|
"GITEA_DUMMY": "99",
|
|
},
|
|
}
|
|
res = mcp_namespace_health.classify_namespace_probe(
|
|
"gitea-author",
|
|
configured=True,
|
|
registered_tools=["gitea_whoami"],
|
|
probe_result={"success": True},
|
|
process=process,
|
|
probe_source="client_namespace",
|
|
)
|
|
self.assertEqual(res["provenance"], "manual_launch")
|
|
self.assertFalse(res["is_client_managed"])
|
|
self.assertEqual(res["unconsumed_gitea_env"], {"GITEA_DUMMY": "99"})
|
|
self.assertEqual(res["diagnostics"]["provenance"], "manual_launch")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|