diff --git a/bambu_run/mcp/toolkit.py b/bambu_run/mcp/toolkit.py new file mode 100644 index 0000000..956347e --- /dev/null +++ b/bambu_run/mcp/toolkit.py @@ -0,0 +1,28 @@ +"""Assembles the 8-tool bambu_run MCP toolkit, capability-tagged per the spec.""" +from __future__ import annotations + +from mcp_gate import Toolkit + +from bambu_run.mcp.filaments import get_filament, list_filaments +from bambu_run.mcp.jobs import get_print_job, get_print_statistics, list_print_jobs +from bambu_run.mcp.status import get_hardware_health, get_metrics_history, get_printer_status + + +def build_toolkit() -> Toolkit: + """Build the bambu_run MCP toolkit: 8 tools across read_printer, + read_print_history, and read_filament. + + Called fresh by each entry point rather than shared as a module-level + singleton, so tests (and multiple concurrent Gates, if ever needed) get + an isolated Toolkit instance rather than sharing registration state. + """ + kit = Toolkit("bambu_run") + kit.tool(requires="read_printer")(get_printer_status) + kit.tool(requires="read_printer")(get_metrics_history) + kit.tool(requires="read_printer")(get_hardware_health) + kit.tool(requires="read_print_history")(list_print_jobs) + kit.tool(requires="read_print_history")(get_print_job) + kit.tool(requires="read_print_history")(get_print_statistics) + kit.tool(requires="read_filament")(list_filaments) + kit.tool(requires="read_filament")(get_filament) + return kit diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..1e5215a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,30 @@ +"""Suite-wide test fixtures. + +test_mcp_toolkit.py drives real MCP dispatch through `mcp.client.Client`, +which routes each tool call's synchronous Django ORM work through +`sync_to_async` onto its own dedicated worker thread rather than the test's +own thread. pytest-django's `db` fixture isolates each test by wrapping it in +an outer atomic transaction on *its* thread's connection and rolling that +back afterwards -- but the worker thread opens a separate, thread-local +connection that was never part of that transaction, so writes made through +it commit for real and survive into later tests (e.g. Printer rows from +test_mcp_toolkit.py leaking into test_resolve_printer_device.py's counts). + +This fixture flushes bambu_run's Printer table after each test in that file +specifically, restoring isolation without touching the test file itself +(which mirrors mcp-gate's proven end-to-end test pattern verbatim). +""" +from __future__ import annotations + +import pytest + + +@pytest.fixture(autouse=True) +def _isolate_async_mcp_dispatch_writes(request, django_db_blocker): + yield + if request.node.module.__name__ != "tests.test_mcp_toolkit": + return + with django_db_blocker.unblock(): + from bambu_run.models import Printer + + Printer.objects.all().delete() diff --git a/tests/settings.py b/tests/settings.py index 3855817..a2a3cbf 100644 --- a/tests/settings.py +++ b/tests/settings.py @@ -1,4 +1,16 @@ """Minimal Django settings for running bambu_run's pytest suite (in-memory SQLite).""" +import os + +# test_mcp_toolkit.py drives real MCP dispatch through `mcp.client.Client`, +# which runs the test body on the asyncio event loop (via pytest-anyio) while +# still using plain synchronous Django ORM calls (Printer.objects.create(), +# and the ORM calls inside the tool functions themselves). Django's asyncio +# safety net treats any sync ORM call made while an event loop is running as +# unsafe by default. It's genuinely safe here: sqlite is in-memory, tests run +# single-threaded, and there's no concurrent request interleaving to corrupt +# a shared connection. Set at settings-import time (before any test body +# runs) rather than per-test, since Django reads this env var at call time. +os.environ.setdefault("DJANGO_ALLOW_ASYNC_UNSAFE", "1") SECRET_KEY = "test-secret-key" diff --git a/tests/test_mcp_toolkit.py b/tests/test_mcp_toolkit.py new file mode 100644 index 0000000..5a121e5 --- /dev/null +++ b/tests/test_mcp_toolkit.py @@ -0,0 +1,83 @@ +"""Proves this toolkit's capability assignments are wired correctly through +real MCP dispatch -- not by calling tool functions directly. See +~/src/mcp-gate/tests/test_end_to_end.py for the pattern this is modeled on; +mcp-gate's own HTTP-shaped dispatch machinery is already tested there, so +this file only needs the stdio-shaped Client, which is enough to drive +server.middleware and prove which tools this toolkit's capability tags +actually gate. +""" +import pytest +from mcp.client import Client +from mcp.shared.exceptions import MCPError + +from mcp_gate import Gate, OpenIdentity + +from bambu_run.mcp.toolkit import build_toolkit +from bambu_run.models import Printer + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +async def test_full_capability_caller_sees_all_eight_tools(db): + Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + gate = Gate( + name="test", + toolkits=[build_toolkit()], + identity=OpenIdentity(["read_printer", "read_print_history", "read_filament"]), + ) + server = gate.build_server() + + async with Client(server, mode="legacy") as client: + result = await client.list_tools() + + names = {t.name for t in result.tools} + assert names == { + "get_printer_status", "get_metrics_history", "get_hardware_health", + "list_print_jobs", "get_print_job", "get_print_statistics", + "list_filaments", "get_filament", + } + + +async def test_read_printer_only_caller_cannot_see_or_call_filament_tool(db): + Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + gate = Gate( + name="test", + toolkits=[build_toolkit()], + identity=OpenIdentity(["read_printer"]), + ) + server = gate.build_server() + + async with Client(server, mode="legacy") as client: + visible = await client.list_tools() + visible_names = {t.name for t in visible.tools} + assert "list_filaments" not in visible_names + assert "get_printer_status" in visible_names + + with pytest.raises(MCPError): + await client.call_tool("list_filaments", {}) + + +async def test_read_printer_caller_can_call_get_printer_status(db): + Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + gate = Gate( + name="test", + toolkits=[build_toolkit()], + identity=OpenIdentity(["read_printer"]), + ) + server = gate.build_server() + + async with Client(server, mode="legacy") as client: + result = await client.call_tool("get_printer_status", {}) + assert result is not None + + +def test_build_toolkit_has_no_name_collisions(): + # Toolkit.tool() raises ValueError on a name collision at registration + # time -- building it twice without raising is itself the assertion. + build_toolkit() + build_toolkit()