diff --git a/pyproject.toml b/pyproject.toml index 20616529f..c5c8cfa6b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", diff --git a/tests/conftest.py b/tests/conftest.py index 4fba247c6..d82413029 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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 diff --git a/tests/tools/test_tool.py b/tests/tools/test_tool.py index a4583a060..7f8f2077e 100644 --- a/tests/tools/test_tool.py +++ b/tests/tools/test_tool.py @@ -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 diff --git a/tests/utilities/openapi/test_openapi.py b/tests/utilities/openapi/test_openapi.py index 6cbad7cad..f9d613281 100644 --- a/tests/utilities/openapi/test_openapi.py +++ b/tests/utilities/openapi/test_openapi.py @@ -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) diff --git a/tests/utilities/test_json_schema.py b/tests/utilities/test_json_schema.py index 4d5d8a1e9..65ba9b2e3 100644 --- a/tests/utilities/test_json_schema.py +++ b/tests/utilities/test_json_schema.py @@ -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"] diff --git a/uv.lock b/uv.lock index 446e8e09a..65c81ac8f 100644 --- a/uv.lock +++ b/uv.lock @@ -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"