直接对工具进行单元测试
报告错误代码
复制
询问AI
def test_add():
assert add(2, 3) == 5
async def test_async_tool():
result = await fetch_user("123")
assert result["id"] == "123"
使用上下文的工具
get_context() 的工具需要有一个正在进行的请求。通过连接真实服务器的集成测试来验证它们,或者调整你的 tool 设计,使依赖上下文的部分便于进行 mock:
报告错误代码
复制
询问AI
@tool(description="Process with logging")
async def process(data: str) -> dict:
ctx = get_context()
await ctx.info("Processing")
return do_work(data) # 单独测试 do_work
def test_do_work():
assert do_work("input") == {"result": "output"}
使用 MCPClient 进行集成测试
报告错误代码
复制
询问AI
import pytest
from dedalus_mcp import MCPServer, tool
from dedalus_mcp.client import MCPClient
@tool(description="Add")
def add(a: int, b: int) -> int:
return a + b
@pytest.fixture
async def server():
server = MCPServer("test")
server.collect(add)
# 在后台启动服务器
task = asyncio.create_task(server.serve())
await asyncio.sleep(0.1) # 等待启动
yield server
task.cancel()
@pytest.fixture
async def client(server):
client = await MCPClient.connect("http://127.0.0.1:8000/mcp")
yield client
await client.close()
async def test_call_tool(client):
result = await client.call_tool("add", {"a": 2, "b": 3})
assert result.content[0].text == "5"
测试注册
报告错误代码
复制
询问AI
def test_registration():
server = MCPServer("test")
server.collect(add, multiply)
names = list(server.tool_names)
assert "add" in names
assert "multiply" in names
隔离
报告错误代码
复制
询问AI
def test_a():
server = MCPServer("test-a")
server.collect(tool_a)
# 无需清理
def test_b():
server = MCPServer("test-b")
server.collect(tool_b)
# 完全独立