Gitea-Tools/tests/test_fleet_inventory_tool.py

"""Native exposure of the fleet inventory capability (#949).

Covers the parts the pure classifier cannot: that the tool is registered and
documented, that it is gated on ``gitea.read`` rather than a role, that it
accepts no caller-supplied evidence, that it fails closed on an unreadable
registry, and that the workflow documentation explains gate consumption.
"""

import inspect
import os
import pathlib
import unittest
from unittest import mock

import gitea_mcp_server
import mcp_fleet_inventory as mfi
import task_capability_map


REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
TOOL_NAME = "gitea_assess_fleet_inventory"


class TestCapabilityMapExposure(unittest.TestCase):
    """AC14: the invariant is provable through sanctioned native calls."""

    def test_task_resolves_to_a_read_permission(self):
        self.assertEqual(
            task_capability_map.required_permission("assess_fleet_inventory"),
            "gitea.read",
        )
        self.assertEqual(
            task_capability_map.required_permission(TOOL_NAME), "gitea.read"
        )

    def test_task_and_tool_keys_agree(self):
        self.assertEqual(
            task_capability_map.TASK_CAPABILITY_MAP["assess_fleet_inventory"],
            task_capability_map.TASK_CAPABILITY_MAP[TOOL_NAME],
        )

    def test_task_is_not_role_exclusive(self):
        """Controller and reconciler both need it; so does any read-only gate."""
        self.assertNotIn(
            "assess_fleet_inventory", task_capability_map.ROLE_EXCLUSIVE_TASKS
        )
        self.assertNotIn(TOOL_NAME, task_capability_map.ROLE_EXCLUSIVE_TASKS)

    def test_task_is_not_registered_as_an_issue_mutation(self):
        self.assertNotIn(TOOL_NAME, task_capability_map.ISSUE_MUTATION_TOOL_TASKS)

    def test_unknown_task_still_fails_closed(self):
        with self.assertRaises(KeyError):
            task_capability_map.required_permission("assess_fleet_inventory_typo")


class TestToolRegistration(unittest.TestCase):
    def test_tool_is_registered_on_the_server(self):
        self.assertTrue(hasattr(gitea_mcp_server, TOOL_NAME))

    def test_tool_accepts_no_caller_supplied_evidence(self):
        """The defect #949 names: evidence parameters the caller controls."""
        signature = inspect.signature(gitea_mcp_server.gitea_assess_fleet_inventory)
        self.assertEqual(sorted(signature.parameters), ["host", "org", "remote", "repo"])
        for forbidden in ("process", "probe_result", "registered_tools", "processes"):
            self.assertNotIn(forbidden, signature.parameters)

    def test_tool_is_documented_in_the_canonical_inventory(self):
        doc = (REPO_ROOT / "docs" / "mcp-tool-inventory.md").read_text()
        self.assertIn(f"`{TOOL_NAME}`", doc)

    def test_docstring_states_the_read_only_guarantee(self):
        doc = (gitea_mcp_server.gitea_assess_fleet_inventory.__doc__ or "").lower()
        self.assertIn("read-only", doc)
        self.assertIn("fails closed", doc)


class TestNamespaceResolution(unittest.TestCase):
    """Every configured profile must resolve to its real fleet namespace.

    ``role_namespace_gate.infer_mcp_namespace`` recognises only author and
    reviewer and echoes the profile name for the rest, which would label the
    controller, merger, and reconciler members with namespaces that do not
    exist — and then flag all three as role mismatches.
    """

    def test_every_expected_profile_maps_to_its_namespace(self):
        for entry in mfi.EXPECTED_PRGS_FLEET:
            self.assertEqual(
                mfi.namespace_for_profile(entry["profile"]),
                entry["namespace"],
                entry["profile"],
            )

    def test_server_helper_agrees_with_the_roster(self):
        for entry in mfi.EXPECTED_PRGS_FLEET:
            self.assertEqual(
                gitea_mcp_server._fleet_namespace_for_profile(entry["profile"]),
                entry["namespace"],
                entry["profile"],
            )

    def test_unknown_profile_falls_back_to_the_supplied_default(self):
        self.assertEqual(
            mfi.namespace_for_profile("prgs-shadow", default="gitea-shadow"),
            "gitea-shadow",
        )

    def test_unknown_profile_without_a_default_returns_the_profile(self):
        self.assertEqual(mfi.namespace_for_profile("prgs-shadow"), "prgs-shadow")

    def test_registered_row_uses_the_roster_namespace(self):
        fake_db = mock.Mock()
        with mock.patch.object(
            gitea_mcp_server, "_control_plane_db_or_error", return_value=(fake_db, [])
        ), mock.patch.object(
            gitea_mcp_server, "get_profile", return_value={"allowed_operations": []}
        ), mock.patch.object(
            gitea_mcp_server.gitea_config,
            "selected_profile_name",
            return_value="prgs-reconciler",
        ):
            record = gitea_mcp_server._register_fleet_runtime(transport="stdio")
        self.assertEqual(record["namespace"], "gitea-reconciler")


class TestToolBehavior(unittest.TestCase):
    """The tool wires the registry and the process scan into the classifier."""

    @staticmethod
    def _healthy_rows():
        return [
            {
                "runtime_id": f"{entry['namespace']}:{43000 + index}:boot",
                "namespace": entry["namespace"],
                "profile": entry["profile"],
                "role": entry["role"],
                "remote": "prgs",
                "org": "Scaled-Tech-Consulting",
                "repo": "Gitea-Tools",
                "repository_root": "/checkout/Gitea-Tools",
                "pid": 43000 + index,
                "cohort_id": "ppid:40990",
                "cohort_source": "parent_process",
                "client_provenance": "client_managed",
                "boot_id": "boot",
                "startup_head": "82d71b77028a7abd4f8ab4a4e4d89658a187f73d",
                "daemon_start_head": "82d71b77028a7abd4f8ab4a4e4d89658a187f73d",
                "transport": "stdio",
                "registered_at": "2026-07-28T02:00:00Z",
                "last_heartbeat_at": "2026-07-28T02:00:00Z",
                "status": "running",
            }
            for index, entry in enumerate(mfi.EXPECTED_PRGS_FLEET)
        ]

    def _call(self, rows, *, scan=None, db=None, db_errors=None):
        fake_db = db
        if fake_db is None and db_errors is None:
            fake_db = mock.Mock()
            fake_db.list_mcp_server_runtimes.return_value = rows
        scan = scan or {
            "available": True,
            "processes": [
                {"pid": r["pid"], "started_at": "2026-07-28T01:00:00Z"} for r in rows
            ],
            "reason": None,
        }
        with mock.patch.object(
            gitea_mcp_server, "_profile_operation_gate", return_value=[]
        ), mock.patch.object(
            gitea_mcp_server,
            "_control_plane_db_or_error",
            return_value=(fake_db, db_errors or []),
        ), mock.patch.object(
            gitea_mcp_server, "_active_profile_name", return_value="prgs-controller"
        ), mock.patch.object(
            mfi, "scan_mcp_server_processes", return_value=scan
        ), mock.patch.object(
            mfi, "probe_pid_alive", return_value=True
        ):
            return gitea_mcp_server.gitea_assess_fleet_inventory(
                remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools"
            )

    def test_healthy_fleet_satisfies_the_gate_through_the_tool(self):
        result = self._call(self._healthy_rows())
        self.assertTrue(result["inventory_complete"])
        self.assertTrue(result["mutation_gate_satisfied"])
        self.assertEqual(result["running_member_count"], 5)
        self.assertEqual(result["mutations_performed"], [])

    def test_tool_reports_the_expected_repository_binding(self):
        result = self._call(self._healthy_rows())
        self.assertEqual(
            result["expected_repository_binding"],
            {"remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools"},
        )

    def test_tool_reads_only_running_rows(self):
        rows = self._healthy_rows()
        fake_db = mock.Mock()
        fake_db.list_mcp_server_runtimes.return_value = rows
        self._call(rows, db=fake_db)
        fake_db.list_mcp_server_runtimes.assert_called_once_with(statuses=("running",))

    def test_tool_never_writes_to_the_registry(self):
        rows = self._healthy_rows()
        fake_db = mock.Mock()
        fake_db.list_mcp_server_runtimes.return_value = rows
        self._call(rows, db=fake_db)
        fake_db.register_mcp_server_runtime.assert_not_called()
        fake_db.heartbeat_mcp_server_runtime.assert_not_called()
        fake_db.mark_mcp_server_runtime_stopped.assert_not_called()

    def test_unavailable_control_plane_fails_closed(self):
        result = self._call([], db_errors=["control-plane DB substrate unavailable"])
        self.assertFalse(result["inventory_complete"])
        self.assertFalse(result["mutation_gate_satisfied"])
        self.assertFalse(result["registry"]["available"])
        self.assertIn("unavailable", result["blocked_reason"])

    def test_registry_read_failure_fails_closed(self):
        rows = self._healthy_rows()
        fake_db = mock.Mock()
        fake_db.list_mcp_server_runtimes.side_effect = RuntimeError("db locked")
        result = self._call(rows, db=fake_db)
        self.assertFalse(result["inventory_complete"])
        self.assertFalse(result["mutation_gate_satisfied"])
        self.assertIn("could not be read", result["registry"]["error"])

    def test_missing_read_permission_blocks_without_touching_the_registry(self):
        fake_db = mock.Mock()
        with mock.patch.object(
            gitea_mcp_server,
            "_profile_operation_gate",
            return_value=["profile may not read"],
        ), mock.patch.object(
            gitea_mcp_server, "_control_plane_db_or_error", return_value=(fake_db, [])
        ):
            result = gitea_mcp_server.gitea_assess_fleet_inventory(remote="prgs")
        self.assertFalse(result["success"])
        self.assertFalse(result["mutation_gate_satisfied"])
        self.assertIn("permission_report", result)
        self.assertEqual(result["mutations_performed"], [])
        fake_db.list_mcp_server_runtimes.assert_not_called()

    def test_answering_namespace_is_reported(self):
        result = self._call(self._healthy_rows())
        self.assertEqual(result["answering_namespace"], "gitea-controller")

    def test_summary_is_present(self):
        result = self._call(self._healthy_rows())
        self.assertIn("5 of 5", result["summary"])


class TestStartupRegistration(unittest.TestCase):
    """The registry row is written by the process it describes."""

    def test_registration_helper_exists_on_the_entrypoint_module(self):
        self.assertTrue(hasattr(gitea_mcp_server, "_register_fleet_runtime"))

    def test_registration_writes_one_row_for_this_process(self):
        fake_db = mock.Mock()
        with mock.patch.object(
            gitea_mcp_server, "_control_plane_db_or_error", return_value=(fake_db, [])
        ), mock.patch.object(
            gitea_mcp_server, "get_profile", return_value={"allowed_operations": []}
        ):
            record = gitea_mcp_server._register_fleet_runtime(transport="stdio")
        self.assertIsNotNone(record)
        fake_db.register_mcp_server_runtime.assert_called_once()
        written = fake_db.register_mcp_server_runtime.call_args.args[0]
        self.assertEqual(written["pid"], os.getpid())
        self.assertEqual(written["transport"], "stdio")

    def test_registration_failure_never_blocks_startup(self):
        with mock.patch.object(
            gitea_mcp_server,
            "_control_plane_db_or_error",
            side_effect=RuntimeError("boom"),
        ):
            self.assertIsNone(gitea_mcp_server._register_fleet_runtime())

    def test_registration_is_skipped_when_the_control_plane_is_unavailable(self):
        with mock.patch.object(
            gitea_mcp_server,
            "_control_plane_db_or_error",
            return_value=(None, ["unavailable"]),
        ):
            self.assertIsNone(gitea_mcp_server._register_fleet_runtime())

    def test_entrypoint_registers_after_binding_native_transport(self):
        """Order matters: only a transport-bound process may claim a row."""
        source = (REPO_ROOT / "gitea_mcp_server.py").read_text()
        bind_at = source.index('bind_native_mcp_transport(transport="stdio")')
        register_at = source.index('_register_fleet_runtime(transport="stdio")')
        run_at = source.index('mcp.run(transport="stdio")')
        self.assertLess(bind_at, register_at)
        self.assertLess(register_at, run_at)


class TestWorkflowDocumentation(unittest.TestCase):
    """AC13: documentation explains how the gates consume the result."""

    def setUp(self):
        self.doc = (REPO_ROOT / "docs" / "mcp-fleet-inventory.md").read_text()
        self.skill = (
            REPO_ROOT / "skills" / "llm-project-workflow" / "SKILL.md"
        ).read_text()

    def test_dedicated_document_exists(self):
        self.assertIn("# Authoritative MCP fleet inventory", self.doc)

    def test_document_names_both_consuming_namespaces(self):
        self.assertIn("gitea-controller", self.doc)
        self.assertIn("gitea-reconciler", self.doc)

    def test_document_explains_gate_consumption(self):
        self.assertIn("mutation_gate_satisfied", self.doc)
        self.assertIn("blocked_reason", self.doc)

    def test_document_states_the_non_inferences(self):
        self.assertIn("Configuration is not existence", self.doc)
        self.assertIn("Matching revisions are not a cohort", self.doc)

    def test_document_preserves_the_neighbouring_issue_boundaries(self):
        for issue in ("#950", "#951", "#952"):
            self.assertIn(issue, self.doc)

    def test_canonical_workflow_skill_links_the_document(self):
        self.assertIn("docs/mcp-fleet-inventory.md", self.skill)
        self.assertIn(TOOL_NAME, self.skill)


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