prefecthq--fastmcp/tests/utilities/openapi/test_direct_array_schemas.py

"""Test handling of direct array schemas in request bodies (FastAPI list parameters)."""

from fastmcp.utilities.openapi.models import (
    HTTPRoute,
    ParameterInfo,
    RequestBodyInfo,
)
from fastmcp.utilities.openapi.schemas import (
    _combine_schemas_and_map_params,
)


class TestDirectArraySchemas:
    """Test handling of direct array schemas like those generated by FastAPI for list[str] parameters."""

    def test_simple_direct_array_schema(self):
        """Test route with direct array request body schema."""
        route = HTTPRoute(
            path="/simple_list",
            method="POST",
            operation_id="simple_list_tool",
            summary="Simple List Tool",
            description="A simple tool that takes a list of strings.",
            parameters=[],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {
                        "items": {"type": "string"},
                        "type": "array",
                        "title": "Values",
                    }
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should create a single parameter from the title
        assert combined_schema["type"] == "object"
        assert "values" in combined_schema["properties"]
        assert "values" in combined_schema["required"]

        # Check the array schema is preserved
        values_schema = combined_schema["properties"]["values"]
        assert values_schema["type"] == "array"
        assert values_schema["items"]["type"] == "string"
        assert values_schema["title"] == "Values"

        # Check parameter mapping
        assert "values" in param_map
        assert param_map["values"]["location"] == "body"
        assert param_map["values"]["openapi_name"] == "values"

    def test_int_array_schema(self):
        """Test route with integer array request body schema."""
        route = HTTPRoute(
            path="/int_list",
            method="POST",
            operation_id="int_list_tool",
            summary="Integer List Tool",
            description="A simple tool that takes a list of integers.",
            parameters=[],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {
                        "items": {"type": "integer"},
                        "type": "array",
                        "title": "Numbers",
                    }
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should create a single parameter from the title
        assert combined_schema["type"] == "object"
        assert "numbers" in combined_schema["properties"]
        assert "numbers" in combined_schema["required"]

        # Check the array schema is preserved
        numbers_schema = combined_schema["properties"]["numbers"]
        assert numbers_schema["type"] == "array"
        assert numbers_schema["items"]["type"] == "integer"
        assert numbers_schema["title"] == "Numbers"

    def test_mixed_params_with_direct_array(self):
        """Test route with both URL parameters and direct array request body."""
        route = HTTPRoute(
            path="/mixed_params",
            method="POST",
            operation_id="mixed_params_tool",
            summary="Mixed Params Tool",
            description="A tool with both list and simple parameters.",
            parameters=[
                ParameterInfo(
                    name="prefix",
                    location="query",
                    required=True,
                    schema={"type": "string", "title": "Prefix"},
                )
            ],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {
                        "type": "array",
                        "items": {"type": "string"},
                        "title": "Values",
                    }
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should have both parameters
        assert combined_schema["type"] == "object"
        assert "prefix" in combined_schema["properties"]  # From query param
        assert "values" in combined_schema["properties"]  # From body array
        assert "prefix" in combined_schema["required"]
        assert "values" in combined_schema["required"]

        # Check query parameter
        prefix_schema = combined_schema["properties"]["prefix"]
        assert prefix_schema["type"] == "string"

        # Check body array parameter
        values_schema = combined_schema["properties"]["values"]
        assert values_schema["type"] == "array"
        assert values_schema["items"]["type"] == "string"

        # Check parameter mappings
        assert param_map["prefix"]["location"] == "query"
        assert param_map["values"]["location"] == "body"

    def test_direct_array_no_title(self):
        """Test direct array schema without title."""
        route = HTTPRoute(
            path="/no_title",
            method="POST",
            operation_id="no_title_tool",
            parameters=[],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {
                        "type": "array",
                        "items": {"type": "string"},
                        # No title
                    }
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should fall back to "body" (default title falls back)
        assert combined_schema["type"] == "object"
        assert "body" in combined_schema["properties"]
        assert "body" in combined_schema["required"]

        # Check parameter mapping
        assert param_map["body"]["location"] == "body"
        assert param_map["body"]["openapi_name"] == "body"

    def test_direct_primitive_schema(self):
        """Test direct primitive (non-array) request body schema."""
        route = HTTPRoute(
            path="/primitive",
            method="POST",
            operation_id="primitive_tool",
            parameters=[],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {"type": "string", "title": "Message"}
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should create a single parameter from the title
        assert combined_schema["type"] == "object"
        assert "message" in combined_schema["properties"]
        assert "message" in combined_schema["required"]

        # Check the primitive schema is preserved
        message_schema = combined_schema["properties"]["message"]
        assert message_schema["type"] == "string"
        assert message_schema["title"] == "Message"

    def test_direct_array_optional(self):
        """Test direct array schema that's not required."""
        route = HTTPRoute(
            path="/optional_list",
            method="POST",
            operation_id="optional_list_tool",
            parameters=[],
            request_body=RequestBodyInfo(
                required=False,  # Not required
                content_schema={
                    "application/json": {
                        "type": "array",
                        "items": {"type": "string"},
                        "title": "OptionalValues",
                    }
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should create parameter but not mark as required
        assert combined_schema["type"] == "object"
        assert "optionalvalues" in combined_schema["properties"]
        assert "optionalvalues" not in combined_schema["required"]

    def test_direct_array_title_sanitization(self):
        """Test that titles with special characters are sanitized."""
        route = HTTPRoute(
            path="/special_chars",
            method="POST",
            operation_id="special_chars_tool",
            parameters=[],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {
                        "type": "array",
                        "items": {"type": "string"},
                        "title": "Special-Chars & Spaces!",
                    }
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should sanitize the title to a valid parameter name (& becomes multiple underscores)
        assert combined_schema["type"] == "object"
        assert "special_chars___spaces_" in combined_schema["properties"]

    def test_direct_array_title_starting_with_number(self):
        """Test that titles starting with numbers are handled."""
        route = HTTPRoute(
            path="/number_start",
            method="POST",
            operation_id="number_start_tool",
            parameters=[],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {
                        "type": "array",
                        "items": {"type": "string"},
                        "title": "123Values",
                    }
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should fall back to "body_data" since "123values" starts with a number
        assert combined_schema["type"] == "object"
        assert "body_data" in combined_schema["properties"]

    def test_preserve_existing_behavior_object_body(self):
        """Test that existing object-based request bodies still work."""
        route = HTTPRoute(
            path="/object_body",
            method="POST",
            operation_id="object_body_tool",
            parameters=[],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "age": {"type": "integer"},
                        },
                        "required": ["name"],
                    }
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should handle object bodies as before
        assert combined_schema["type"] == "object"
        assert "name" in combined_schema["properties"]
        assert "age" in combined_schema["properties"]
        assert "name" in combined_schema["required"]
        assert "age" not in combined_schema["required"]

    def test_preserve_existing_behavior_ref_body(self):
        """Test that $ref-based request bodies still work."""
        route = HTTPRoute(
            path="/ref_body",
            method="POST",
            operation_id="ref_body_tool",
            parameters=[],
            request_body=RequestBodyInfo(
                required=True,
                content_schema={
                    "application/json": {"$ref": "#/components/schemas/UserCreate"}
                },
            ),
        )

        combined_schema, param_map = _combine_schemas_and_map_params(route)

        # Should handle $ref bodies as before (refs get converted to $defs)
        assert combined_schema["type"] == "object"
        assert "body" in combined_schema["properties"]
        assert combined_schema["properties"]["body"]["$ref"] == "#/$defs/UserCreate"