Merge pull request openSUSE#321 from plusky/test/pin-mcp-leftovers

test(mcp): pin scattered registry/slim/session/tools behavior

Author: mimi1vx

tests/test_mcp_registry_leftovers_pins.py

@@ -0,0 +1,116 @@
+"""Mutation-killing pin for the idle-sweeper's re-validation loop.
+
+``tests/test_mcp_registry.py::test_sweep_spares_session_reactivated_during_the_sweep``
+already pins that a session re-activated mid-sweep is spared. But in that
+test the re-activated key is the *last* entry in the sweep's stale batch,
+so the "re-activated -> skip this one" ``continue`` is indistinguishable
+from a ``break``: either way the loop has nothing left to do. A full
+mutmut run found exactly that survivor (``continue`` -> ``break`` in
+``SessionRegistry._sweep_loop``'s re-validation branch).
+
+This test mirrors that fixture but inserts a third, genuinely-idle key
+*after* the reactivated one in stale order, so ``continue`` (proceed to
+the next stale key) and ``break`` (abandon the rest of this sweep round)
+produce observably different outcomes: only ``continue`` still evicts
+the trailing idle key in the same round.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from pathlib import Path
+from unittest.mock import MagicMock
+
+from mtui.mcp.main import build_session
+from mtui.mcp.registry import SessionRegistry
+
+_LOG = logging.getLogger("test.mcp.registry.leftovers")
+
+
+def _config(tmp_path: Path) -> MagicMock:
+    """The minimal Config shape McpSession's constructor touches."""
+    cfg = MagicMock()
+    cfg.template_dir = tmp_path
+    cfg.target_tempdir = tmp_path / "target"
+    cfg.chdir_to_template_dir = False
+    cfg.connection_timeout = 30
+    cfg.session_user = "testuser"
+    return cfg
+
+
+def _registry(tmp_path: Path, *, idle_timeout: float) -> SessionRegistry:
+    return SessionRegistry(
+        build_session,
+        _config(tmp_path),
+        _LOG,
+        max_sessions=32,
+        idle_timeout=idle_timeout,
+    )
+
+
+def test_sweep_continues_past_a_reactivated_key_to_evict_a_later_idle_one(
+    tmp_path: Path,
+) -> None:
+    """A stale batch of [k1, k2, k3] where k2 is reactivated mid-sweep.
+
+    k1's close is slow (buys the window needed to reactivate k2 before the
+    sweep reaches it); k2 is refreshed during that window and must be
+    spared (hits the "re-activated" branch); k3 is never touched and must
+    still be evicted in the *same* sweep round. A ``continue`` -> ``break``
+    mutation on the re-activation branch would stop the round at k2 and
+    leave k3 (genuinely idle) un-evicted.
+    """
+    reg = _registry(tmp_path, idle_timeout=4.0)  # sweep interval = 2s
+
+    close_started = asyncio.Event()
+    release_close = asyncio.Event()
+    k2_closed = {"n": 0}
+    k3_closed = {"n": 0}
+
+    async def driver() -> None:
+        s1 = await reg.get_or_create("k1")
+        s2 = await reg.get_or_create("k2")
+        s3 = await reg.get_or_create("k3")
+
+        async def _slow_close() -> None:
+            close_started.set()
+            await release_close.wait()
+
+        async def _k2_close() -> None:
+            k2_closed["n"] += 1
+
+        async def _k3_close() -> None:
+            k3_closed["n"] += 1
+
+        s1.close = _slow_close  # ty: ignore[invalid-assignment]
+        s2.close = _k2_close  # ty: ignore[invalid-assignment]
+        s3.close = _k3_close  # ty: ignore[invalid-assignment]
+
+        # Age all three well past the TTL so the next sweep round snapshots
+        # stale = [k1, k2, k3] (insertion order).
+        aged = time.monotonic() - 100
+        reg._last_touch["k1"] = aged  # noqa: SLF001
+        reg._last_touch["k2"] = aged  # noqa: SLF001
+        reg._last_touch["k3"] = aged  # noqa: SLF001
+
+        # Wait until the sweeper is mid-eviction of k1 (blocked in close),
+        # then do what a live client does: grab k2, refreshing its touch.
+        await asyncio.wait_for(close_started.wait(), timeout=10)
+        again = await reg.get_or_create("k2")
+        assert again is s2
+
+        # Let the k1 eviction finish; the sweep round proceeds to k2 (spared)
+        # and must then continue on to k3 (genuinely idle).
+        release_close.set()
+        await asyncio.sleep(0.5)
+
+        assert "k1" not in reg._sessions  # noqa: SLF001 -- genuinely idle: reaped
+        assert "k2" in reg._sessions  # noqa: SLF001 -- re-activated: spared
+        assert k2_closed["n"] == 0
+        assert "k3" not in reg._sessions  # noqa: SLF001 -- idle: reaped despite k2's skip
+        assert k3_closed["n"] == 1
+        await reg.aclose()
+
+    asyncio.run(driver())

tests/test_mcp_session_leftovers_pins.py

@@ -0,0 +1,139 @@
+"""Mutation-killing pins for small leftovers in :mod:`mtui.mcp.session`.
+
+Two narrow gaps a full mutmut run left unasserted:
+
+* ``McpSession.start_jobs``: when an explicit ``-T <rrid>`` narrows the
+  fan-out to a single template, ``_resolve_job_rrids`` returns exactly one
+  rrid and ``len(rrids) <= 1`` takes the single-job path (the stable
+  ``<command>-<n>`` id shape, with no RRID token embedded). The existing
+  ``test_start_jobs_explicit_template_yields_single_job`` (in
+  ``tests/test_mcp_jobs.py``) asserts only the job count and output, not
+  the id shape, so a ``<=`` -> ``<`` mutant (which pushes this case onto
+  the fan-out path instead) survives. This module re-drives that same
+  scenario and additionally pins the id shape.
+* ``job_status``'s unknown-id refusal raises the same uniform
+  :class:`McpCommandError` envelope (``stdout=""``, ``exit_code=1``)
+  every other tool refusal uses, but the existing
+  ``test_job_status_unknown_id_raises`` only regex-matches the stderr
+  message, so mutants on the envelope's ``stdout``/``exit_code``
+  constants survive. ``job_result`` and ``job_cancel`` raise the
+  identical envelope for the same unknown-id case, so one parametrized
+  test pins all three.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from pathlib import Path
+from typing import ClassVar
+from unittest.mock import MagicMock
+
+import pytest
+
+from mtui.commands import Command
+from mtui.mcp.session import McpCommandError, McpSession
+
+
+def _config(tmp_path: Path) -> MagicMock:
+    cfg = MagicMock()
+    cfg.template_dir = tmp_path
+    cfg.target_tempdir = tmp_path / "target"
+    cfg.chdir_to_template_dir = False
+    cfg.connection_timeout = 30
+    cfg.session_user = "testuser"
+    return cfg
+
+
+def _make_session(tmp_path: Path) -> McpSession:
+    return McpSession(
+        _config(tmp_path), logging.getLogger("test.mcp.session.leftovers")
+    )
+
+
+def _load_two_reports(sess: McpSession) -> None:
+    """Add two MagicMock reports so a fan-out command resolves to both."""
+    for rrid in ("SUSE:Maintenance:1:1", "SUSE:Maintenance:2:1"):
+        report = MagicMock()
+        report.id = rrid
+        report.targets = {}
+        sess.templates.add(report)
+    sess.templates.set_active("SUSE:Maintenance:1:1")
+
+
+def _fanout_probe():
+    """Build a throwaway fast fan-out command; caller must unregister it."""
+
+    class _FanoutJobProbeLeftovers(Command):
+        command = "fanout_job_probe_leftovers_tmp"
+        scope: ClassVar[str] = "fanout"
+
+        @classmethod
+        def _add_arguments(cls, parser) -> None:
+            cls._add_template_arg(parser)
+
+        def __call__(self) -> None:
+            self.println(str(self.metadata.id))
+
+    return _FanoutJobProbeLeftovers
+
+
+# --------------------------------------------------------------------------- #
+# start_jobs: explicit -T narrowing keeps the single-job id shape             #
+# --------------------------------------------------------------------------- #
+
+
+def test_start_jobs_explicit_template_keeps_the_single_job_id_shape(
+    tmp_path: Path,
+) -> None:
+    """A client-supplied ``-T`` narrows to one template: legacy id, no RRID.
+
+    ``len(rrids) <= 1`` must route here even though ``rrids`` is
+    non-empty; a ``<=`` -> ``<`` mutant instead sends this down the
+    fan-out path, which embeds the (sanitised) RRID in the job id.
+    """
+    sess = _make_session(tmp_path)
+    _load_two_reports(sess)
+    cls = _fanout_probe()
+
+    async def driver() -> list[str]:
+        ids = await sess.start_jobs(cls, ["-T", "SUSE:Maintenance:2:1"])
+        for jid in ids:
+            await sess._jobs[jid]["task"]
+        return ids
+
+    try:
+        ids = asyncio.run(driver())
+    finally:
+        Command.registry.pop(cls.command, None)
+
+    assert len(ids) == 1
+    assert ids[0].startswith(f"{cls.command}-")
+    assert "SUSE" not in ids[0]
+    assert sess.job_result(ids[0]).strip() == "SUSE:Maintenance:2:1"
+
+
+# --------------------------------------------------------------------------- #
+# Unknown job id: the uniform McpCommandError envelope                       #
+# --------------------------------------------------------------------------- #
+
+
+@pytest.mark.parametrize(
+    "invoke",
+    [
+        lambda s: s.job_status("nope-1"),
+        lambda s: s.job_result("nope-1"),
+        lambda s: asyncio.run(s.job_cancel("nope-1")),
+    ],
+    ids=["job_status", "job_result", "job_cancel"],
+)
+def test_unknown_job_id_raises_the_uniform_envelope(invoke, tmp_path: Path) -> None:
+    """Every unknown-id lookup raises stdout='', exit_code=1 verbatim."""
+    sess = _make_session(tmp_path)
+
+    with pytest.raises(McpCommandError) as ei:
+        invoke(sess)
+
+    assert ei.value.stdout == ""
+    assert ei.value.stderr == "no such job: nope-1"
+    assert ei.value.exit_code == 1

tests/test_mcp_slim_leftovers_pins.py

@@ -0,0 +1,126 @@
+"""Mutation-killing pins for two leftover gaps in :mod:`mtui.mcp._slim`.
+
+* ``_slim``'s ``keys_are_names`` flag must propagate correctly through
+  ``items`` sub-schemas and ``anyOf`` list members so a nested ``title``
+  still gets dropped. ``tests/test_mcp_slim.py`` drives the real
+  pydantic-generated schemas for mtui commands, none of which happen to
+  carry a ``title`` inside an ``items`` dict or an ``anyOf`` list member,
+  so an ``and`` -> ``or`` mutant on the propagation expression (and the
+  twin "list items always recurse as non-names" mutant) survived. This
+  test hand-builds a schema shaped exactly like that to exercise both
+  paths directly.
+* ``slim_registered_tools``: the existing
+  ``test_slim_registered_tools_reduces_bytes_keeps_count`` only checks
+  the aggregate byte count, tool count, and absence of ``title``
+  keywords — all of which pass even if every tool's ``parameters`` were
+  replaced by ``None`` (``json.dumps(None)`` is 4 bytes, and the
+  title-walker no-ops on non-dict/non-list nodes). This test spot-checks
+  one specific tool still has a ``dict`` schema with its expected
+  parameter names after slimming.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+pytest.importorskip("mcp")
+
+from mcp.server.fastmcp import FastMCP  # noqa: E402
+
+from mtui.mcp._slim import _slim, slim_registered_tools  # noqa: E402
+from mtui.mcp.tools import build_tools, register_job_tools  # noqa: E402
+
+
+class _Provider:
+    async def get_or_create(self, key):  # noqa: ANN001, ANN201, D102
+        raise AssertionError("not used in schema tests")
+
+
+# --------------------------------------------------------------------------- #
+# _slim: keys_are_names propagates into items/ and anyOf members             #
+# --------------------------------------------------------------------------- #
+
+
+def test_slim_drops_title_nested_in_items_and_anyof_members() -> None:
+    """A hand-built schema with ``title`` inside ``items`` and ``anyOf``.
+
+    Also carries an ``enum`` sibling *after* ``description`` inside the
+    ``items`` sub-schema so a ``continue`` -> ``break`` mutant on the
+    description branch (which would abandon the rest of that dict's
+    keys) is caught too: ``enum`` must still survive.
+    """
+    schema = {
+        "properties": {
+            "x": {
+                "type": "array",
+                "items": {
+                    "title": "T",
+                    "description": "d",
+                    "enum": ["a", "b"],
+                },
+            }
+        },
+        "anyOf": [
+            {"title": "U", "type": "string"},
+            {"type": "integer"},
+        ],
+    }
+
+    result = _slim(schema, keys_are_names=False)
+
+    assert result == {
+        "properties": {
+            "x": {
+                "type": "array",
+                "items": {"description": "d", "enum": ["a", "b"]},
+            }
+        },
+        "anyOf": [
+            {"type": "string"},
+            {"type": "integer"},
+        ],
+    }
+
+    def _walk_no_title(node: object) -> None:
+        if isinstance(node, dict):
+            assert "title" not in node
+            for value in node.values():
+                _walk_no_title(value)
+        elif isinstance(node, list):
+            for item in node:
+                _walk_no_title(item)
+
+    _walk_no_title(result)
+
+
+# --------------------------------------------------------------------------- #
+# slim_registered_tools: schemas stay real dicts, not a wholesale wipe        #
+# --------------------------------------------------------------------------- #
+
+
+def test_slim_registered_tools_keeps_a_real_schema_on_a_known_tool() -> None:
+    """After slimming, ``job_status`` still has a dict schema with ``job_id``.
+
+    A mutant that replaces every tool's ``parameters`` with ``None``
+    (or slims ``None`` instead of the real schema, yielding the same
+    ``None`` result) satisfies the byte-count and no-title checks in
+    ``test_slim_registered_tools_reduces_bytes_keeps_count`` but would
+    break every MCP client's tool calls. This spot-checks one concrete,
+    stable tool's schema shape survives slimming intact.
+    """
+    mcp = FastMCP(name="mtui-test")
+    prov = _Provider()
+    build_tools(mcp, prov)
+    register_job_tools(mcp, prov)
+
+    tool = mcp._tool_manager.get_tool("job_status")  # noqa: SLF001
+    assert tool is not None
+
+    n = slim_registered_tools(mcp)
+    assert n > 0
+
+    params = tool.parameters
+    assert isinstance(params, dict)
+    assert "properties" in params
+    assert "job_id" in params["properties"]
+    assert params["properties"]["job_id"].get("type") == "string"