Gitea-Tools/tests/test_issue_686_manual_mcp_provenance.py

"""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.

        #948 narrowed the vocabulary here. This process carries no client-managed
        declaration, so the old code labelled it ``manual_launch`` — asserting a
        hand-launched terminal process it had no evidence for, and contradicting
        ``gitea_get_runtime_context``, which read the same process and reported
        ``client_managed``. Absence of proof is now reported as ``unproven``.

        The #686 wall itself is unchanged and still asserted below:
        ``is_client_managed`` stays False, so nothing previously refused is now
        permitted. Only the label on the *reason* changed, so remediation names
        the proof that is actually missing.
        """
        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"], "unproven")
        self.assertFalse(res["is_client_managed"])
        # The wall is intact: no client-managed proof still fails closed.
        self.assertTrue(res["provenance_fail_closed"])
        self.assertEqual(res["unconsumed_gitea_env"], {"GITEA_DUMMY": "99"})
        self.assertEqual(res["diagnostics"]["provenance"], "unproven")


if __name__ == "__main__":
    unittest.main()