leocrm/tests/test_mcp_server.py

"""Tests for MCP Server plugin — tool listing, execution, config.

Tests Task 5.16 acceptance criteria.
"""

from __future__ import annotations

import pytest

from tests.conftest import ORIGIN_HEADER


# ─── AC1: List MCP tools ───


@pytest.mark.asyncio
async def test_ac1_list_mcp_tools(mcp_authed_client):
    """AC1: GET /api/v1/mcp/tools → 200 + tool list with 9 tools."""
    client, _ = mcp_authed_client
    resp = await client.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
    assert resp.status_code == 200
    data = resp.json()
    assert data["count"] >= 1
    assert len(data["tools"]) == data["count"]
    tool_names = [t["name"] for t in data["tools"]]
    assert "call_crm_api" in tool_names


# ─── AC2: Get MCP config ───


@pytest.mark.asyncio
async def test_ac2_get_mcp_config(mcp_authed_client):
    """AC2: GET /api/v1/mcp/config → 200 + server config."""
    client, _ = mcp_authed_client
    resp = await client.get("/api/v1/mcp/config", headers=ORIGIN_HEADER)
    assert resp.status_code == 200
    data = resp.json()
    assert data["server_name"] == "LeoCRM"
    assert data["server_version"] == "1.0.0"
    assert data["protocol_version"] == "2024-11-05"
    assert data["auth_method"] == "api-token"
    assert "call_crm_api" in data["available_tools"]
    assert len(data["available_tools"]) >= 1


# ─── AC3: Execute search_contacts tool ───


@pytest.mark.asyncio
async def test_ac3_execute_search_contacts(mcp_authed_client):
    """AC3: POST /api/v1/mcp/tools/call_crm_api/execute → 200 + results."""
    client, _ = mcp_authed_client
    resp = await client.post(
        "/api/v1/mcp/tools/call_crm_api/execute",
        json={"arguments": {"method": "GET", "path": "/api/v1/contacts"}},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["tool"] == "call_crm_api"
    assert data["success"] is True
    assert "result" in data


# ─── AC4: Execute non-existent tool returns 404 ───


@pytest.mark.asyncio
async def test_ac4_execute_nonexistent_tool(mcp_authed_client):
    """AC4: POST /api/v1/mcp/tools/nonexistent/execute → 404."""
    client, _ = mcp_authed_client
    resp = await client.post(
        "/api/v1/mcp/tools/nonexistent_tool/execute",
        json={"arguments": {}},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 404
    assert "tool_not_found" in resp.text


# ─── AC5: Tool definitions have correct schema ───


@pytest.mark.asyncio
async def test_ac5_tool_definitions_schema(mcp_authed_client):
    """AC5: GET /api/v1/mcp/tools → tools have proper schema with parameters."""
    client, _ = mcp_authed_client
    resp = await client.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
    assert resp.status_code == 200
    tools = resp.json()["tools"]

    # Check call_crm_api has method, path, body params
    api_tool = next(t for t in tools if t["name"] == "call_crm_api")
    param_names = [p["name"] for p in api_tool["parameters"]]
    assert "method" in param_names
    assert "path" in param_names
    method_param = next(p for p in api_tool["parameters"] if p["name"] == "method")
    assert method_param["required"] is True


# ─── AC6: Unauthorized access is rejected ───


@pytest.mark.asyncio
async def test_ac6_unauthorized_access(mcp_client_fixture):
    """AC6: GET /api/v1/mcp/tools without auth → 401."""
    resp = await mcp_client_fixture.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
    assert resp.status_code == 401


# ─── AC7: Execute create_contact tool ───


@pytest.mark.asyncio
async def test_ac7_execute_create_contact(mcp_authed_client):
    """AC7: POST /api/v1/mcp/tools/call_crm_api/execute → 200 + created contact."""
    client, _ = mcp_authed_client
    resp = await client.post(
        "/api/v1/mcp/tools/call_crm_api/execute",
        json={"arguments": {"method": "POST", "path": "/api/v1/contacts", "body": {"firstname": "MCP", "surname": "Test", "email_1": "mcp@test.com"}}},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["tool"] == "call_crm_api"
    assert data["success"] is True
    assert "result" in data