78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""MCP server API 测试（respx mock MCP server HTTP）。"""
|
|
|
|
from collections.abc import AsyncGenerator
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.main import app
|
|
|
|
|
|
@pytest.fixture
|
|
async def client() -> AsyncGenerator[AsyncClient, None]:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as http_client:
|
|
yield http_client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_register_mcp_server(client: AsyncClient) -> None:
|
|
jrpc_result = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"result": {"tools": [{"name": "weather", "description": "查天气"}]},
|
|
}
|
|
respx.post("http://mcp.local/").mock(return_value=httpx.Response(200, json=jrpc_result))
|
|
resp = await client.post(
|
|
"/api/v1/tools/mcp-servers",
|
|
json={"name": "weather-srv", "endpoint_url": "http://mcp.local"},
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["name"] == "weather-srv"
|
|
assert data["has_auth_token"] is False
|
|
assert len(data["tools"]) == 1
|
|
assert data["tools"][0]["name"] == "weather"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_list_and_delete_mcp_server(client: AsyncClient) -> None:
|
|
jrpc_result = {"jsonrpc": "2.0", "id": 1, "result": {"tools": []}}
|
|
respx.post("http://mcp.local/").mock(return_value=httpx.Response(200, json=jrpc_result))
|
|
create = await client.post(
|
|
"/api/v1/tools/mcp-servers",
|
|
json={"name": "srv", "endpoint_url": "http://mcp.local"},
|
|
)
|
|
sid = create.json()["id"]
|
|
|
|
listed = await client.get("/api/v1/tools/mcp-servers")
|
|
assert listed.status_code == 200
|
|
assert len(listed.json()) == 1
|
|
|
|
deleted = await client.delete(f"/api/v1/tools/mcp-servers/{sid}")
|
|
assert deleted.status_code == 204
|
|
|
|
listed2 = await client.get("/api/v1/tools/mcp-servers")
|
|
assert listed2.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_register_duplicate_conflict(client: AsyncClient) -> None:
|
|
jrpc_result = {"jsonrpc": "2.0", "id": 1, "result": {"tools": []}}
|
|
respx.post("http://mcp.local/").mock(return_value=httpx.Response(200, json=jrpc_result))
|
|
await client.post(
|
|
"/api/v1/tools/mcp-servers",
|
|
json={"name": "dup", "endpoint_url": "http://mcp.local"},
|
|
)
|
|
resp = await client.post(
|
|
"/api/v1/tools/mcp-servers",
|
|
json={"name": "dup", "endpoint_url": "http://mcp.local"},
|
|
)
|
|
assert resp.status_code == 409
|
|
assert resp.json()["error"]["code"] == "conflict"
|