102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
"""Shared test helpers (standard library ``unittest`` only).
|
|
|
|
Run all tests: python -m unittest
|
|
Skip subprocess tests: MCP_CALC_SKIP_SLOW=1 python -m unittest
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import math
|
|
import os
|
|
import unittest
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any
|
|
|
|
from mcp import Client
|
|
|
|
from mcp_calculator.config import Limits, load_settings
|
|
from mcp_calculator.executor import Executor
|
|
from mcp_calculator.ops.context import set_limits
|
|
from mcp_calculator.server import create_server
|
|
|
|
# Tests that start subprocesses, worker pools or servers (a few seconds in total).
|
|
slow = unittest.skipIf(os.environ.get("MCP_CALC_SKIP_SLOW") == "1", "slow test skipped (MCP_CALC_SKIP_SLOW=1)")
|
|
|
|
|
|
class MathTestCase(unittest.TestCase):
|
|
"""Resets the global limits around every test."""
|
|
|
|
def setUp(self) -> None:
|
|
set_limits(Limits())
|
|
self.addCleanup(set_limits, Limits())
|
|
|
|
def assertApprox(self, actual: float, expected: float, rel: float = 1e-6) -> None:
|
|
self.assertTrue(math.isclose(actual, expected, rel_tol=rel, abs_tol=1e-12), f"{actual!r} != approx {expected!r}")
|
|
|
|
|
|
@asynccontextmanager
|
|
async def in_process_client(**env: str) -> AsyncIterator[Client]:
|
|
"""MCP client connected in-process to a server with the inline executor (fast, no subprocesses).
|
|
|
|
Opened inside each test: anyio cancel scopes must be entered and exited in the same task,
|
|
which rules out asyncSetUp/asyncTearDown.
|
|
"""
|
|
settings = load_settings([], env={"MCP_CALC_EXECUTOR": "inline", **env})
|
|
async with Client(create_server(Executor(settings))) as client:
|
|
yield client
|
|
|
|
|
|
async def asgi_request(app, method: str, path: str, *, headers: dict[str, str] | None = None, body: Any = None) -> tuple[int, Any]:
|
|
"""Call an ASGI app directly (no HTTP server, no extra dependency) and return (status, JSON body)."""
|
|
payload = json.dumps(body).encode() if body is not None else b""
|
|
scope = {
|
|
"type": "http",
|
|
"asgi": {"version": "3.0"},
|
|
"http_version": "1.1",
|
|
"method": method,
|
|
"scheme": "http",
|
|
"path": path,
|
|
"raw_path": path.encode(),
|
|
"query_string": b"",
|
|
"root_path": "",
|
|
"headers": [(b"content-type", b"application/json")] + [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()],
|
|
"client": ("127.0.0.1", 50000),
|
|
"server": ("testserver", 80),
|
|
}
|
|
delivered = False
|
|
|
|
async def receive():
|
|
nonlocal delivered
|
|
if not delivered:
|
|
delivered = True
|
|
return {"type": "http.request", "body": payload, "more_body": False}
|
|
await asyncio.Event().wait()
|
|
|
|
messages = []
|
|
|
|
async def send(message):
|
|
messages.append(message)
|
|
|
|
await app(scope, receive, send)
|
|
status = next(m["status"] for m in messages if m["type"] == "http.response.start")
|
|
raw = b"".join(m.get("body", b"") for m in messages if m["type"] == "http.response.body")
|
|
return status, json.loads(raw) if raw else None
|
|
|
|
|
|
class ClientTestCase(unittest.IsolatedAsyncioTestCase):
|
|
async def call(self, tool: str, **arguments: Any) -> dict[str, Any]:
|
|
async with in_process_client() as client:
|
|
result = await client.call_tool(tool, arguments)
|
|
self.assertFalse(result.is_error, result.content[0].text)
|
|
self.assertIsNotNone(result.structured_content)
|
|
self.assertEqual(json.loads(result.content[0].text), result.structured_content)
|
|
return result.structured_content
|
|
|
|
async def call_error(self, tool: str, **arguments: Any) -> str:
|
|
async with in_process_client() as client:
|
|
result = await client.call_tool(tool, arguments)
|
|
self.assertTrue(result.is_error, result.structured_content)
|
|
return result.content[0].text
|