Skip to content

Commit ae37120

Browse files
KludexKludex
committed
refactor: drop test classes
1 parent 656f887 commit ae37120

12 files changed

+1611
-1562
lines changed

‎tests/client/test_output_schema_validation.py‎

Lines changed: 173 additions & 172 deletions
Original file line numberDiff line numberDiff line change
@@ -38,175 +38,176 @@ def selective_mock(instance: Any = None, schema: Any = None, *args: Any, **kwarg
3838
yield
3939

4040

41-
class TestClientOutputSchemaValidation:
42-
"""Test client-side validation of structured output from tools"""
43-
44-
@pytest.mark.anyio
45-
async def test_tool_structured_output_client_side_validation_basemodel(self):
46-
"""Test that client validates structured content against schema for BaseModel outputs"""
47-
# Create a malicious low-level server that returns invalid structured content
48-
server = Server("test-server")
49-
50-
# Define the expected schema for our tool
51-
output_schema = {
52-
"type": "object",
53-
"properties": {"name": {"type": "string", "title": "Name"}, "age": {"type": "integer", "title": "Age"}},
54-
"required": ["name", "age"],
55-
"title": "UserOutput",
56-
}
57-
58-
@server.list_tools()
59-
async def list_tools():
60-
return [
61-
Tool(
62-
name="get_user",
63-
description="Get user data",
64-
input_schema={"type": "object"},
65-
output_schema=output_schema,
66-
)
67-
]
68-
69-
@server.call_tool()
70-
async def call_tool(name: str, arguments: dict[str, Any]):
71-
# Return invalid structured content - age is string instead of integer
72-
# The low-level server will wrap this in CallToolResult
73-
return {"name": "John", "age": "invalid"} # Invalid: age should be int
74-
75-
# Test that client validates the structured content
76-
with bypass_server_output_validation():
77-
async with Client(server) as client:
78-
# The client validates structured content and should raise an error
79-
with pytest.raises(RuntimeError) as exc_info:
80-
await client.call_tool("get_user", {})
81-
# Verify it's a validation error
82-
assert "Invalid structured content returned by tool get_user" in str(exc_info.value)
83-
84-
@pytest.mark.anyio
85-
async def test_tool_structured_output_client_side_validation_primitive(self):
86-
"""Test that client validates structured content for primitive outputs"""
87-
server = Server("test-server")
88-
89-
# Primitive types are wrapped in {"result": value}
90-
output_schema = {
91-
"type": "object",
92-
"properties": {"result": {"type": "integer", "title": "Result"}},
93-
"required": ["result"],
94-
"title": "calculate_Output",
95-
}
96-
97-
@server.list_tools()
98-
async def list_tools():
99-
return [
100-
Tool(
101-
name="calculate",
102-
description="Calculate something",
103-
input_schema={"type": "object"},
104-
output_schema=output_schema,
105-
)
106-
]
107-
108-
@server.call_tool()
109-
async def call_tool(name: str, arguments: dict[str, Any]):
110-
# Return invalid structured content - result is string instead of integer
111-
return {"result": "not_a_number"} # Invalid: should be int
112-
113-
with bypass_server_output_validation():
114-
async with Client(server) as client:
115-
# The client validates structured content and should raise an error
116-
with pytest.raises(RuntimeError) as exc_info:
117-
await client.call_tool("calculate", {})
118-
assert "Invalid structured content returned by tool calculate" in str(exc_info.value)
119-
120-
@pytest.mark.anyio
121-
async def test_tool_structured_output_client_side_validation_dict_typed(self):
122-
"""Test that client validates dict[str, T] structured content"""
123-
server = Server("test-server")
124-
125-
# dict[str, int] schema
126-
output_schema = {"type": "object", "additionalProperties": {"type": "integer"}, "title": "get_scores_Output"}
127-
128-
@server.list_tools()
129-
async def list_tools():
130-
return [
131-
Tool(
132-
name="get_scores",
133-
description="Get scores",
134-
input_schema={"type": "object"},
135-
output_schema=output_schema,
136-
)
137-
]
138-
139-
@server.call_tool()
140-
async def call_tool(name: str, arguments: dict[str, Any]):
141-
# Return invalid structured content - values should be integers
142-
return {"alice": "100", "bob": "85"} # Invalid: values should be int
143-
144-
with bypass_server_output_validation():
145-
async with Client(server) as client:
146-
# The client validates structured content and should raise an error
147-
with pytest.raises(RuntimeError) as exc_info:
148-
await client.call_tool("get_scores", {})
149-
assert "Invalid structured content returned by tool get_scores" in str(exc_info.value)
150-
151-
@pytest.mark.anyio
152-
async def test_tool_structured_output_client_side_validation_missing_required(self):
153-
"""Test that client validates missing required fields"""
154-
server = Server("test-server")
155-
156-
output_schema = {
157-
"type": "object",
158-
"properties": {"name": {"type": "string"}, "age": {"type": "integer"}, "email": {"type": "string"}},
159-
"required": ["name", "age", "email"], # All fields required
160-
"title": "PersonOutput",
161-
}
162-
163-
@server.list_tools()
164-
async def list_tools():
165-
return [
166-
Tool(
167-
name="get_person",
168-
description="Get person data",
169-
input_schema={"type": "object"},
170-
output_schema=output_schema,
171-
)
172-
]
173-
174-
@server.call_tool()
175-
async def call_tool(name: str, arguments: dict[str, Any]):
176-
# Return structured content missing required field 'email'
177-
return {"name": "John", "age": 30} # Missing required 'email'
178-
179-
with bypass_server_output_validation():
180-
async with Client(server) as client:
181-
# The client validates structured content and should raise an error
182-
with pytest.raises(RuntimeError) as exc_info:
183-
await client.call_tool("get_person", {})
184-
assert "Invalid structured content returned by tool get_person" in str(exc_info.value)
185-
186-
@pytest.mark.anyio
187-
async def test_tool_not_listed_warning(self, caplog: pytest.LogCaptureFixture):
188-
"""Test that client logs warning when tool is not in list_tools but has output_schema"""
189-
server = Server("test-server")
190-
191-
@server.list_tools()
192-
async def list_tools() -> list[Tool]:
193-
# Return empty list - tool is not listed
194-
return []
195-
196-
@server.call_tool()
197-
async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
198-
# Server still responds to the tool call with structured content
199-
return {"result": 42}
200-
201-
# Set logging level to capture warnings
202-
caplog.set_level(logging.WARNING)
203-
204-
with bypass_server_output_validation():
205-
async with Client(server) as client:
206-
# Call a tool that wasn't listed
207-
result = await client.call_tool("mystery_tool", {})
208-
assert result.structured_content == {"result": 42}
209-
assert result.is_error is False
210-
211-
# Check that warning was logged
212-
assert "Tool mystery_tool not listed" in caplog.text
41+
@pytest.mark.anyio
42+
async def test_tool_structured_output_client_side_validation_basemodel():
43+
"""Test that client validates structured content against schema for BaseModel outputs"""
44+
# Create a malicious low-level server that returns invalid structured content
45+
server = Server("test-server")
46+
47+
# Define the expected schema for our tool
48+
output_schema = {
49+
"type": "object",
50+
"properties": {"name": {"type": "string", "title": "Name"}, "age": {"type": "integer", "title": "Age"}},
51+
"required": ["name", "age"],
52+
"title": "UserOutput",
53+
}
54+
55+
@server.list_tools()
56+
async def list_tools():
57+
return [
58+
Tool(
59+
name="get_user",
60+
description="Get user data",
61+
input_schema={"type": "object"},
62+
output_schema=output_schema,
63+
)
64+
]
65+
66+
@server.call_tool()
67+
async def call_tool(name: str, arguments: dict[str, Any]):
68+
# Return invalid structured content - age is string instead of integer
69+
# The low-level server will wrap this in CallToolResult
70+
return {"name": "John", "age": "invalid"} # Invalid: age should be int
71+
72+
# Test that client validates the structured content
73+
with bypass_server_output_validation():
74+
async with Client(server) as client:
75+
# The client validates structured content and should raise an error
76+
with pytest.raises(RuntimeError) as exc_info:
77+
await client.call_tool("get_user", {})
78+
# Verify it's a validation error
79+
assert "Invalid structured content returned by tool get_user" in str(exc_info.value)
80+
81+
82+
@pytest.mark.anyio
83+
async def test_tool_structured_output_client_side_validation_primitive():
84+
"""Test that client validates structured content for primitive outputs"""
85+
server = Server("test-server")
86+
87+
# Primitive types are wrapped in {"result": value}
88+
output_schema = {
89+
"type": "object",
90+
"properties": {"result": {"type": "integer", "title": "Result"}},
91+
"required": ["result"],
92+
"title": "calculate_Output",
93+
}
94+
95+
@server.list_tools()
96+
async def list_tools():
97+
return [
98+
Tool(
99+
name="calculate",
100+
description="Calculate something",
101+
input_schema={"type": "object"},
102+
output_schema=output_schema,
103+
)
104+
]
105+
106+
@server.call_tool()
107+
async def call_tool(name: str, arguments: dict[str, Any]):
108+
# Return invalid structured content - result is string instead of integer
109+
return {"result": "not_a_number"} # Invalid: should be int
110+
111+
with bypass_server_output_validation():
112+
async with Client(server) as client:
113+
# The client validates structured content and should raise an error
114+
with pytest.raises(RuntimeError) as exc_info:
115+
await client.call_tool("calculate", {})
116+
assert "Invalid structured content returned by tool calculate" in str(exc_info.value)
117+
118+
119+
@pytest.mark.anyio
120+
async def test_tool_structured_output_client_side_validation_dict_typed():
121+
"""Test that client validates dict[str, T] structured content"""
122+
server = Server("test-server")
123+
124+
# dict[str, int] schema
125+
output_schema = {"type": "object", "additionalProperties": {"type": "integer"}, "title": "get_scores_Output"}
126+
127+
@server.list_tools()
128+
async def list_tools():
129+
return [
130+
Tool(
131+
name="get_scores",
132+
description="Get scores",
133+
input_schema={"type": "object"},
134+
output_schema=output_schema,
135+
)
136+
]
137+
138+
@server.call_tool()
139+
async def call_tool(name: str, arguments: dict[str, Any]):
140+
# Return invalid structured content - values should be integers
141+
return {"alice": "100", "bob": "85"} # Invalid: values should be int
142+
143+
with bypass_server_output_validation():
144+
async with Client(server) as client:
145+
# The client validates structured content and should raise an error
146+
with pytest.raises(RuntimeError) as exc_info:
147+
await client.call_tool("get_scores", {})
148+
assert "Invalid structured content returned by tool get_scores" in str(exc_info.value)
149+
150+
151+
@pytest.mark.anyio
152+
async def test_tool_structured_output_client_side_validation_missing_required():
153+
"""Test that client validates missing required fields"""
154+
server = Server("test-server")
155+
156+
output_schema = {
157+
"type": "object",
158+
"properties": {"name": {"type": "string"}, "age": {"type": "integer"}, "email": {"type": "string"}},
159+
"required": ["name", "age", "email"], # All fields required
160+
"title": "PersonOutput",
161+
}
162+
163+
@server.list_tools()
164+
async def list_tools():
165+
return [
166+
Tool(
167+
name="get_person",
168+
description="Get person data",
169+
input_schema={"type": "object"},
170+
output_schema=output_schema,
171+
)
172+
]
173+
174+
@server.call_tool()
175+
async def call_tool(name: str, arguments: dict[str, Any]):
176+
# Return structured content missing required field 'email'
177+
return {"name": "John", "age": 30} # Missing required 'email'
178+
179+
with bypass_server_output_validation():
180+
async with Client(server) as client:
181+
# The client validates structured content and should raise an error
182+
with pytest.raises(RuntimeError) as exc_info:
183+
await client.call_tool("get_person", {})
184+
assert "Invalid structured content returned by tool get_person" in str(exc_info.value)
185+
186+
187+
@pytest.mark.anyio
188+
async def test_tool_not_listed_warning(caplog: pytest.LogCaptureFixture):
189+
"""Test that client logs warning when tool is not in list_tools but has output_schema"""
190+
server = Server("test-server")
191+
192+
@server.list_tools()
193+
async def list_tools() -> list[Tool]:
194+
# Return empty list - tool is not listed
195+
return []
196+
197+
@server.call_tool()
198+
async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
199+
# Server still responds to the tool call with structured content
200+
return {"result": 42}
201+
202+
# Set logging level to capture warnings
203+
caplog.set_level(logging.WARNING)
204+
205+
with bypass_server_output_validation():
206+
async with Client(server) as client:
207+
# Call a tool that wasn't listed
208+
result = await client.call_tool("mystery_tool", {})
209+
assert result.structured_content == {"result": 42}
210+
assert result.is_error is False
211+
212+
# Check that warning was logged
213+
assert "Tool mystery_tool not listed" in caplog.text

0 commit comments

Comments
 (0)