108 lines
4.7 KiB
Python
108 lines
4.7 KiB
Python
"""Tool layer tests through a real MCP client (in-process, inline executor)."""
|
|
|
|
from mcp_calculator.metrics import REGISTRY
|
|
from tests.support import ClientTestCase, in_process_client
|
|
|
|
EXPECTED_TOOLS = {
|
|
"evaluate",
|
|
"transform",
|
|
"solve",
|
|
"calculus",
|
|
"analyze_function",
|
|
"solve_ode",
|
|
"matrix",
|
|
"statistics",
|
|
"regression",
|
|
"probability",
|
|
"hypothesis_test",
|
|
"number_theory",
|
|
"convert_units",
|
|
}
|
|
|
|
|
|
def tool_calls(tool: str, status: str, client: str = "local") -> float:
|
|
return REGISTRY.get_sample_value("mcp_calc_tool_calls_total", {"tool": tool, "client": client, "status": status}) or 0.0
|
|
|
|
|
|
class DiscoveryTest(ClientTestCase):
|
|
async def test_tool_list(self):
|
|
async with in_process_client() as client:
|
|
tools = {t.name: t for t in (await client.list_tools()).tools}
|
|
self.assertEqual(set(tools), EXPECTED_TOOLS)
|
|
for tool in tools.values():
|
|
with self.subTest(tool=tool.name):
|
|
self.assertGreater(len(tool.description or ""), 40)
|
|
self.assertIs(tool.annotations.read_only_hint, True)
|
|
self.assertIs(tool.annotations.open_world_hint, False)
|
|
|
|
async def test_schema_has_descriptions_and_enums(self):
|
|
async with in_process_client() as client:
|
|
tools = {t.name: t for t in (await client.list_tools()).tools}
|
|
props = tools["calculus"].input_schema["properties"]
|
|
self.assertIn("derivative", props["operation"]["enum"])
|
|
self.assertTrue(props["variable"]["description"])
|
|
|
|
async def test_instructions_are_sent(self):
|
|
async with in_process_client() as client:
|
|
self.assertIn("Which tool", client.instructions or "")
|
|
self.assertEqual(client.server_info.name, "calculator")
|
|
|
|
async def test_syntax_resource(self):
|
|
async with in_process_client() as client:
|
|
resources = (await client.list_resources()).resources
|
|
self.assertTrue(any(str(r.uri) == "calculator://syntax" for r in resources))
|
|
|
|
|
|
class ToolCallTest(ClientTestCase):
|
|
async def test_evaluate(self):
|
|
self.assertEqual((await self.call("evaluate", expression="2^64"))["exact"], "18446744073709551616")
|
|
|
|
async def test_solve_accepts_single_string(self):
|
|
r = await self.call("solve", equations="x^2 = 9", domain="real")
|
|
self.assertEqual(r["solutions"], [{"x": "-3"}, {"x": "3"}])
|
|
|
|
async def test_calculus_and_chaining(self):
|
|
derivative = await self.call("calculus", operation="derivative", expression="x^3 - 6x^2 + 9x")
|
|
# 'result' is valid input syntax for the next call
|
|
roots = await self.call("solve", equations=derivative["result"])
|
|
self.assertEqual(roots["solutions"], [{"x": "1"}, {"x": "3"}])
|
|
|
|
async def test_matrix(self):
|
|
self.assertEqual((await self.call("matrix", operation="determinant", matrix=[[1, 2], [3, 4]]))["result"], "-2")
|
|
|
|
async def test_statistics(self):
|
|
self.assertEqual((await self.call("statistics", data=[1, 2, 3, 4]))["mean"], 2.5)
|
|
|
|
async def test_number_theory_big_ints_as_strings(self):
|
|
r = await self.call("number_theory", operation="is_prime", values=["170141183460469231731687303715884105727"])
|
|
self.assertIs(r["result"], True)
|
|
|
|
async def test_convert_units(self):
|
|
r = await self.call("convert_units", quantity="1 mi", to_unit="km")
|
|
self.assertAlmostEqual(r["magnitude"], 1.609344, places=9)
|
|
|
|
|
|
class ToolErrorTest(ClientTestCase):
|
|
async def test_math_error_is_tool_error(self):
|
|
self.assertIn("singular", await self.call_error("matrix", operation="inverse", matrix=[[1, 2], [2, 4]]))
|
|
|
|
async def test_injection_is_rejected(self):
|
|
text = await self.call_error("evaluate", expression="__import__('os').system('id')")
|
|
self.assertRegex(text, "unsupported character|not allowed")
|
|
|
|
async def test_schema_validation_error(self):
|
|
self.assertIn("operation", await self.call_error("matrix", operation="explode", matrix=[[1]]))
|
|
|
|
|
|
class ToolMetricsTest(ClientTestCase):
|
|
async def test_calls_are_counted_by_outcome(self):
|
|
ok_before, error_before = tool_calls("evaluate", "ok"), tool_calls("evaluate", "error")
|
|
await self.call("evaluate", expression="1+1")
|
|
await self.call_error("evaluate", expression="1/")
|
|
self.assertEqual(tool_calls("evaluate", "ok"), ok_before + 1)
|
|
self.assertEqual(tool_calls("evaluate", "error"), error_before + 1)
|
|
|
|
async def test_duration_is_recorded(self):
|
|
before = REGISTRY.get_sample_value("mcp_calc_tool_duration_seconds_count", {"tool": "statistics"}) or 0.0
|
|
await self.call("statistics", data=[1, 2, 3])
|
|
self.assertEqual(REGISTRY.get_sample_value("mcp_calc_tool_duration_seconds_count", {"tool": "statistics"}), before + 1)
|