leocrm/tests/test_mcp_client.py

"""Tests for MCP Client plugin — server config CRUD, tool listing, execution.

Tests Task 5.17 acceptance criteria.
"""

from __future__ import annotations

import pytest

from tests.conftest import ORIGIN_HEADER


# ─── AC1: List MCP client servers (empty) ───


@pytest.mark.asyncio
async def test_ac1_list_servers_empty(mcp_authed_client):
    """AC1: GET /api/v1/mcp-client/servers → 200 + empty list."""
    client, _ = mcp_authed_client
    resp = await client.get("/api/v1/mcp-client/servers", headers=ORIGIN_HEADER)
    assert resp.status_code == 200
    assert resp.json() == []


# ─── AC2: Create MCP server config ───


@pytest.mark.asyncio
async def test_ac2_create_server(mcp_authed_client):
    """AC2: POST /api/v1/mcp-client/servers → 201 + created config."""
    client, _ = mcp_authed_client
    resp = await client.post(
        "/api/v1/mcp-client/servers",
        json={"name": "Web Search MCP", "url": "http://localhost:9000", "api_token": "test-token", "enabled": True, "description": "External web search"},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 201
    data = resp.json()
    assert data["name"] == "Web Search MCP"
    assert data["url"] == "http://localhost:9000"
    assert data["api_token"] == "test-token"
    assert data["enabled"] is True
    assert data["description"] == "External web search"
    assert "id" in data


# ─── AC3: Update MCP server config ───


@pytest.mark.asyncio
async def test_ac3_update_server(mcp_authed_client):
    """AC3: PATCH /api/v1/mcp-client/servers/{id} → 200 + updated."""
    client, _ = mcp_authed_client
    # Create first
    resp = await client.post(
        "/api/v1/mcp-client/servers",
        json={"name": "Test Server", "url": "http://localhost:8000", "enabled": True},
        headers=ORIGIN_HEADER,
    )
    server_id = resp.json()["id"]

    # Update
    resp = await client.patch(
        f"/api/v1/mcp-client/servers/{server_id}",
        json={"name": "Updated Server", "enabled": False},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["name"] == "Updated Server"
    assert data["enabled"] is False
    assert data["url"] == "http://localhost:8000"  # unchanged


# ─── AC4: Delete MCP server config ───


@pytest.mark.asyncio
async def test_ac4_delete_server(mcp_authed_client):
    """AC4: DELETE /api/v1/mcp-client/servers/{id} → 204."""
    client, _ = mcp_authed_client
    # Create first
    resp = await client.post(
        "/api/v1/mcp-client/servers",
        json={"name": "Delete Me", "url": "http://localhost:7000", "enabled": True},
        headers=ORIGIN_HEADER,
    )
    server_id = resp.json()["id"]

    # Delete
    resp = await client.delete(f"/api/v1/mcp-client/servers/{server_id}", headers=ORIGIN_HEADER)
    assert resp.status_code == 204

    # Verify deleted
    resp = await client.get("/api/v1/mcp-client/servers", headers=ORIGIN_HEADER)
    assert resp.status_code == 200
    assert all(s["id"] != server_id for s in resp.json())


# ─── AC5: List servers after creating ───


@pytest.mark.asyncio
async def test_ac5_list_servers_after_create(mcp_authed_client):
    """AC5: GET /api/v1/mcp-client/servers → 200 + list with created server."""
    client, _ = mcp_authed_client
    await client.post(
        "/api/v1/mcp-client/servers",
        json={"name": "Server A", "url": "http://a:8000", "enabled": True},
        headers=ORIGIN_HEADER,
    )
    await client.post(
        "/api/v1/mcp-client/servers",
        json={"name": "Server B", "url": "http://b:8000", "enabled": False},
        headers=ORIGIN_HEADER,
    )
    resp = await client.get("/api/v1/mcp-client/servers", headers=ORIGIN_HEADER)
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 2
    names = [s["name"] for s in data]
    assert "Server A" in names
    assert "Server B" in names


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


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


# ─── AC7: Execute tool on non-existent server returns 404 ───


@pytest.mark.asyncio
async def test_ac7_execute_on_nonexistent_server(mcp_authed_client):
    """AC7: POST /api/v1/mcp-client/servers/{invalid_id}/execute → 400 (invalid UUID) or 404."""
    client, _ = mcp_authed_client
    resp = await client.post(
        "/api/v1/mcp-client/servers/00000000-0000-0000-0000-000000000000/execute",
        json={"tool_name": "test", "arguments": {}},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 404


# ─── AC8: Tool registry integration registers external tools ───


@pytest.mark.asyncio
async def test_ac8_tool_registry_integration(mcp_authed_client):
    """AC8: MCP Client plugin registers tools in the AI tool registry."""
    from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry
    from app.plugins.builtins.mcp_client.tool_registry_integration import unregister_all_external_tools, PLUGIN_NAME

    registry = get_tool_registry()
    # Ensure clean state
    unregister_all_external_tools()

    # Verify no MCP tools initially
    mcp_tools = [t for t in registry.get_all() if t.plugin_name == PLUGIN_NAME]
    assert len(mcp_tools) == 0

    # The integration module exists and has correct functions
    from app.plugins.builtins.mcp_client.tool_registry_integration import sync_external_tools, _make_tool_name
    assert callable(sync_external_tools)
    assert _make_tool_name("My Server", "search") == "mcp__my_server__search"