Kayos
ecb9d76e6d
- mcp/ subpackage: crafting-table-mcp (separate pip install) - Self-contained requests-based HTTP client (mirrors clawdforge_mcp pattern) - 8 tools: list_projects / register_project / run_audit / run_build / run_test / get_job / get_findings / draft_patch (stub) - draft_patch is stubbed — full impl lands in wave 3 / step 9 - tests/: client + tool coverage, 401/404 surfacing - Tools designed for LLM consumption; descriptions tuned for "when to use" guidance Spec: memory/spec-crafting-table.md
665 lines
20 KiB
Python
665 lines
20 KiB
Python
"""Tool-dispatch tests for crafting_table_mcp.server._dispatch.
|
|
|
|
We exercise the full MCP-side path with the HTTP layer mocked via
|
|
``responses``. Each tool gets a happy-path test plus a 404/401 surfacing test.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import unittest
|
|
|
|
import responses
|
|
|
|
from crafting_table_mcp.client import CraftingTableClient
|
|
from crafting_table_mcp.server import (
|
|
TOOL_DRAFT_PATCH,
|
|
TOOL_GET_FINDINGS,
|
|
TOOL_GET_JOB,
|
|
TOOL_LIST_PROJECTS,
|
|
TOOL_REGISTER_PROJECT,
|
|
TOOL_RUN_AUDIT,
|
|
TOOL_RUN_BUILD,
|
|
TOOL_RUN_TEST,
|
|
_dispatch,
|
|
_tool_definitions,
|
|
build_server,
|
|
)
|
|
|
|
|
|
BASE_URL = "http://192.168.0.5:8810"
|
|
TOKEN = "ct_test_token_xxxxxxxx"
|
|
|
|
|
|
def _client() -> CraftingTableClient:
|
|
return CraftingTableClient(base_url=BASE_URL, token=TOKEN, timeout_secs=10)
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.run(coro)
|
|
|
|
|
|
class TestToolDiscovery(unittest.TestCase):
|
|
"""The MCP client calls list_tools first to discover capabilities."""
|
|
|
|
def test_eight_tools_with_valid_schemas(self) -> None:
|
|
tools = _tool_definitions()
|
|
names = [t.name for t in tools]
|
|
self.assertEqual(
|
|
sorted(names),
|
|
sorted(
|
|
[
|
|
TOOL_LIST_PROJECTS,
|
|
TOOL_REGISTER_PROJECT,
|
|
TOOL_RUN_AUDIT,
|
|
TOOL_RUN_BUILD,
|
|
TOOL_RUN_TEST,
|
|
TOOL_GET_JOB,
|
|
TOOL_GET_FINDINGS,
|
|
TOOL_DRAFT_PATCH,
|
|
]
|
|
),
|
|
)
|
|
for t in tools:
|
|
# Every tool must have a non-empty description (the LLM uses
|
|
# this to decide when to call it).
|
|
self.assertTrue(t.description and len(t.description) > 30, t.name)
|
|
self.assertEqual(t.inputSchema.get("type"), "object", t.name)
|
|
# Top-level should explicitly forbid extra args so the LLM
|
|
# doesn't get encouraged to invent keys.
|
|
self.assertFalse(
|
|
t.inputSchema.get("additionalProperties", True),
|
|
f"{t.name} should set additionalProperties=False",
|
|
)
|
|
|
|
def test_run_recipe_tools_share_schema(self) -> None:
|
|
"""run_audit / run_build / run_test must all require project_name."""
|
|
tools = {t.name: t for t in _tool_definitions()}
|
|
for name in (TOOL_RUN_AUDIT, TOOL_RUN_BUILD, TOOL_RUN_TEST):
|
|
schema = tools[name].inputSchema
|
|
self.assertEqual(schema["required"], ["project_name"])
|
|
self.assertEqual(
|
|
sorted(schema["properties"].keys()),
|
|
sorted(["project_name", "subproject", "branch"]),
|
|
)
|
|
|
|
def test_get_job_and_findings_require_job_id(self) -> None:
|
|
tools = {t.name: t for t in _tool_definitions()}
|
|
self.assertEqual(
|
|
tools[TOOL_GET_JOB].inputSchema["required"], ["job_id"]
|
|
)
|
|
self.assertEqual(
|
|
tools[TOOL_GET_FINDINGS].inputSchema["required"], ["job_id"]
|
|
)
|
|
|
|
|
|
class TestListProjects(unittest.TestCase):
|
|
@responses.activate
|
|
def test_happy_path(self) -> None:
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE_URL}/projects",
|
|
json={
|
|
"ok": True,
|
|
"projects": [
|
|
{"name": "alpha", "git_url": "x"},
|
|
{"name": "bravo", "git_url": "y"},
|
|
],
|
|
},
|
|
status=200,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(_dispatch(c, TOOL_LIST_PROJECTS, {}))
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(len(content), 1)
|
|
body = json.loads(content[0].text)
|
|
self.assertEqual(len(body), 2)
|
|
self.assertEqual(body[0]["name"], "alpha")
|
|
|
|
@responses.activate
|
|
def test_401_surfaces_as_mcp_error(self) -> None:
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE_URL}/projects",
|
|
json={"detail": "missing bearer"},
|
|
status=401,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(_dispatch(c, TOOL_LIST_PROJECTS, {}))
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("auth failed", content[0].text)
|
|
self.assertIn("CRAFTING_TABLE_TOKEN", content[0].text)
|
|
|
|
|
|
class TestRegisterProject(unittest.TestCase):
|
|
@responses.activate
|
|
def test_happy_path(self) -> None:
|
|
captured: dict = {}
|
|
|
|
def cb(request):
|
|
captured["body"] = json.loads(request.body)
|
|
return (
|
|
200,
|
|
{},
|
|
json.dumps(
|
|
{
|
|
"ok": True,
|
|
"project": {"name": "alpha", "git_url": "x"},
|
|
}
|
|
),
|
|
)
|
|
|
|
responses.add_callback(responses.POST, f"{BASE_URL}/projects", callback=cb)
|
|
c = _client()
|
|
project = {
|
|
"name": "alpha",
|
|
"git_url": "http://192.168.0.5:3001/Sulkta-Coop/alpha.git",
|
|
"default_branch": "main",
|
|
"languages": ["python"],
|
|
"subprojects": [
|
|
{
|
|
"path": ".",
|
|
"language": "python",
|
|
"test": "pytest",
|
|
}
|
|
],
|
|
}
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_REGISTER_PROJECT,
|
|
{"project_json": json.dumps(project)},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(captured["body"]["name"], "alpha")
|
|
body = json.loads(content[0].text)
|
|
self.assertEqual(body["name"], "alpha")
|
|
|
|
def test_rejects_missing_arg(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(c, TOOL_REGISTER_PROJECT, {})
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("project_json", content[0].text)
|
|
|
|
def test_rejects_invalid_json(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_REGISTER_PROJECT,
|
|
{"project_json": "not valid json {"},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("not valid JSON", content[0].text)
|
|
|
|
def test_rejects_non_object_json(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_REGISTER_PROJECT,
|
|
{"project_json": "[1, 2, 3]"},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("must decode to a JSON object", content[0].text)
|
|
|
|
def test_rejects_empty_string(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_REGISTER_PROJECT,
|
|
{"project_json": " "},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("project_json", content[0].text)
|
|
|
|
@responses.activate
|
|
def test_409_surfaces_as_mcp_error(self) -> None:
|
|
responses.add(
|
|
responses.POST,
|
|
f"{BASE_URL}/projects",
|
|
json={"detail": "project already exists; use PUT to update"},
|
|
status=409,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_REGISTER_PROJECT,
|
|
{
|
|
"project_json": json.dumps(
|
|
{"name": "dup", "git_url": "x"}
|
|
)
|
|
},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("already exists", content[0].text)
|
|
|
|
|
|
class TestRunRecipe(unittest.TestCase):
|
|
"""Shared coverage for run_audit / run_build / run_test."""
|
|
|
|
def _add_jobs_post(self, status_code: int, body: dict) -> dict:
|
|
captured: dict = {}
|
|
|
|
def cb(request):
|
|
captured["body"] = json.loads(request.body)
|
|
captured["url"] = request.url
|
|
return (status_code, {}, json.dumps(body))
|
|
|
|
responses.add_callback(
|
|
responses.POST,
|
|
f"{BASE_URL}/projects/alpha/jobs",
|
|
callback=cb,
|
|
)
|
|
return captured
|
|
|
|
@responses.activate
|
|
def test_run_audit_happy_path(self) -> None:
|
|
captured = self._add_jobs_post(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"job_id": "j-1",
|
|
"status": "queued",
|
|
"job": {"id": "j-1"},
|
|
},
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_RUN_AUDIT,
|
|
{"project_name": "alpha"},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(captured["body"]["recipe"], "audit")
|
|
body = json.loads(content[0].text)
|
|
self.assertEqual(body, {"job_id": "j-1", "status": "queued"})
|
|
|
|
@responses.activate
|
|
def test_run_build_passes_subproject_and_branch(self) -> None:
|
|
captured = self._add_jobs_post(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"job_id": "j-2",
|
|
"status": "queued",
|
|
"job": {"id": "j-2"},
|
|
},
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_RUN_BUILD,
|
|
{
|
|
"project_name": "alpha",
|
|
"subproject": "clients/rust",
|
|
"branch": "feature/x",
|
|
},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(captured["body"]["recipe"], "build")
|
|
self.assertEqual(captured["body"]["subproject"], "clients/rust")
|
|
self.assertEqual(captured["body"]["branch"], "feature/x")
|
|
|
|
@responses.activate
|
|
def test_run_test_recipe_label(self) -> None:
|
|
captured = self._add_jobs_post(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"job_id": "j-3",
|
|
"status": "queued",
|
|
"job": {"id": "j-3"},
|
|
},
|
|
)
|
|
c = _client()
|
|
try:
|
|
_run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_RUN_TEST,
|
|
{"project_name": "alpha"},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertEqual(captured["body"]["recipe"], "test")
|
|
|
|
def test_rejects_missing_project_name(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(c, TOOL_RUN_AUDIT, {})
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("project_name", content[0].text)
|
|
|
|
def test_rejects_non_string_subproject(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_RUN_AUDIT,
|
|
{"project_name": "alpha", "subproject": 42},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("subproject", content[0].text)
|
|
|
|
@responses.activate
|
|
def test_404_surfaces_with_actionable_hint(self) -> None:
|
|
responses.add(
|
|
responses.POST,
|
|
f"{BASE_URL}/projects/missing/jobs",
|
|
json={"detail": "project not found"},
|
|
status=404,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_RUN_AUDIT,
|
|
{"project_name": "missing"},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
# Auth/404 wrapper should give the LLM a hint about list_projects.
|
|
self.assertIn("not found", content[0].text)
|
|
self.assertIn("crafting_table_list_projects", content[0].text)
|
|
|
|
|
|
class TestGetJob(unittest.TestCase):
|
|
@responses.activate
|
|
def test_happy_path_two_blocks(self) -> None:
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE_URL}/jobs/j-1",
|
|
json={
|
|
"ok": True,
|
|
"job": {
|
|
"id": "j-1",
|
|
"project_name": "alpha",
|
|
"subproject_path": ".",
|
|
"recipe": "audit",
|
|
"status": "succeeded",
|
|
"exit_code": 0,
|
|
},
|
|
"log_tail": ["pip install ...", "everything is fine"],
|
|
},
|
|
status=200,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(c, TOOL_GET_JOB, {"job_id": "j-1"})
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
# Two-block response: prose + JSON.
|
|
self.assertEqual(len(content), 2)
|
|
self.assertIn("succeeded", content[0].text)
|
|
self.assertIn("alpha::.", content[0].text)
|
|
self.assertIn("everything is fine", content[0].text)
|
|
body = json.loads(content[1].text)
|
|
self.assertEqual(body["job"]["status"], "succeeded")
|
|
self.assertEqual(body["log_tail"], ["pip install ...", "everything is fine"])
|
|
|
|
@responses.activate
|
|
def test_404_surfaces_as_mcp_error(self) -> None:
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE_URL}/jobs/missing",
|
|
json={"detail": "job not found"},
|
|
status=404,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(c, TOOL_GET_JOB, {"job_id": "missing"})
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("not found", content[0].text)
|
|
self.assertIn("crafting_table_list_projects", content[0].text)
|
|
|
|
def test_rejects_missing_job_id(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(_dispatch(c, TOOL_GET_JOB, {}))
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("job_id", content[0].text)
|
|
|
|
|
|
class TestGetFindings(unittest.TestCase):
|
|
@responses.activate
|
|
def test_happy_path_two_blocks(self) -> None:
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE_URL}/jobs/j-1/findings",
|
|
json={
|
|
"ok": True,
|
|
"findings": [
|
|
{
|
|
"id": 1,
|
|
"kind": "lint",
|
|
"severity": "warn",
|
|
"file": "src/x.py",
|
|
"line": 10,
|
|
"code": "ruff::E501",
|
|
"message": "line too long",
|
|
},
|
|
{
|
|
"id": 2,
|
|
"kind": "cve",
|
|
"severity": "high",
|
|
"message": "openssl bump",
|
|
},
|
|
],
|
|
},
|
|
status=200,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(c, TOOL_GET_FINDINGS, {"job_id": "j-1"})
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(len(content), 2)
|
|
# Block 0: prose summary should mention both severities + kinds.
|
|
prose = content[0].text
|
|
self.assertIn("2 finding", prose)
|
|
self.assertIn("warn", prose)
|
|
self.assertIn("high", prose)
|
|
self.assertIn("lint", prose)
|
|
self.assertIn("cve", prose)
|
|
# Block 1: full JSON with the findings list.
|
|
body = json.loads(content[1].text)
|
|
self.assertEqual(len(body["findings"]), 2)
|
|
|
|
@responses.activate
|
|
def test_empty_findings_prose_explains(self) -> None:
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE_URL}/jobs/j-2/findings",
|
|
json={"ok": True, "findings": []},
|
|
status=200,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(c, TOOL_GET_FINDINGS, {"job_id": "j-2"})
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
# Prose block should still help the LLM understand "empty doesn't
|
|
# necessarily mean clean — could be no parser yet".
|
|
self.assertIn("no findings", content[0].text.lower())
|
|
|
|
@responses.activate
|
|
def test_404_surfaces_as_mcp_error(self) -> None:
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE_URL}/jobs/missing/findings",
|
|
json={"detail": "job not found"},
|
|
status=404,
|
|
)
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(c, TOOL_GET_FINDINGS, {"job_id": "missing"})
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("not found", content[0].text)
|
|
|
|
|
|
class TestDraftPatchStub(unittest.TestCase):
|
|
"""Wave 2B stub: tool surface present, but returns a 'pending' message."""
|
|
|
|
def test_returns_pending_message(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_DRAFT_PATCH,
|
|
{"job_id": "j-1"},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
body = json.loads(content[0].text)
|
|
self.assertFalse(body["ok"])
|
|
self.assertTrue(body["pending"])
|
|
self.assertIn("not yet implemented", body["message"])
|
|
self.assertIn("wave 3", body["message"])
|
|
|
|
def test_with_finding_id(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_DRAFT_PATCH,
|
|
{"job_id": "j-1", "finding_id": 42},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertFalse(is_error)
|
|
body = json.loads(content[0].text)
|
|
self.assertEqual(body["finding_id"], 42)
|
|
|
|
def test_rejects_bool_finding_id(self) -> None:
|
|
# bool is a subclass of int — defense-in-depth.
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(
|
|
_dispatch(
|
|
c,
|
|
TOOL_DRAFT_PATCH,
|
|
{"job_id": "j-1", "finding_id": True},
|
|
)
|
|
)
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("finding_id", content[0].text)
|
|
|
|
def test_rejects_missing_job_id(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(_dispatch(c, TOOL_DRAFT_PATCH, {}))
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("job_id", content[0].text)
|
|
|
|
|
|
class TestUnknownTool(unittest.TestCase):
|
|
def test_unknown_tool_returns_error(self) -> None:
|
|
c = _client()
|
|
try:
|
|
content, is_error = _run(_dispatch(c, "not_a_tool", {}))
|
|
finally:
|
|
c.close()
|
|
self.assertTrue(is_error)
|
|
self.assertIn("unknown tool", content[0].text)
|
|
|
|
|
|
class TestServerFactory(unittest.TestCase):
|
|
"""Smoke test that the SDK accepts our wiring."""
|
|
|
|
def test_build_server_returns_named_server(self) -> None:
|
|
c = _client()
|
|
try:
|
|
server = build_server(c)
|
|
self.assertEqual(server.name, "crafting-table-mcp")
|
|
init = server.create_initialization_options()
|
|
self.assertEqual(init.server_name, "crafting-table-mcp")
|
|
finally:
|
|
c.close()
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
unittest.main()
|