Gitea-Tools/tests/test_mcp_restart_paths.py

"""Tests for the MCP restart-path inventory and guards (#657).

Covers:
* the registry is well-formed and every path is classified;
* unknown restart attempts fail closed (AC "fail closed on unknown restart");
* the previously-unguarded full-restart primitives stay guarded/absent
  against the real source tree (AC "tests for at least one previously
  unguarded path");
* pkill of the daemon is still classified as contamination (#630, AC3);
* the inventory doc and module stay in lock-step.
"""

import os
import tempfile
import unittest
from pathlib import Path

import mcp_restart_paths as rp
import runtime_recovery_guard

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DOC_PATH = os.path.join(REPO_ROOT, "docs", "mcp-restart-path-inventory.md")


class TestRegistryWellformed(unittest.TestCase):
    def test_registry_is_wellformed(self):
        # Must not raise.
        rp.assert_registry_wellformed()

    def test_every_path_has_valid_classification(self):
        for path in rp.iter_restart_paths():
            self.assertIn(path.classification, rp.VALID_CLASSIFICATIONS)
            self.assertTrue(path.guard.strip(), path.path_id)
            self.assertTrue(path.references, path.path_id)
            self.assertTrue(path.locations, path.path_id)

    def test_ids_are_unique(self):
        ids = [p.path_id for p in rp.iter_restart_paths()]
        self.assertEqual(len(ids), len(set(ids)))

    def test_covers_every_classification(self):
        present = {p.classification for p in rp.iter_restart_paths()}
        self.assertEqual(present, set(rp.VALID_CLASSIFICATIONS))


class TestUnknownAttemptFailsClosed(unittest.TestCase):
    def test_unknown_path_raises(self):
        with self.assertRaises(rp.UnknownRestartPathError):
            rp.assert_restart_attempt_registered("totally_novel_restart_hack")

    def test_get_unknown_raises(self):
        with self.assertRaises(rp.UnknownRestartPathError):
            rp.get_restart_path("nope")

    def test_registered_attempt_returns_path(self):
        path = rp.assert_restart_attempt_registered("manual_daemon_kill")
        self.assertEqual(path.classification, rp.CLASS_FORBIDDEN)


class TestDaemonNeverSelfReplaces(unittest.TestCase):
    """Previously-unguarded full-restart primitive: daemon self-replacement."""

    def test_no_self_replacement_in_source(self):
        # The live daemon modules must contain no os.execv/os.kill/os._exit
        # self-restart call. Must not raise.
        rp.assert_no_daemon_self_replacement(REPO_ROOT)

    def test_scanner_flags_injected_violation(self):
        # Guard the guard: prove the scanner catches a real self-replace call.
        with tempfile.TemporaryDirectory() as tmp:
            bad = Path(tmp) / "gitea_mcp_server.py"
            bad.write_text(
                "import os\n"
                "def restart():\n"
                "    os.execv('/usr/bin/python', ['python'])\n",
                encoding="utf-8",
            )
            found = rp.scan_daemon_self_replacement(tmp)
            self.assertTrue(found)
            with self.assertRaises(AssertionError):
                rp.assert_no_daemon_self_replacement(tmp)

    def test_scanner_ignores_comment_and_docstring_mentions(self):
        with tempfile.TemporaryDirectory() as tmp:
            ok = Path(tmp) / "gitea_mcp_server.py"
            ok.write_text(
                "import os\n"
                "# NOT os.execv() to re-point the interpreter here.\n"
                '"""Never calls os._exit to restart."""\n'
                "value = 1\n",
                encoding="utf-8",
            )
            self.assertEqual(rp.scan_daemon_self_replacement(tmp), [])


class TestLegacyAutoRestartHelperRemoved(unittest.TestCase):
    """Previously-unguarded full-restart path: _trigger_mcp_auto_restart."""

    def test_helper_absent_in_source(self):
        # Must not raise: helper was removed in #685.
        rp.assert_auto_restart_helper_absent(REPO_ROOT)

    def test_scanner_flags_reintroduced_helper(self):
        with tempfile.TemporaryDirectory() as tmp:
            bad = Path(tmp) / "mcp_server.py"
            bad.write_text(
                "def _trigger_mcp_auto_restart():\n    return True\n",
                encoding="utf-8",
            )
            with self.assertRaises(AssertionError):
                rp.assert_auto_restart_helper_absent(tmp)


class TestPkillStaysForbidden(unittest.TestCase):
    """AC3: pkill of the daemon remains forbidden/contaminating (#630)."""

    def test_manual_daemon_kill_registered_as_forbidden(self):
        path = rp.get_restart_path("manual_daemon_kill")
        self.assertEqual(path.classification, rp.CLASS_FORBIDDEN)

    def test_pkill_classified_as_contamination(self):
        assessment = runtime_recovery_guard.assess_recovery_command(
            "pkill -f mcp_server.py"
        )
        self.assertTrue(assessment["contaminated"])

    def test_read_only_probe_not_contamination(self):
        assessment = runtime_recovery_guard.assess_recovery_command(
            "ps aux | grep mcp_server"
        )
        self.assertFalse(assessment["contaminated"])


class TestInventoryDocInSync(unittest.TestCase):
    def test_doc_exists(self):
        self.assertTrue(os.path.exists(DOC_PATH), DOC_PATH)

    def test_doc_mentions_every_path_id(self):
        with open(DOC_PATH, encoding="utf-8") as handle:
            doc = handle.read()
        for path in rp.iter_restart_paths():
            self.assertIn(path.path_id, doc, f"doc missing {path.path_id}")


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