ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-21 13:04:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat: introduce inline snapshots (#1605)

Browse source 
Co-authored-by: William Easton <strawgate@users.noreply.github.com>
Co-authored-by: marvin-context-protocol[bot] <225465937+marvin-context-protocol[bot]@users.noreply.github.com>
This commit is contained in:
William Easton 2025-08-25 09:08:55 -05:00 • committed by GitHub
commit 94b1eb9d6e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 540 additions and 279 deletions
Download patch file Download diff file Expand all files Collapse all files

1
pyproject.toml
View file

@ -45,6 +45,7 @@ dev = [
"copychat>=0.5.2",
"dirty-equals>=0.9.0",
"fastapi>=0.115.12",
"inline-snapshot[dirty-equals]>=0.27.2",
"ipython>=8.12.3",
"pdbpp>=0.10.3",
"pre-commit",

8
tests/conftest.py
View file

@ -7,3 +7,11 @@ def pytest_collection_modifyitems(items):
# Check if the test is in the integration_tests folder
if "integration_tests" in str(item.fspath):
item.add_marker(pytest.mark.integration)
@pytest.fixture(autouse=True)
def import_rich_rule():
# What a hack
import rich.rule # noqa: F401
yield

310
tests/tools/test_tool.py
View file

@ -3,6 +3,8 @@ from dataclasses import dataclass
from typing import Annotated, Any
import pytest
from dirty_equals import HasName
from inline_snapshot import snapshot
from mcp.types import (
AudioContent,
EmbeddedResource,
@ -13,7 +15,7 @@ from mcp.types import (
from pydantic import AnyUrl, BaseModel, Field, TypeAdapter
from typing_extensions import TypedDict
from fastmcp.tools.tool import Tool, _convert_to_content
from fastmcp.tools.tool import Tool, ToolResult, _convert_to_content
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.tests import caplog_for_fastmcp
from fastmcp.utilities.types import Audio, File, Image
@ -29,20 +31,30 @@ class TestToolFromFunction:
tool = Tool.from_function(add)
assert tool.name == "add"
assert tool.description == "Add two numbers."
assert len(tool.parameters["properties"]) == 2
assert tool.parameters["properties"]["a"]["type"] == "integer"
assert tool.parameters["properties"]["b"]["type"] == "integer"
# With primitive wrapping, int return type becomes object with result property
expected_schema = {
"type": "object",
"properties": {"result": {"type": "integer", "title": "Result"}},
"required": ["result"],
"title": "_WrappedResult",
"x-fastmcp-wrap-result": True,
}
assert tool.output_schema == expected_schema
assert tool.model_dump(exclude_none=True) == snapshot(
{
"name": "add",
"description": "Add two numbers.",
"tags": set(),
"enabled": True,
"parameters": {
"properties": {
"a": {"title": "A", "type": "integer"},
"b": {"title": "B", "type": "integer"},
},
"required": ["a", "b"],
"type": "object",
},
"output_schema": {
"properties": {"result": {"title": "Result", "type": "integer"}},
"required": ["result"],
"title": "_WrappedResult",
"type": "object",
"x-fastmcp-wrap-result": True,
},
"fn": HasName("add"),
}
)
def test_meta_parameter(self):
"""Test that meta parameter is properly handled."""
@ -56,6 +68,7 @@ class TestToolFromFunction:
assert tool.meta == meta_data
mcp_tool = tool.to_mcp_tool()
# MCP tool includes fastmcp meta, so check that our meta is included
assert mcp_tool.meta is not None
assert meta_data.items() <= mcp_tool.meta.items()
@ -69,9 +82,27 @@ class TestToolFromFunction:
tool = Tool.from_function(fetch_data)
assert tool.name == "fetch_data"
assert tool.description == "Fetch data from URL."
assert tool.parameters["properties"]["url"]["type"] == "string"
assert tool.model_dump(exclude_none=True) == snapshot(
{
"name": "fetch_data",
"description": "Fetch data from URL.",
"tags": set(),
"enabled": True,
"parameters": {
"properties": {"url": {"title": "Url", "type": "string"}},
"required": ["url"],
"type": "object",
},
"output_schema": {
"properties": {"result": {"title": "Result", "type": "string"}},
"required": ["result"],
"title": "_WrappedResult",
"type": "object",
"x-fastmcp-wrap-result": True,
},
"fn": HasName("fetch_data"),
}
)
def test_callable_object(self):
class Adder:
@ -82,11 +113,30 @@ class TestToolFromFunction:
return x + y
tool = Tool.from_function(Adder())
assert tool.name == "Adder"
assert tool.description == "Adds two numbers."
assert len(tool.parameters["properties"]) == 2
assert tool.parameters["properties"]["x"]["type"] == "integer"
assert tool.parameters["properties"]["y"]["type"] == "integer"
assert tool.model_dump(exclude_none=True, exclude={"fn"}) == snapshot(
{
"name": "Adder",
"description": "Adds two numbers.",
"tags": set(),
"enabled": True,
"parameters": {
"properties": {
"x": {"title": "X", "type": "integer"},
"y": {"title": "Y", "type": "integer"},
},
"required": ["x", "y"],
"type": "object",
},
"output_schema": {
"properties": {"result": {"title": "Result", "type": "integer"}},
"required": ["result"],
"title": "_WrappedResult",
"type": "object",
"x-fastmcp-wrap-result": True,
},
}
)
def test_async_callable_object(self):
class Adder:
@ -97,11 +147,30 @@ class TestToolFromFunction:
return x + y
tool = Tool.from_function(Adder())
assert tool.name == "Adder"
assert tool.description == "Adds two numbers."
assert len(tool.parameters["properties"]) == 2
assert tool.parameters["properties"]["x"]["type"] == "integer"
assert tool.parameters["properties"]["y"]["type"] == "integer"
assert tool.model_dump(exclude_none=True, exclude={"fn"}) == snapshot(
{
"name": "Adder",
"description": "Adds two numbers.",
"tags": set(),
"enabled": True,
"parameters": {
"properties": {
"x": {"title": "X", "type": "integer"},
"y": {"title": "Y", "type": "integer"},
},
"required": ["x", "y"],
"type": "object",
},
"output_schema": {
"properties": {"result": {"title": "Result", "type": "integer"}},
"required": ["result"],
"title": "_WrappedResult",
"type": "object",
"x-fastmcp-wrap-result": True,
},
}
)
def test_pydantic_model_function(self):
"""Test registering a function that takes a Pydantic model."""
@ -116,20 +185,45 @@ class TestToolFromFunction:
tool = Tool.from_function(create_user)
assert tool.name == "create_user"
assert tool.description == "Create a new user."
assert "name" in tool.parameters["$defs"]["UserInput"]["properties"]
assert "age" in tool.parameters["$defs"]["UserInput"]["properties"]
assert "flag" in tool.parameters["properties"]
assert tool.model_dump(exclude_none=True) == snapshot(
{
"name": "create_user",
"description": "Create a new user.",
"tags": set(),
"enabled": True,
"parameters": {
"$defs": {
"UserInput": {
"properties": {
"name": {"title": "Name", "type": "string"},
"age": {"title": "Age", "type": "integer"},
},
"required": ["name", "age"],
"title": "UserInput",
"type": "object",
}
},
"properties": {
"user": {"$ref": "#/$defs/UserInput", "title": "User"},
"flag": {"title": "Flag", "type": "boolean"},
},
"required": ["user", "flag"],
"type": "object",
},
"output_schema": {"additionalProperties": True, "type": "object"},
"fn": HasName("create_user"),
}
)
async def test_tool_with_image_return(self):
def image_tool(data: bytes) -> Image:
return Image(data=data)
tool = Tool.from_function(image_tool)
assert tool.parameters["properties"]["data"]["type"] == "string"
assert tool.output_schema is None
result = await tool.run({"data": "test.png"})
assert tool.parameters["properties"]["data"]["type"] == "string"
assert isinstance(result.content[0], ImageContent)
async def test_tool_with_audio_return(self):
@ -137,9 +231,10 @@ class TestToolFromFunction:
return Audio(data=data)
tool = Tool.from_function(audio_tool)
assert tool.parameters["properties"]["data"]["type"] == "string"
assert tool.output_schema is None
result = await tool.run({"data": "test.wav"})
assert tool.parameters["properties"]["data"]["type"] == "string"
assert isinstance(result.content[0], AudioContent)
async def test_tool_with_file_return(self):
@ -147,15 +242,20 @@ class TestToolFromFunction:
return File(data=data, format="octet-stream")
tool = Tool.from_function(file_tool)
result = await tool.run({"data": "test.bin"})
assert tool.parameters["properties"]["data"]["type"] == "string"
assert len(result.content) == 1
assert isinstance(result.content[0], EmbeddedResource)
assert result.content[0].type == "resource"
assert hasattr(result.content[0], "resource")
resource = result.content[0].resource
assert resource.mimeType == "application/octet-stream"
assert tool.output_schema is None
result: ToolResult = await tool.run({"data": "test.bin"})
assert result.content[0].model_dump(exclude_none=True) == snapshot(
{
"type": "resource",
"resource": {
"uri": AnyUrl("file:///resource.octet-stream"),
"mimeType": "application/octet-stream",
"blob": "dGVzdC5iaW4=",
},
}
)
def test_non_callable_fn(self):
with pytest.raises(TypeError, match="not a callable object"):
@ -163,7 +263,18 @@ class TestToolFromFunction:
def test_lambda(self):
tool = Tool.from_function(lambda x: x, name="my_tool")
assert tool.name == "my_tool"
assert tool.model_dump(exclude_none=True, exclude={"fn"}) == snapshot(
{
"name": "my_tool",
"tags": set(),
"enabled": True,
"parameters": {
"properties": {"x": {"title": "X"}},
"required": ["x"],
"type": "object",
},
}
)
def test_lambda_with_no_name(self):
with pytest.raises(
@ -177,8 +288,25 @@ class TestToolFromFunction:
return _a + _b
tool = Tool.from_function(add)
assert tool.parameters["properties"]["_a"]["type"] == "integer"
assert tool.parameters["properties"]["_b"]["type"] == "integer"
assert tool.model_dump(
exclude_none=True, exclude={"output_schema", "fn"}
) == snapshot(
{
"name": "add",
"description": "Add two numbers.",
"tags": set(),
"enabled": True,
"parameters": {
"properties": {
"_a": {"title": "A", "type": "integer"},
"_b": {"title": "B", "type": "integer"},
},
"required": ["_a", "_b"],
"type": "object",
},
}
)
def test_tool_with_varargs_not_allowed(self):
def func(a: int, b: int, *args: int) -> int:
@ -209,10 +337,32 @@ class TestToolFromFunction:
obj = MyClass()
tool = Tool.from_function(obj.add)
assert tool.name == "add"
assert tool.description == "Add two numbers."
assert "self" not in tool.parameters["properties"]
assert tool.model_dump(exclude_none=True, exclude={"fn"}) == snapshot(
{
"name": "add",
"description": "Add two numbers.",
"tags": set(),
"enabled": True,
"parameters": {
"properties": {
"x": {"title": "X", "type": "integer"},
"y": {"title": "Y", "type": "integer"},
},
"required": ["x", "y"],
"type": "object",
},
"output_schema": {
"properties": {"result": {"title": "Result", "type": "integer"}},
"required": ["result"],
"title": "_WrappedResult",
"type": "object",
"x-fastmcp-wrap-result": True,
},
}
)
async def test_instance_method_with_varargs_not_allowed(self):
class MyClass:
def add(self, x: int, y: int, *args: int) -> int:
@ -322,6 +472,7 @@ class TestToolFromFunctionOutputSchema:
"x-fastmcp-wrap-result": True,
}
assert tool.output_schema == expected_schema
# # Note: Parameterized test - keeping original assertion for multiple parameter values
else:
# Object types remain unwrapped
assert tool.output_schema == base_schema
@ -339,8 +490,8 @@ class TestToolFromFunctionOutputSchema:
return 1
tool = Tool.from_function(func)
base_schema = TypeAdapter(annotation).json_schema()
base_schema = TypeAdapter(annotation).json_schema()
expected_schema = {
"type": "object",
"properties": {"result": {**base_schema, "title": "Result"}},
@ -405,8 +556,18 @@ class TestToolFromFunctionOutputSchema:
return Person(name="John", age=30)
tool = Tool.from_function(func)
expected_schema = compress_schema(TypeAdapter(Person).json_schema())
assert tool.output_schema == expected_schema
assert tool.output_schema == snapshot(
{
"properties": {
"name": {"title": "Name", "type": "string"},
"age": {"title": "Age", "type": "integer"},
},
"required": ["name", "age"],
"title": "Person",
"type": "object",
}
)
async def test_typeddict_return_annotation(self):
class Person(TypedDict):
@ -417,8 +578,17 @@ class TestToolFromFunctionOutputSchema:
return Person(name="John", age=30)
tool = Tool.from_function(func)
expected_schema = compress_schema(TypeAdapter(Person).json_schema())
assert tool.output_schema == expected_schema
assert tool.output_schema == snapshot(
{
"properties": {
"name": {"title": "Name", "type": "string"},
"age": {"title": "Age", "type": "integer"},
},
"required": ["name", "age"],
"title": "Person",
"type": "object",
}
)
async def test_unserializable_return_annotation(self):
class Unserializable:
@ -593,14 +763,15 @@ class TestToolFromFunctionOutputSchema:
# Don't specify output_schema - should infer and wrap
tool = Tool.from_function(func)
expected_schema = {
"type": "object",
"properties": {"result": {"type": "integer", "title": "Result"}},
"required": ["result"],
"title": "_WrappedResult",
"x-fastmcp-wrap-result": True,
}
assert tool.output_schema == expected_schema
assert tool.output_schema == snapshot(
{
"properties": {"result": {"title": "Result", "type": "integer"}},
"required": ["result"],
"title": "_WrappedResult",
"type": "object",
"x-fastmcp-wrap-result": True,
}
)
result = await tool.run({})
assert result.structured_content == {"result": 42}
@ -665,14 +836,15 @@ class TestToolFromFunctionOutputSchema:
# Inferred schema should wrap string type
tool = Tool.from_function(func)
expected_schema = {
"type": "object",
"properties": {"result": {"type": "string", "title": "Result"}},
"required": ["result"],
"title": "_WrappedResult",
"x-fastmcp-wrap-result": True,
}
assert tool.output_schema == expected_schema
assert tool.output_schema == snapshot(
{
"properties": {"result": {"title": "Result", "type": "string"}},
"required": ["result"],
"title": "_WrappedResult",
"type": "object",
"x-fastmcp-wrap-result": True,
}
)
result = await tool.run({})
# Unstructured content

477
tests/utilities/openapi/test_openapi.py
View file

@ -1,12 +1,17 @@
"""Tests for the OpenAPI parsing utilities."""
from collections.abc import Sequence
from typing import Any
import pytest
from fastapi import Body, FastAPI, Path, Query
from inline_snapshot import snapshot
from pydantic import BaseModel, Field
from fastmcp.utilities.openapi import (
HttpMethod,
HTTPRoute,
ParameterInfo,
_combine_schemas,
_replace_ref_with_defs,
parse_openapi_to_http_routes,
@ -107,7 +112,7 @@ def petstore_schema() -> dict[str, Any]:
@pytest.fixture
def parsed_petstore_routes(petstore_schema):
def parsed_petstore_routes(petstore_schema: dict[str, Any]) -> list[HTTPRoute]:
"""Return parsed routes from the PetStore schema."""
return parse_openapi_to_http_routes(petstore_schema)
@ -214,11 +219,30 @@ def bookstore_schema() -> dict[str, Any]:
@pytest.fixture
def parsed_bookstore_routes(bookstore_schema):
def parsed_bookstore_routes(bookstore_schema: dict[str, Any]) -> list[HTTPRoute]:
"""Return parsed routes from the BookStore schema."""
return parse_openapi_to_http_routes(bookstore_schema)
def get_route(
routes: list[HTTPRoute], method: HttpMethod, path: str
) -> HTTPRoute | None:
"""Get a route by method and path."""
return next((r for r in routes if r.method == method and r.path == path), None)
def get_parameter(
parameters: Sequence[ParameterInfo], name: str
) -> ParameterInfo | None:
"""Get a parameter by name."""
return next((p for p in parameters if p.name == name), None)
def dump_models(models: Sequence[BaseModel], **kwargs: Any) -> list[dict[str, Any]]:
"""Dump a list of models to a list of dictionaries."""
return [m.model_dump(**kwargs) for m in models]
# --- FastAPI App Fixtures --- #
@ -302,13 +326,13 @@ def fastapi_openapi_schema(fastapi_app) -> dict[str, Any]:
@pytest.fixture
def parsed_fastapi_routes(fastapi_openapi_schema):
def parsed_fastapi_routes(fastapi_openapi_schema: dict[str, Any]) -> list[HTTPRoute]:
"""Return parsed routes from a FastAPI OpenAPI schema."""
return parse_openapi_to_http_routes(fastapi_openapi_schema)
@pytest.fixture
def fastapi_route_map(parsed_fastapi_routes):
def fastapi_route_map(parsed_fastapi_routes: list[HTTPRoute]) -> dict[str, HTTPRoute]:
"""Return a dictionary of routes by operation ID."""
return {
r.operation_id: r for r in parsed_fastapi_routes if r.operation_id is not None
@ -484,128 +508,114 @@ def openapi_31_with_references() -> dict[str, Any]:
# --- Tests for PetStore schema --- #
def test_petstore_route_count(parsed_petstore_routes):
def test_petstore_route_count(parsed_petstore_routes: list[HTTPRoute]):
"""Test that parsing the PetStore schema correctly identifies the number of routes."""
assert len(parsed_petstore_routes) == 3
def test_petstore_get_pets_operation_id(parsed_petstore_routes):
def test_petstore_get_pets_operation_id(parsed_petstore_routes: list[HTTPRoute]):
"""Test that GET /pets operation_id is correctly parsed."""
get_pets = next(
(r for r in parsed_petstore_routes if r.method == "GET" and r.path == "/pets"),
None,
)
get_pets = get_route(parsed_petstore_routes, "GET", "/pets")
assert get_pets is not None
assert get_pets.operation_id == "listPets"
def test_petstore_query_parameter(parsed_petstore_routes):
def test_petstore_query_parameter(parsed_petstore_routes: list[HTTPRoute]):
"""Test that query parameter 'limit' is correctly parsed from the schema."""
get_pets = next(
(r for r in parsed_petstore_routes if r.method == "GET" and r.path == "/pets"),
None,
)
get_pets = get_route(parsed_petstore_routes, "GET", "/pets")
assert get_pets is not None
assert len(get_pets.parameters) == 1
param = get_pets.parameters[0]
assert param.name == "limit"
assert param.location == "query"
assert param.required is False
assert param.schema_.get("type") == "integer"
assert param.schema_.get("format") == "int32"
assert dump_models(get_pets.parameters, exclude_none=True) == snapshot(
[
{
"name": "limit",
"location": "query",
"required": False,
"schema_": {"type": "integer", "format": "int32"},
"description": "How many items to return",
}
]
)
def test_petstore_path_parameter(parsed_petstore_routes):
def test_petstore_path_parameter(parsed_petstore_routes: list[HTTPRoute]):
"""Test that path parameter 'petId' is correctly parsed from the schema."""
get_pet = next(
(
r
for r in parsed_petstore_routes
if r.method == "GET" and r.path == "/pets/{petId}"
),
None,
)
get_pet = get_route(parsed_petstore_routes, "GET", "/pets/{petId}")
assert get_pet is not None
path_param = next((p for p in get_pet.parameters if p.name == "petId"), None)
path_param = get_parameter(get_pet.parameters, "petId")
assert path_param is not None
assert path_param.location == "path"
assert path_param.required is True
assert path_param.schema_.get("type") == "string"
assert path_param.model_dump(exclude_none=True) == snapshot(
{
"name": "petId",
"location": "path",
"required": True,
"schema_": {"type": "string"},
"description": "The id of the pet",
}
)
def test_petstore_header_parameters(parsed_petstore_routes):
def test_petstore_header_parameters(parsed_petstore_routes: list[HTTPRoute]):
"""Test that header parameters are correctly parsed from the schema."""
get_pet = next(
(
r
for r in parsed_petstore_routes
if r.method == "GET" and r.path == "/pets/{petId}"
),
None,
get_pet = get_route(parsed_petstore_routes, "GET", "/pets/{petId}")
assert get_pet is not None
header_params = [p for p in get_pet.parameters if p.location == "header"]
assert dump_models(header_params, exclude_none=True) == snapshot(
[
{
"name": "X-Request-ID",
"location": "header",
"required": False,
"schema_": {"type": "string", "format": "uuid"},
},
{
"name": "traceId",
"location": "header",
"required": False,
"schema_": {"type": "string"},
"description": "Common trace ID",
},
]
)
assert get_pet is not None
header_params = [p for p in get_pet.parameters if p.location == "header"]
assert len(header_params) == 2
def test_petstore_header_parameter_names(parsed_petstore_routes):
"""Test that header parameter names are correctly parsed."""
get_pet = next(
(
r
for r in parsed_petstore_routes
if r.method == "GET" and r.path == "/pets/{petId}"
),
None,
)
assert get_pet is not None
header_params = [p for p in get_pet.parameters if p.location == "header"]
header_names = [p.name for p in header_params]
assert "X-Request-ID" in header_names
assert "traceId" in header_names
def test_petstore_path_level_parameters(parsed_petstore_routes):
def test_petstore_path_level_parameters(parsed_petstore_routes: list[HTTPRoute]):
"""Test that path-level parameters are correctly merged into the operation."""
get_pet = next(
(
r
for r in parsed_petstore_routes
if r.method == "GET" and r.path == "/pets/{petId}"
),
None,
)
get_pet = get_route(parsed_petstore_routes, "GET", "/pets/{petId}")
assert get_pet is not None
trace_param = next((p for p in get_pet.parameters if p.name == "traceId"), None)
trace_param = get_parameter(get_pet.parameters, "traceId")
assert trace_param is not None
assert trace_param.location == "header"
assert trace_param.required is False
def test_petstore_request_body_reference_resolution(parsed_petstore_routes):
"""Test that request body references are correctly resolved."""
create_pet = next(
(r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"),
None,
assert trace_param.model_dump(exclude_none=True) == snapshot(
{
"name": "traceId",
"location": "header",
"required": False,
"schema_": {"type": "string"},
"description": "Common trace ID",
}
)
def test_petstore_request_body_reference_resolution(
parsed_petstore_routes: list[HTTPRoute],
):
"""Test that request body references are correctly resolved."""
create_pet = get_route(parsed_petstore_routes, "POST", "/pets")
assert create_pet is not None
assert create_pet.request_body is not None
assert create_pet.request_body.required is True
assert "application/json" in create_pet.request_body.content_schema
def test_petstore_schema_reference_resolution(parsed_petstore_routes):
def test_petstore_schema_reference_resolution(parsed_petstore_routes: list[HTTPRoute]):
"""Test that schema references in request bodies are correctly resolved."""
create_pet = next(
(r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"),
None,
)
create_pet = get_route(parsed_petstore_routes, "POST", "/pets")
assert create_pet is not None
assert create_pet.request_body is not None
@ -617,12 +627,9 @@ def test_petstore_schema_reference_resolution(parsed_petstore_routes):
assert "tag" in properties
def test_petstore_required_fields_resolution(parsed_petstore_routes):
def test_petstore_required_fields_resolution(parsed_petstore_routes: list[HTTPRoute]):
"""Test that required fields are correctly resolved from referenced schemas."""
create_pet = next(
(r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"),
None,
)
create_pet = get_route(parsed_petstore_routes, "POST", "/pets")
assert create_pet is not None
assert create_pet.request_body is not None
@ -630,7 +637,7 @@ def test_petstore_required_fields_resolution(parsed_petstore_routes):
assert json_schema.get("required") == ["id", "name"]
def test_tags_parsing_in_petstore_routes(parsed_petstore_routes):
def test_tags_parsing_in_petstore_routes(parsed_petstore_routes: list[HTTPRoute]):
"""Test that tags are correctly parsed from the OpenAPI schema."""
# All petstore routes should have the "pets" tag
for route in parsed_petstore_routes:
@ -639,7 +646,7 @@ def test_tags_parsing_in_petstore_routes(parsed_petstore_routes):
)
def test_tag_list_structure(parsed_petstore_routes):
def test_tag_list_structure(parsed_petstore_routes: list[HTTPRoute]):
"""Test that tags are stored as a list of strings."""
for route in parsed_petstore_routes:
assert isinstance(route.tags, list), "Tags should be stored as a list"
@ -647,7 +654,7 @@ def test_tag_list_structure(parsed_petstore_routes):
assert isinstance(tag, str), "Each tag should be a string"
def test_empty_tags_handling(bookstore_schema):
def test_empty_tags_handling(bookstore_schema: dict[str, Any]):
"""Test that routes with no tags are handled correctly with empty lists."""
# Modify a route to remove tags
if "tags" in bookstore_schema["paths"]["/books"]["get"]:
@ -657,16 +664,14 @@ def test_empty_tags_handling(bookstore_schema):
routes = parse_openapi_to_http_routes(bookstore_schema)
# Find the GET /books route
get_books = next(
(r for r in routes if r.method == "GET" and r.path == "/books"), None
)
get_books = get_route(routes, "GET", "/books")
assert get_books is not None
# Should have an empty list, not None
assert get_books.tags == [], "Routes without tags should have empty tag lists"
def test_multiple_tags_preserved(bookstore_schema):
def test_multiple_tags_preserved(bookstore_schema: dict[str, Any]):
"""Test that multiple tags are preserved during parsing."""
# Add multiple tags to a route
bookstore_schema["paths"]["/books"]["get"]["tags"] = ["books", "catalog", "api"]
@ -675,9 +680,7 @@ def test_multiple_tags_preserved(bookstore_schema):
routes = parse_openapi_to_http_routes(bookstore_schema)
# Find the GET /books route
get_books = next(
(r for r in routes if r.method == "GET" and r.path == "/books"), None
)
get_books = get_route(routes, "GET", "/books")
assert get_books is not None
# Should have all tags
@ -687,7 +690,7 @@ def test_multiple_tags_preserved(bookstore_schema):
assert len(get_books.tags) == 3
def test_openapi_extensions(petstore_schema):
def test_openapi_extensions(petstore_schema: dict[str, Any]):
"""Test that OpenAPI extensions (x-*) are correctly parsed from operations."""
# Add extensions to a route
petstore_schema["paths"]["/pets"]["get"]["x-rate-limit"] = 100
@ -698,9 +701,7 @@ def test_openapi_extensions(petstore_schema):
routes = parse_openapi_to_http_routes(petstore_schema)
# Find the GET /pets route
get_pets = next(
(r for r in routes if r.method == "GET" and r.path == "/pets"), None
)
get_pets = get_route(routes, "GET", "/pets")
assert get_pets is not None
# Should have extensions
@ -713,26 +714,22 @@ def test_openapi_extensions(petstore_schema):
# --- Tests for BookStore schema --- #
def test_bookstore_route_count(parsed_bookstore_routes):
def test_bookstore_route_count(parsed_bookstore_routes: list[HTTPRoute]):
"""Test that parsing the BookStore schema correctly identifies the number of routes."""
assert len(parsed_bookstore_routes) == 4
def test_bookstore_query_parameter_count(parsed_bookstore_routes):
def test_bookstore_query_parameter_count(parsed_bookstore_routes: list[HTTPRoute]):
"""Test that the correct number of query parameters are parsed."""
list_books = next(
(r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None
)
list_books = get_route(parsed_bookstore_routes, "GET", "/books")
assert list_books is not None
assert len(list_books.parameters) == 3
def test_bookstore_query_parameter_names(parsed_bookstore_routes):
def test_bookstore_query_parameter_names(parsed_bookstore_routes: list[HTTPRoute]):
"""Test that query parameter names are correctly parsed."""
list_books = next(
(r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None
)
list_books = get_route(parsed_bookstore_routes, "GET", "/books")
assert list_books is not None
param_map = {p.name: p for p in list_books.parameters}
@ -741,33 +738,29 @@ def test_bookstore_query_parameter_names(parsed_bookstore_routes):
assert "limit" in param_map
def test_bookstore_query_parameter_formats(parsed_bookstore_routes):
def test_bookstore_query_parameter_formats(parsed_bookstore_routes: list[HTTPRoute]):
"""Test that query parameter formats are correctly parsed."""
list_books = next(
(r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None
)
list_books = get_route(parsed_bookstore_routes, "GET", "/books")
assert list_books is not None
param_map = {p.name: p for p in list_books.parameters}
assert param_map["published_after"].schema_.get("format") == "date"
def test_bookstore_query_parameter_defaults(parsed_bookstore_routes):
def test_bookstore_query_parameter_defaults(parsed_bookstore_routes: list[HTTPRoute]):
"""Test that query parameter default values are correctly parsed."""
list_books = next(
(r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None
)
list_books = get_route(parsed_bookstore_routes, "GET", "/books")
assert list_books is not None
param_map = {p.name: p for p in list_books.parameters}
assert param_map["limit"].schema_.get("default") == 10
def test_bookstore_inline_request_body_presence(parsed_bookstore_routes):
def test_bookstore_inline_request_body_presence(
parsed_bookstore_routes: list[HTTPRoute],
):
"""Test that request bodies with inline schemas are present."""
create_book = next(
(r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None
)
create_book = get_route(parsed_bookstore_routes, "POST", "/books")
assert create_book is not None
assert create_book.request_body is not None
@ -775,31 +768,37 @@ def test_bookstore_inline_request_body_presence(parsed_bookstore_routes):
assert "application/json" in create_book.request_body.content_schema
def test_bookstore_inline_request_body_properties(parsed_bookstore_routes):
def test_bookstore_inline_request_body_properties(
parsed_bookstore_routes: list[HTTPRoute],
):
"""Test that request body properties are correctly parsed from inline schemas."""
create_book = next(
(r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None
)
create_book = get_route(parsed_bookstore_routes, "POST", "/books")
assert create_book is not None
assert create_book.request_body is not None
json_schema = create_book.request_body.content_schema["application/json"]
properties = json_schema.get("properties", {})
assert "title" in properties
assert "author" in properties
assert "isbn" in properties
assert "published" in properties
assert "genre" in properties
def test_bookstore_inline_request_body_required_fields(parsed_bookstore_routes):
"""Test that required fields in inline schema are correctly parsed."""
create_book = next(
(r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None
assert json_schema == snapshot(
{
"properties": {
"title": {"type": "string"},
"author": {"type": "string"},
"isbn": {"type": "string"},
"published": {"type": "string", "format": "date"},
"genre": {"type": "string"},
},
"type": "object",
"required": ["title", "author"],
}
)
def test_bookstore_inline_request_body_required_fields(
parsed_bookstore_routes: list[HTTPRoute],
):
"""Test that required fields in inline schema are correctly parsed."""
create_book = get_route(parsed_bookstore_routes, "POST", "/books")
assert create_book is not None
assert create_book.request_body is not None
@ -807,22 +806,18 @@ def test_bookstore_inline_request_body_required_fields(parsed_bookstore_routes):
assert json_schema.get("required") == ["title", "author"]
def test_bookstore_delete_method(parsed_bookstore_routes):
def test_bookstore_delete_method(parsed_bookstore_routes: list[HTTPRoute]):
"""Test that DELETE method is correctly parsed from the schema."""
delete_book = next(
(r for r in parsed_bookstore_routes if r.method == "DELETE"), None
)
delete_book = get_route(parsed_bookstore_routes, "DELETE", "/books/{isbn}")
assert delete_book is not None
assert delete_book.operation_id == "deleteBook"
assert delete_book.path == "/books/{isbn}"
def test_bookstore_delete_method_parameters(parsed_bookstore_routes):
def test_bookstore_delete_method_parameters(parsed_bookstore_routes: list[HTTPRoute]):
"""Test that parameters for DELETE method are correctly parsed."""
delete_book = next(
(r for r in parsed_bookstore_routes if r.method == "DELETE"), None
)
delete_book = get_route(parsed_bookstore_routes, "DELETE", "/books/{isbn}")
assert delete_book is not None
assert len(delete_book.parameters) == 1
@ -832,12 +827,12 @@ def test_bookstore_delete_method_parameters(parsed_bookstore_routes):
# --- Tests for FastAPI Generated Schema --- #
def test_fastapi_route_count(parsed_fastapi_routes):
def test_fastapi_route_count(parsed_fastapi_routes: list[HTTPRoute]):
"""Test that parsing a FastAPI-generated schema correctly identifies the number of routes."""
assert len(parsed_fastapi_routes) == 7
def test_fastapi_parameter_default_values(fastapi_route_map):
def test_fastapi_parameter_default_values(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that default parameter values are correctly parsed from the schema."""
list_items = fastapi_route_map["list_items"]
@ -846,7 +841,7 @@ def test_fastapi_parameter_default_values(fastapi_route_map):
assert "limit" in param_map
def test_fastapi_skip_parameter_default(fastapi_route_map):
def test_fastapi_skip_parameter_default(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that skip parameter default value is correctly parsed."""
list_items = fastapi_route_map["list_items"]
@ -854,7 +849,7 @@ def test_fastapi_skip_parameter_default(fastapi_route_map):
assert param_map["skip"].schema_.get("default") == 0
def test_fastapi_limit_parameter_default(fastapi_route_map):
def test_fastapi_limit_parameter_default(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that limit parameter default value is correctly parsed."""
list_items = fastapi_route_map["list_items"]
@ -862,7 +857,7 @@ def test_fastapi_limit_parameter_default(fastapi_route_map):
assert param_map["limit"].schema_.get("default") == 10
def test_fastapi_request_body_from_pydantic(fastapi_route_map):
def test_fastapi_request_body_from_pydantic(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that request bodies from Pydantic models are present."""
create_item = fastapi_route_map["create_item"]
@ -870,10 +865,12 @@ def test_fastapi_request_body_from_pydantic(fastapi_route_map):
assert "application/json" in create_item.request_body.content_schema
def test_fastapi_request_body_properties(fastapi_route_map):
def test_fastapi_request_body_properties(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that request body properties from Pydantic models are correctly parsed."""
create_item = fastapi_route_map["create_item"]
assert create_item.request_body is not None
json_schema = create_item.request_body.content_schema["application/json"]
properties = json_schema.get("properties", {})
@ -884,10 +881,12 @@ def test_fastapi_request_body_properties(fastapi_route_map):
assert "tags" in properties
def test_fastapi_request_body_required_fields(fastapi_route_map):
def test_fastapi_request_body_required_fields(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that required fields from Pydantic models are correctly parsed."""
create_item = fastapi_route_map["create_item"]
assert create_item.request_body is not None
json_schema = create_item.request_body.content_schema["application/json"]
required = json_schema.get("required", [])
@ -895,7 +894,7 @@ def test_fastapi_request_body_required_fields(fastapi_route_map):
assert "price" in required
def test_fastapi_path_parameter_presence(fastapi_route_map):
def test_fastapi_path_parameter_presence(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that path parameters are present in FastAPI schema."""
get_item = fastapi_route_map["get_item"]
@ -903,7 +902,7 @@ def test_fastapi_path_parameter_presence(fastapi_route_map):
assert len(path_params) == 1
def test_fastapi_path_parameter_properties(fastapi_route_map):
def test_fastapi_path_parameter_properties(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that path parameters properties are correctly parsed."""
get_item = fastapi_route_map["get_item"]
@ -912,7 +911,7 @@ def test_fastapi_path_parameter_properties(fastapi_route_map):
assert path_params[0].required is True
def test_fastapi_optional_query_parameter(fastapi_route_map):
def test_fastapi_optional_query_parameter(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that optional query parameters are correctly parsed."""
get_item = fastapi_route_map["get_item"]
@ -922,7 +921,7 @@ def test_fastapi_optional_query_parameter(fastapi_route_map):
assert query_params[0].required is False
def test_fastapi_multiple_path_parameter_count(fastapi_route_map):
def test_fastapi_multiple_path_parameter_count(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that multiple path parameters count is correct."""
get_item_tag = fastapi_route_map["get_item_tag"]
@ -930,7 +929,7 @@ def test_fastapi_multiple_path_parameter_count(fastapi_route_map):
assert len(path_params) == 2
def test_fastapi_multiple_path_parameter_names(fastapi_route_map):
def test_fastapi_multiple_path_parameter_names(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that multiple path parameter names are correctly parsed."""
get_item_tag = fastapi_route_map["get_item_tag"]
@ -940,16 +939,41 @@ def test_fastapi_multiple_path_parameter_names(fastapi_route_map):
assert "tag_id" in param_names
def test_fastapi_post_with_query_parameters(fastapi_route_map):
def test_fastapi_post_with_query_parameters(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that query parameters for POST methods are correctly parsed."""
upload_file = fastapi_route_map["upload_file"]
assert upload_file.method == "POST"
query_params = [p for p in upload_file.parameters if p.location == "query"]
assert len(query_params) == 2
assert dump_models(query_params, exclude_none=True) == snapshot(
[
{
"name": "file_name",
"location": "query",
"required": True,
"schema_": {
"type": "string",
"title": "File Name",
"description": "Name of the file to upload",
},
"description": "Name of the file to upload",
},
{
"name": "content_type",
"location": "query",
"required": True,
"schema_": {
"type": "string",
"title": "Content Type",
"description": "Content type of the file",
},
"description": "Content type of the file",
},
]
)
def test_fastapi_post_query_parameter_names(fastapi_route_map):
def test_fastapi_post_query_parameter_names(fastapi_route_map: dict[str, HTTPRoute]):
"""Test that query parameter names for POST methods are correctly parsed."""
upload_file = fastapi_route_map["upload_file"]
@ -959,7 +983,7 @@ def test_fastapi_post_query_parameter_names(fastapi_route_map):
assert "content_type" in param_names
def test_openapi_30_compatibility(openapi_30_schema):
def test_openapi_30_compatibility(openapi_30_schema: dict[str, Any]):
"""Test that OpenAPI 3.0 schemas can be parsed correctly."""
# This will raise an exception if the parser doesn't support 3.0.0
routes = parse_openapi_to_http_routes(openapi_30_schema)
@ -974,7 +998,7 @@ def test_openapi_30_compatibility(openapi_30_schema):
assert route.parameters[0].name == "limit"
def test_openapi_31_compatibility(openapi_31_schema):
def test_openapi_31_compatibility(openapi_31_schema: dict[str, Any]):
"""Test that OpenAPI 3.1 schemas can be parsed correctly."""
routes = parse_openapi_to_http_routes(openapi_31_schema)
@ -1017,7 +1041,7 @@ def test_version_detection_logic():
pytest.fail(f"Failed to parse OpenAPI {version} schema: {e}")
def test_openapi_30_reference_resolution(openapi_30_with_references):
def test_openapi_30_reference_resolution(openapi_30_with_references: dict[str, Any]):
"""Test that references are correctly resolved in OpenAPI 3.0 schemas."""
routes = parse_openapi_to_http_routes(openapi_30_with_references)
@ -1031,30 +1055,46 @@ def test_openapi_30_reference_resolution(openapi_30_with_references):
assert route.request_body.required is True
assert "application/json" in route.request_body.content_schema
# Check schema structure
# Check schema structure with snapshots
json_schema = route.request_body.content_schema["application/json"]
assert json_schema["type"] == "object"
assert "properties" in json_schema
assert set(json_schema["required"]) == {"name", "price"}
# Check primary fields are properly resolved
props = json_schema["properties"]
assert "id" in props
assert "name" in props
assert "price" in props
assert "category" in props
# The category might be a reference or resolved object
category = props["category"]
# Either it's directly resolved with properties
# or it still has a $ref field
assert "properties" in category or "$ref" in category
assert json_schema == snapshot(
{
"required": ["name", "price"],
"type": "object",
"properties": {
"id": {"type": "string", "format": "uuid"},
"name": {"type": "string"},
"price": {"type": "number"},
"category": {"$ref": "#/$defs/Category"},
},
}
)
combined_schema = _combine_schemas(route)
assert "#/$defs/" in combined_schema["properties"]["category"]["$ref"]
assert combined_schema == snapshot(
{
"type": "object",
"properties": {
"id": {"type": "string", "format": "uuid"},
"name": {"type": "string"},
"price": {"type": "number"},
"category": {"$ref": "#/$defs/Category"},
},
"required": ["name", "price"],
"$defs": {
"Category": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
},
}
},
}
)
def test_openapi_31_reference_resolution(openapi_31_with_references):
def test_openapi_31_reference_resolution(openapi_31_with_references: dict[str, Any]):
"""Test that references are correctly resolved in OpenAPI 3.1 schemas."""
routes = parse_openapi_to_http_routes(openapi_31_with_references)
@ -1070,29 +1110,46 @@ def test_openapi_31_reference_resolution(openapi_31_with_references):
# Check schema structure
json_schema = route.request_body.content_schema["application/json"]
assert json_schema["type"] == "object"
assert "properties" in json_schema
assert set(json_schema["required"]) == {"name", "price"}
# Check primary fields are properly resolved
props = json_schema["properties"]
assert "id" in props
assert "name" in props
assert "price" in props
assert "category" in props
# The category might be a reference or resolved object
category = props["category"]
# Either it's directly resolved with properties
# or it still has a $ref field
assert "properties" in category or "$ref" in category
assert json_schema == snapshot(
{
"properties": {
"id": {"type": "string", "format": "uuid"},
"name": {"type": "string"},
"price": {"type": "number"},
"category": {"$ref": "#/$defs/Category"},
},
"type": "object",
"required": ["name", "price"],
}
)
combined_schema = _combine_schemas(route)
assert "#/$defs/" in combined_schema["properties"]["category"]["$ref"]
assert combined_schema == snapshot(
{
"type": "object",
"properties": {
"id": {"type": "string", "format": "uuid"},
"name": {"type": "string"},
"price": {"type": "number"},
"category": {"$ref": "#/$defs/Category"},
},
"required": ["name", "price"],
"$defs": {
"Category": {
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
},
"type": "object",
}
},
}
)
def test_consistent_output_across_versions(
openapi_30_with_references, openapi_31_with_references
openapi_30_with_references: dict[str, Any],
openapi_31_with_references: dict[str, Any],
):
"""Test that both parsers produce equivalent output for equivalent schemas."""
routes_30 = parse_openapi_to_http_routes(openapi_30_with_references)

1
tests/utilities/test_json_schema.py
View file

@ -74,6 +74,7 @@ class TestPruneUnusedDefs:
prune_additional_properties=False,
prune_titles=False,
)
assert "foo_def" in result["$defs"]
assert "unused_def" not in result["$defs"]

32
uv.lock generated
View file

@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 2
requires-python = ">=3.10"
resolution-markers = [
"python_full_version >= '3.11'",
@ -231,10 +231,9 @@ wheels = [
[[package]]
name = "copychat"
version = "0.7.2"
version = "0.6.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "fastmcp" },
{ name = "gitpython" },
{ name = "pathspec" },
{ name = "pyperclip" },
@ -242,9 +241,9 @@ dependencies = [
{ name = "tiktoken" },
{ name = "typer" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d4/77/a72f890207b33eb542e9507a9d167e8ff734080a9d265472d7a774bd46e4/copychat-0.7.2.tar.gz", hash = "sha256:3f8c21039f0f8874fb84d2163e467e2e003ac625d218225800732244c55176fa", size = 95779, upload-time = "2025-06-19T18:20:27.501Z" }
sdist = { url = "https://files.pythonhosted.org/packages/06/d9/112fd77fdc21e89dee79583d326edca3597493be5666281b87e393de2cf9/copychat-0.6.3.tar.gz", hash = "sha256:39ffb493506f20e72d26673490d5a7228cf40f3712d6a60ad6a9ac9f7106f5e4", size = 78328, upload-time = "2025-06-03T15:53:13.368Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fb/b7/266a72b4e843c61bffe2082539c7b634bbe42dd47d8110a5c15e3ee8d66a/copychat-0.7.2-py3-none-any.whl", hash = "sha256:ac2dcb86b70abeb5f8483fc6c70695c93c60b4e851a5b57b165edd36f3e15e8c", size = 23920, upload-time = "2025-06-19T18:20:26.405Z" },
{ url = "https://files.pythonhosted.org/packages/05/0b/e2f61f7bba857850b5022ce609ba9fcff8308458f1608ec20b35132263b9/copychat-0.6.3-py3-none-any.whl", hash = "sha256:1460cd02c09b6495550f6a4aa2ab0bacbf2b95176e876fc268b35d22243a4d97", size = 21617, upload-time = "2025-06-03T15:53:11.378Z" },
]
[[package]]
@ -561,6 +560,7 @@ dev = [
{ name = "copychat" },
{ name = "dirty-equals" },
{ name = "fastapi" },
{ name = "inline-snapshot", extra = ["dirty-equals"] },
{ name = "ipython", version = "8.37.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
{ name = "ipython", version = "9.4.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
{ name = "pdbpp" },
@ -604,6 +604,7 @@ dev = [
{ name = "copychat", specifier = ">=0.5.2" },
{ name = "dirty-equals", specifier = ">=0.9.0" },
{ name = "fastapi", specifier = ">=0.115.12" },
{ name = "inline-snapshot", extras = ["dirty-equals"], specifier = ">=0.27.2" },
{ name = "ipython", specifier = ">=8.12.3" },
{ name = "pdbpp", specifier = ">=0.10.3" },
{ name = "pre-commit" },
@ -729,6 +730,27 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/2c/e1/e6716421ea10d38022b952c159d5161ca1193197fb744506875fbb87ea7b/iniconfig-2.1.0-py3-none-any.whl", hash = "sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760", size = 6050, upload-time = "2025-03-19T20:10:01.071Z" },
]
[[package]]
name = "inline-snapshot"
version = "0.28.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "asttokens" },
{ name = "executing" },
{ name = "pytest" },
{ name = "rich" },
{ name = "tomli", marker = "python_full_version < '3.11'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e4/9e/83aaa750e9c8115d34b2d80646c1988941f2252c5548caf35aad5e529bad/inline_snapshot-0.28.0.tar.gz", hash = "sha256:6904bfc383240b6bea64de2f5d2992f04109b13def19395bdd13fb0ebcf5cf20", size = 348554, upload-time = "2025-08-24T21:48:04.056Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e6/04/190b336a006d4e1275c2dde1bf953336e818d18b779f24947579bb4ba48d/inline_snapshot-0.28.0-py3-none-any.whl", hash = "sha256:9988f82ee5e719445bbc437d0dc01e0a3c4c94f0ba910f8ad8b573cf15aa8348", size = 69026, upload-time = "2025-08-24T21:48:02.342Z" },
]
[package.optional-dependencies]
dirty-equals = [
{ name = "dirty-equals" },
]
[[package]]
name = "ipython"
version = "8.37.0"