crafting-table/mcp/tests/test_tools.py

"""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://crafting-table.test: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://git.example.com/org/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 TestDraftPatch(unittest.TestCase):
    """Wave 3: real call to POST /jobs/{id}/patches; two-block return."""

    @responses.activate
    def test_pr_opened_two_block_return(self) -> None:
        """Server returns a pr_opened attempt → MCP returns prose + JSON."""
        responses.add(
            responses.POST,
            f"{BASE_URL}/jobs/j-1/patches",
            json={
                "ok": True,
                "attempt": {
                    "id": 7,
                    "finding_id": 42,
                    "job_id": "j-1",
                    "project_name": "demo",
                    "attempt_number": 1,
                    "status": "pr_opened",
                    "branch_name": "crafting-table/auto/j-1-42",
                    "pr_url": "http://git.example.com/X/Y/pulls/9",
                    "diff_excerpt": "--- a/x\n+++ b/x",
                    "session_id": "s-1",
                    "error": None,
                },
            },
            status=200,
        )
        c = _client()
        try:
            content, is_error = _run(
                _dispatch(c, TOOL_DRAFT_PATCH, {"job_id": "j-1"})
            )
        finally:
            c.close()
        self.assertFalse(is_error)
        # Two-content-block return: prose + JSON.
        self.assertEqual(len(content), 2)
        prose = content[0].text
        self.assertIn("pr_opened", prose)
        self.assertIn("crafting-table/auto/j-1-42", prose)
        self.assertIn("/pulls/9", prose)
        body = json.loads(content[1].text)
        self.assertTrue(body["ok"])
        self.assertEqual(body["attempt"]["status"], "pr_opened")

    @responses.activate
    def test_no_actionable_finding(self) -> None:
        responses.add(
            responses.POST,
            f"{BASE_URL}/jobs/j-1/patches",
            json={"ok": True, "attempt": None, "reason": "no_actionable_finding"},
            status=200,
        )
        c = _client()
        try:
            content, is_error = _run(
                _dispatch(c, TOOL_DRAFT_PATCH, {"job_id": "j-1"})
            )
        finally:
            c.close()
        self.assertFalse(is_error)
        self.assertIn("no actionable finding", content[0].text)

    @responses.activate
    def test_with_finding_id_passes_through(self) -> None:
        responses.add(
            responses.POST,
            f"{BASE_URL}/jobs/j-1/patches",
            json={
                "ok": True,
                "attempt": {
                    "id": 1, "finding_id": 42, "job_id": "j-1",
                    "project_name": "demo", "attempt_number": 1,
                    "status": "drafted", "branch_name": None, "pr_url": None,
                    "diff_excerpt": None, "session_id": None,
                    "error": "malformed_response",
                },
            },
            status=200,
        )
        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[1].text)
        self.assertEqual(body["attempt"]["finding_id"], 42)
        self.assertEqual(body["attempt"]["status"], "drafted")

    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)

    @responses.activate
    def test_503_when_patcher_disabled(self) -> None:
        responses.add(
            responses.POST,
            f"{BASE_URL}/jobs/j-1/patches",
            json={"detail": "patcher not configured"},
            status=503,
        )
        c = _client()
        try:
            content, is_error = _run(
                _dispatch(c, TOOL_DRAFT_PATCH, {"job_id": "j-1"})
            )
        finally:
            c.close()
        self.assertTrue(is_error)
        self.assertIn("503", 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()