mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Fix nullable field handling in OpenAPI to JSON Schema conversion (#1279)
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Jeremiah Lowin
GitHub
parent
7709ef0c8a
commit
52a0fc078a
4 changed files with 688 additions and 6 deletions
|
|
@ -199,6 +199,86 @@ def _make_optional_parameter_nullable(schema: dict[str, Any]) -> dict[str, Any]:
|
|||
return schema
|
||||
|
||||
|
||||
def _add_null_to_type(schema: dict[str, Any]) -> None:
|
||||
"""Add 'null' to the schema's type field or handle oneOf/anyOf/allOf constructs if not already present."""
|
||||
if "type" in schema:
|
||||
current_type = schema["type"]
|
||||
|
||||
if isinstance(current_type, str):
|
||||
# Convert string type to array with null
|
||||
schema["type"] = [current_type, "null"]
|
||||
elif isinstance(current_type, list):
|
||||
# Add null to array if not already present
|
||||
if "null" not in current_type:
|
||||
schema["type"] = current_type + ["null"]
|
||||
elif "oneOf" in schema:
|
||||
# Convert oneOf to anyOf with null type
|
||||
schema["anyOf"] = schema.pop("oneOf") + [{"type": "null"}]
|
||||
elif "anyOf" in schema:
|
||||
# Add null type to anyOf if not already present
|
||||
if not any(item.get("type") == "null" for item in schema["anyOf"]):
|
||||
schema["anyOf"].append({"type": "null"})
|
||||
elif "allOf" in schema:
|
||||
# For allOf, wrap in anyOf with null - this means (all conditions) OR null
|
||||
schema["anyOf"] = [{"allOf": schema.pop("allOf")}, {"type": "null"}]
|
||||
|
||||
|
||||
def _handle_nullable_fields(schema: dict[str, Any] | Any) -> dict[str, Any] | Any:
|
||||
"""Convert OpenAPI nullable fields to JSON Schema format: {"type": "string",
|
||||
"nullable": true} -> {"type": ["string", "null"]}"""
|
||||
|
||||
if not isinstance(schema, dict):
|
||||
return schema
|
||||
|
||||
# Check if we need to modify anything first to avoid unnecessary copying
|
||||
has_root_nullable_field = "nullable" in schema
|
||||
has_root_nullable_true = (
|
||||
has_root_nullable_field
|
||||
and schema["nullable"]
|
||||
and (
|
||||
"type" in schema
|
||||
or "oneOf" in schema
|
||||
or "anyOf" in schema
|
||||
or "allOf" in schema
|
||||
)
|
||||
)
|
||||
|
||||
has_property_nullable_field = False
|
||||
if "properties" in schema:
|
||||
for prop_schema in schema["properties"].values():
|
||||
if isinstance(prop_schema, dict) and "nullable" in prop_schema:
|
||||
has_property_nullable_field = True
|
||||
break
|
||||
|
||||
# If no nullable fields at all, return original schema unchanged
|
||||
if not has_root_nullable_field and not has_property_nullable_field:
|
||||
return schema
|
||||
|
||||
# Only copy if we need to modify
|
||||
result = schema.copy()
|
||||
|
||||
# Handle root level nullable - always remove the field, convert type if true
|
||||
if has_root_nullable_field:
|
||||
result.pop("nullable")
|
||||
if has_root_nullable_true:
|
||||
_add_null_to_type(result)
|
||||
|
||||
# Handle properties nullable fields
|
||||
if has_property_nullable_field and "properties" in result:
|
||||
for prop_name, prop_schema in result["properties"].items():
|
||||
if isinstance(prop_schema, dict) and "nullable" in prop_schema:
|
||||
nullable_value = prop_schema.pop("nullable")
|
||||
if nullable_value and (
|
||||
"type" in prop_schema
|
||||
or "oneOf" in prop_schema
|
||||
or "anyOf" in prop_schema
|
||||
or "allOf" in prop_schema
|
||||
):
|
||||
_add_null_to_type(prop_schema)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _combine_schemas_and_map_params(
|
||||
route: HTTPRoute,
|
||||
) -> tuple[dict[str, Any], dict[str, dict[str, str]]]:
|
||||
|
|
@ -499,6 +579,19 @@ def extract_output_schema_from_responses(
|
|||
# Clean and copy the schema
|
||||
output_schema = schema.copy()
|
||||
|
||||
# If schema has a $ref, resolve it first before processing nullable fields
|
||||
if "$ref" in output_schema and schema_definitions:
|
||||
ref_path = output_schema["$ref"]
|
||||
if ref_path.startswith("#/components/schemas/"):
|
||||
schema_name = ref_path.split("/")[-1]
|
||||
if schema_name in schema_definitions:
|
||||
# Replace $ref with the actual schema definition
|
||||
output_schema = schema_definitions[schema_name].copy()
|
||||
|
||||
# Handle OpenAPI nullable fields by converting them to JSON Schema format
|
||||
# This prevents "None is not of type 'string'" validation errors
|
||||
output_schema = _handle_nullable_fields(output_schema)
|
||||
|
||||
# MCP requires output schemas to be objects. If this schema is not an object,
|
||||
# we need to wrap it similar to how ParsedFunction.from_function() does it
|
||||
if output_schema.get("type") != "object":
|
||||
|
|
@ -511,9 +604,13 @@ def extract_output_schema_from_responses(
|
|||
}
|
||||
output_schema = wrapped_schema
|
||||
|
||||
# Add schema definitions if available
|
||||
if schema_definitions:
|
||||
output_schema["$defs"] = schema_definitions.copy()
|
||||
# Add schema definitions if available and handle nullable fields in them
|
||||
# Only add $defs if we didn't resolve the $ref inline above
|
||||
if schema_definitions and "$ref" not in schema.copy():
|
||||
processed_defs = {}
|
||||
for def_name, def_schema in schema_definitions.items():
|
||||
processed_defs[def_name] = _handle_nullable_fields(def_schema)
|
||||
output_schema["$defs"] = processed_defs
|
||||
|
||||
# Use lightweight compression - prune additionalProperties and unused definitions
|
||||
if output_schema.get("additionalProperties") is False:
|
||||
|
|
@ -564,5 +661,6 @@ __all__ = [
|
|||
"extract_output_schema_from_responses",
|
||||
"_replace_ref_with_defs",
|
||||
"_make_optional_parameter_nullable",
|
||||
"_handle_nullable_fields",
|
||||
"_adjust_union_types",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -187,6 +187,7 @@ __all__ = [
|
|||
"parse_openapi_to_http_routes",
|
||||
"extract_output_schema_from_responses",
|
||||
"format_deep_object_parameter",
|
||||
"_handle_nullable_fields",
|
||||
]
|
||||
|
||||
# Type variables for generic parser
|
||||
|
|
@ -1169,6 +1170,86 @@ def _make_optional_parameter_nullable(schema: dict[str, Any]) -> dict[str, Any]:
|
|||
return schema
|
||||
|
||||
|
||||
def _add_null_to_type(schema: dict[str, Any]) -> None:
|
||||
"""Add 'null' to the schema's type field or handle oneOf/anyOf/allOf constructs if not already present."""
|
||||
if "type" in schema:
|
||||
current_type = schema["type"]
|
||||
|
||||
if isinstance(current_type, str):
|
||||
# Convert string type to array with null
|
||||
schema["type"] = [current_type, "null"]
|
||||
elif isinstance(current_type, list):
|
||||
# Add null to array if not already present
|
||||
if "null" not in current_type:
|
||||
schema["type"] = current_type + ["null"]
|
||||
elif "oneOf" in schema:
|
||||
# Convert oneOf to anyOf with null type
|
||||
schema["anyOf"] = schema.pop("oneOf") + [{"type": "null"}]
|
||||
elif "anyOf" in schema:
|
||||
# Add null type to anyOf if not already present
|
||||
if not any(item.get("type") == "null" for item in schema["anyOf"]):
|
||||
schema["anyOf"].append({"type": "null"})
|
||||
elif "allOf" in schema:
|
||||
# For allOf, wrap in anyOf with null - this means (all conditions) OR null
|
||||
schema["anyOf"] = [{"allOf": schema.pop("allOf")}, {"type": "null"}]
|
||||
|
||||
|
||||
def _handle_nullable_fields(schema: dict[str, Any] | Any) -> dict[str, Any] | Any:
|
||||
"""Convert OpenAPI nullable fields to JSON Schema format: {"type": "string",
|
||||
"nullable": true} -> {"type": ["string", "null"]}"""
|
||||
|
||||
if not isinstance(schema, dict):
|
||||
return schema
|
||||
|
||||
# Check if we need to modify anything first to avoid unnecessary copying
|
||||
has_root_nullable_field = "nullable" in schema
|
||||
has_root_nullable_true = (
|
||||
has_root_nullable_field
|
||||
and schema["nullable"]
|
||||
and (
|
||||
"type" in schema
|
||||
or "oneOf" in schema
|
||||
or "anyOf" in schema
|
||||
or "allOf" in schema
|
||||
)
|
||||
)
|
||||
|
||||
has_property_nullable_field = False
|
||||
if "properties" in schema:
|
||||
for prop_schema in schema["properties"].values():
|
||||
if isinstance(prop_schema, dict) and "nullable" in prop_schema:
|
||||
has_property_nullable_field = True
|
||||
break
|
||||
|
||||
# If no nullable fields at all, return original schema unchanged
|
||||
if not has_root_nullable_field and not has_property_nullable_field:
|
||||
return schema
|
||||
|
||||
# Only copy if we need to modify
|
||||
result = schema.copy()
|
||||
|
||||
# Handle root level nullable - always remove the field, convert type if true
|
||||
if has_root_nullable_field:
|
||||
result.pop("nullable")
|
||||
if has_root_nullable_true:
|
||||
_add_null_to_type(result)
|
||||
|
||||
# Handle properties nullable fields
|
||||
if has_property_nullable_field and "properties" in result:
|
||||
for prop_name, prop_schema in result["properties"].items():
|
||||
if isinstance(prop_schema, dict) and "nullable" in prop_schema:
|
||||
nullable_value = prop_schema.pop("nullable")
|
||||
if nullable_value and (
|
||||
"type" in prop_schema
|
||||
or "oneOf" in prop_schema
|
||||
or "anyOf" in prop_schema
|
||||
or "allOf" in prop_schema
|
||||
):
|
||||
_add_null_to_type(prop_schema)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _combine_schemas(route: HTTPRoute) -> dict[str, Any]:
|
||||
"""
|
||||
Combines parameter and request body schemas into a single schema.
|
||||
|
|
@ -1388,6 +1469,19 @@ def extract_output_schema_from_responses(
|
|||
# Clean and copy the schema
|
||||
output_schema = schema.copy()
|
||||
|
||||
# If schema has a $ref, resolve it first before processing nullable fields
|
||||
if "$ref" in output_schema and schema_definitions:
|
||||
ref_path = output_schema["$ref"]
|
||||
if ref_path.startswith("#/components/schemas/"):
|
||||
schema_name = ref_path.split("/")[-1]
|
||||
if schema_name in schema_definitions:
|
||||
# Replace $ref with the actual schema definition
|
||||
output_schema = schema_definitions[schema_name].copy()
|
||||
|
||||
# Handle OpenAPI nullable fields by converting them to JSON Schema format
|
||||
# This prevents "None is not of type 'string'" validation errors
|
||||
output_schema = _handle_nullable_fields(output_schema)
|
||||
|
||||
# MCP requires output schemas to be objects. If this schema is not an object,
|
||||
# we need to wrap it similar to how ParsedFunction.from_function() does it
|
||||
if output_schema.get("type") != "object":
|
||||
|
|
@ -1400,9 +1494,13 @@ def extract_output_schema_from_responses(
|
|||
}
|
||||
output_schema = wrapped_schema
|
||||
|
||||
# Add schema definitions if available
|
||||
if schema_definitions:
|
||||
output_schema["$defs"] = schema_definitions.copy()
|
||||
# Add schema definitions if available and handle nullable fields in them
|
||||
# Only add $defs if we didn't resolve the $ref inline above
|
||||
if schema_definitions and "$ref" not in schema.copy():
|
||||
processed_defs = {}
|
||||
for def_name, def_schema in schema_definitions.items():
|
||||
processed_defs[def_name] = _handle_nullable_fields(def_schema)
|
||||
output_schema["$defs"] = processed_defs
|
||||
|
||||
# Use lightweight compression - prune additionalProperties and unused definitions
|
||||
if output_schema.get("additionalProperties") is False:
|
||||
|
|
|
|||
243
tests/experimental/utilities/openapi/test_nullable_fields.py
Normal file
243
tests/experimental/utilities/openapi/test_nullable_fields.py
Normal file
|
|
@ -0,0 +1,243 @@
|
|||
"""Tests for nullable field handling in OpenAPI schemas."""
|
||||
|
||||
from fastmcp.experimental.utilities.openapi.schemas import _handle_nullable_fields
|
||||
|
||||
|
||||
class TestHandleNullableFields:
|
||||
"""Test conversion of OpenAPI nullable fields to JSON Schema format."""
|
||||
|
||||
def test_root_level_nullable_string(self):
|
||||
"""Test nullable string at root level."""
|
||||
input_schema = {"type": "string", "nullable": True}
|
||||
expected = {"type": ["string", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_root_level_nullable_integer(self):
|
||||
"""Test nullable integer at root level."""
|
||||
input_schema = {"type": "integer", "nullable": True}
|
||||
expected = {"type": ["integer", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_root_level_nullable_boolean(self):
|
||||
"""Test nullable boolean at root level."""
|
||||
input_schema = {"type": "boolean", "nullable": True}
|
||||
expected = {"type": ["boolean", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_property_level_nullable_fields(self):
|
||||
"""Test nullable fields in properties."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"company": {"type": "string", "nullable": True},
|
||||
"age": {"type": "integer", "nullable": True},
|
||||
"active": {"type": "boolean", "nullable": True},
|
||||
},
|
||||
}
|
||||
expected = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"company": {"type": ["string", "null"]},
|
||||
"age": {"type": ["integer", "null"]},
|
||||
"active": {"type": ["boolean", "null"]},
|
||||
},
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_mixed_nullable_and_non_nullable(self):
|
||||
"""Test mix of nullable and non-nullable fields."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"required_field": {"type": "string"},
|
||||
"optional_nullable": {"type": "string", "nullable": True},
|
||||
"optional_non_nullable": {"type": "string"},
|
||||
},
|
||||
"required": ["required_field"],
|
||||
}
|
||||
expected = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"required_field": {"type": "string"},
|
||||
"optional_nullable": {"type": ["string", "null"]},
|
||||
"optional_non_nullable": {"type": "string"},
|
||||
},
|
||||
"required": ["required_field"],
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_nullable_false_ignored(self):
|
||||
"""Test that nullable: false is ignored (removed but no type change)."""
|
||||
input_schema = {"type": "string", "nullable": False}
|
||||
expected = {"type": "string"}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_no_nullable_field_unchanged(self):
|
||||
"""Test that schemas without nullable field are unchanged."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
}
|
||||
expected = input_schema.copy()
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_nullable_without_type_removes_nullable(self):
|
||||
"""Test that nullable field is removed even without type."""
|
||||
input_schema = {"nullable": True, "description": "Some field"}
|
||||
expected = {"description": "Some field"}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_preserves_other_fields(self):
|
||||
"""Test that other fields are preserved during conversion."""
|
||||
input_schema = {
|
||||
"type": "string",
|
||||
"nullable": True,
|
||||
"description": "A nullable string",
|
||||
"example": "test",
|
||||
"format": "email",
|
||||
}
|
||||
expected = {
|
||||
"type": ["string", "null"],
|
||||
"description": "A nullable string",
|
||||
"example": "test",
|
||||
"format": "email",
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_non_dict_input_unchanged(self):
|
||||
"""Test that non-dict inputs are returned unchanged."""
|
||||
assert _handle_nullable_fields("string") == "string" # type: ignore[arg-type]
|
||||
assert _handle_nullable_fields(123) == 123 # type: ignore[arg-type]
|
||||
assert _handle_nullable_fields(None) is None # type: ignore[arg-type]
|
||||
assert _handle_nullable_fields([1, 2, 3]) == [1, 2, 3] # type: ignore[arg-type]
|
||||
|
||||
def test_performance_optimization_no_copy_when_unchanged(self):
|
||||
"""Test that schemas without nullable fields return the same object (no copy)."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
# Should return the exact same object, not a copy
|
||||
assert result is input_schema
|
||||
|
||||
def test_union_types_with_nullable(self):
|
||||
"""Test nullable handling with existing union types (type as array)."""
|
||||
input_schema = {"type": ["string", "integer"], "nullable": True}
|
||||
expected = {"type": ["string", "integer", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_already_nullable_union_unchanged(self):
|
||||
"""Test that union types already containing null are not modified."""
|
||||
input_schema = {"type": ["string", "null"], "nullable": True}
|
||||
expected = {"type": ["string", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_property_level_union_with_nullable(self):
|
||||
"""Test nullable handling with union types in properties."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {"value": {"type": ["string", "integer"], "nullable": True}},
|
||||
}
|
||||
expected = {
|
||||
"type": "object",
|
||||
"properties": {"value": {"type": ["string", "integer", "null"]}},
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_complex_union_nullable_scenarios(self):
|
||||
"""Test various complex union type scenarios."""
|
||||
# Already has null in different position
|
||||
input1 = {"type": ["null", "string", "integer"], "nullable": True}
|
||||
result1 = _handle_nullable_fields(input1)
|
||||
assert result1 == {"type": ["null", "string", "integer"]}
|
||||
|
||||
# Single item array
|
||||
input2 = {"type": ["string"], "nullable": True}
|
||||
result2 = _handle_nullable_fields(input2)
|
||||
assert result2 == {"type": ["string", "null"]}
|
||||
|
||||
def test_oneof_with_nullable(self):
|
||||
"""Test nullable handling with oneOf constructs."""
|
||||
input_schema = {
|
||||
"oneOf": [{"type": "string"}, {"type": "integer"}],
|
||||
"nullable": True,
|
||||
}
|
||||
expected = {
|
||||
"anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "null"}]
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_anyof_with_nullable(self):
|
||||
"""Test nullable handling with anyOf constructs."""
|
||||
input_schema = {
|
||||
"anyOf": [{"type": "string"}, {"type": "integer"}],
|
||||
"nullable": True,
|
||||
}
|
||||
expected = {
|
||||
"anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "null"}]
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_anyof_already_nullable(self):
|
||||
"""Test anyOf that already contains null type."""
|
||||
input_schema = {
|
||||
"anyOf": [{"type": "string"}, {"type": "null"}],
|
||||
"nullable": True,
|
||||
}
|
||||
expected = {"anyOf": [{"type": "string"}, {"type": "null"}]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_allof_with_nullable(self):
|
||||
"""Test nullable handling with allOf constructs."""
|
||||
input_schema = {
|
||||
"allOf": [{"type": "string"}, {"minLength": 1}],
|
||||
"nullable": True,
|
||||
}
|
||||
expected = {
|
||||
"anyOf": [
|
||||
{"allOf": [{"type": "string"}, {"minLength": 1}]},
|
||||
{"type": "null"},
|
||||
]
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_property_level_oneof_with_nullable(self):
|
||||
"""Test nullable handling with oneOf in properties."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"value": {
|
||||
"oneOf": [{"type": "string"}, {"type": "integer"}],
|
||||
"nullable": True,
|
||||
}
|
||||
},
|
||||
}
|
||||
expected = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"value": {
|
||||
"anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "null"}]
|
||||
}
|
||||
},
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
243
tests/utilities/openapi/test_nullable_fields.py
Normal file
243
tests/utilities/openapi/test_nullable_fields.py
Normal file
|
|
@ -0,0 +1,243 @@
|
|||
"""Tests for nullable field handling in OpenAPI schemas."""
|
||||
|
||||
from fastmcp.utilities.openapi import _handle_nullable_fields
|
||||
|
||||
|
||||
class TestHandleNullableFields:
|
||||
"""Test conversion of OpenAPI nullable fields to JSON Schema format."""
|
||||
|
||||
def test_root_level_nullable_string(self):
|
||||
"""Test nullable string at root level."""
|
||||
input_schema = {"type": "string", "nullable": True}
|
||||
expected = {"type": ["string", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_root_level_nullable_integer(self):
|
||||
"""Test nullable integer at root level."""
|
||||
input_schema = {"type": "integer", "nullable": True}
|
||||
expected = {"type": ["integer", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_root_level_nullable_boolean(self):
|
||||
"""Test nullable boolean at root level."""
|
||||
input_schema = {"type": "boolean", "nullable": True}
|
||||
expected = {"type": ["boolean", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_property_level_nullable_fields(self):
|
||||
"""Test nullable fields in properties."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"company": {"type": "string", "nullable": True},
|
||||
"age": {"type": "integer", "nullable": True},
|
||||
"active": {"type": "boolean", "nullable": True},
|
||||
},
|
||||
}
|
||||
expected = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"company": {"type": ["string", "null"]},
|
||||
"age": {"type": ["integer", "null"]},
|
||||
"active": {"type": ["boolean", "null"]},
|
||||
},
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_mixed_nullable_and_non_nullable(self):
|
||||
"""Test mix of nullable and non-nullable fields."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"required_field": {"type": "string"},
|
||||
"optional_nullable": {"type": "string", "nullable": True},
|
||||
"optional_non_nullable": {"type": "string"},
|
||||
},
|
||||
"required": ["required_field"],
|
||||
}
|
||||
expected = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"required_field": {"type": "string"},
|
||||
"optional_nullable": {"type": ["string", "null"]},
|
||||
"optional_non_nullable": {"type": "string"},
|
||||
},
|
||||
"required": ["required_field"],
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_nullable_false_ignored(self):
|
||||
"""Test that nullable: false is ignored (removed but no type change)."""
|
||||
input_schema = {"type": "string", "nullable": False}
|
||||
expected = {"type": "string"}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_no_nullable_field_unchanged(self):
|
||||
"""Test that schemas without nullable field are unchanged."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
}
|
||||
expected = input_schema.copy()
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_nullable_without_type_removes_nullable(self):
|
||||
"""Test that nullable field is removed even without type."""
|
||||
input_schema = {"nullable": True, "description": "Some field"}
|
||||
expected = {"description": "Some field"}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_preserves_other_fields(self):
|
||||
"""Test that other fields are preserved during conversion."""
|
||||
input_schema = {
|
||||
"type": "string",
|
||||
"nullable": True,
|
||||
"description": "A nullable string",
|
||||
"example": "test",
|
||||
"format": "email",
|
||||
}
|
||||
expected = {
|
||||
"type": ["string", "null"],
|
||||
"description": "A nullable string",
|
||||
"example": "test",
|
||||
"format": "email",
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_non_dict_input_unchanged(self):
|
||||
"""Test that non-dict inputs are returned unchanged."""
|
||||
assert _handle_nullable_fields("string") == "string" # type: ignore[arg-type]
|
||||
assert _handle_nullable_fields(123) == 123 # type: ignore[arg-type]
|
||||
assert _handle_nullable_fields(None) is None # type: ignore[arg-type]
|
||||
assert _handle_nullable_fields([1, 2, 3]) == [1, 2, 3] # type: ignore[arg-type]
|
||||
|
||||
def test_performance_optimization_no_copy_when_unchanged(self):
|
||||
"""Test that schemas without nullable fields return the same object (no copy)."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
# Should return the exact same object, not a copy
|
||||
assert result is input_schema
|
||||
|
||||
def test_union_types_with_nullable(self):
|
||||
"""Test nullable handling with existing union types (type as array)."""
|
||||
input_schema = {"type": ["string", "integer"], "nullable": True}
|
||||
expected = {"type": ["string", "integer", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_already_nullable_union_unchanged(self):
|
||||
"""Test that union types already containing null are not modified."""
|
||||
input_schema = {"type": ["string", "null"], "nullable": True}
|
||||
expected = {"type": ["string", "null"]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_property_level_union_with_nullable(self):
|
||||
"""Test nullable handling with union types in properties."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {"value": {"type": ["string", "integer"], "nullable": True}},
|
||||
}
|
||||
expected = {
|
||||
"type": "object",
|
||||
"properties": {"value": {"type": ["string", "integer", "null"]}},
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_complex_union_nullable_scenarios(self):
|
||||
"""Test various complex union type scenarios."""
|
||||
# Already has null in different position
|
||||
input1 = {"type": ["null", "string", "integer"], "nullable": True}
|
||||
result1 = _handle_nullable_fields(input1)
|
||||
assert result1 == {"type": ["null", "string", "integer"]}
|
||||
|
||||
# Single item array
|
||||
input2 = {"type": ["string"], "nullable": True}
|
||||
result2 = _handle_nullable_fields(input2)
|
||||
assert result2 == {"type": ["string", "null"]}
|
||||
|
||||
def test_oneof_with_nullable(self):
|
||||
"""Test nullable handling with oneOf constructs."""
|
||||
input_schema = {
|
||||
"oneOf": [{"type": "string"}, {"type": "integer"}],
|
||||
"nullable": True,
|
||||
}
|
||||
expected = {
|
||||
"anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "null"}]
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_anyof_with_nullable(self):
|
||||
"""Test nullable handling with anyOf constructs."""
|
||||
input_schema = {
|
||||
"anyOf": [{"type": "string"}, {"type": "integer"}],
|
||||
"nullable": True,
|
||||
}
|
||||
expected = {
|
||||
"anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "null"}]
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_anyof_already_nullable(self):
|
||||
"""Test anyOf that already contains null type."""
|
||||
input_schema = {
|
||||
"anyOf": [{"type": "string"}, {"type": "null"}],
|
||||
"nullable": True,
|
||||
}
|
||||
expected = {"anyOf": [{"type": "string"}, {"type": "null"}]}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_allof_with_nullable(self):
|
||||
"""Test nullable handling with allOf constructs."""
|
||||
input_schema = {
|
||||
"allOf": [{"type": "string"}, {"minLength": 1}],
|
||||
"nullable": True,
|
||||
}
|
||||
expected = {
|
||||
"anyOf": [
|
||||
{"allOf": [{"type": "string"}, {"minLength": 1}]},
|
||||
{"type": "null"},
|
||||
]
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
|
||||
def test_property_level_oneof_with_nullable(self):
|
||||
"""Test nullable handling with oneOf in properties."""
|
||||
input_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"value": {
|
||||
"oneOf": [{"type": "string"}, {"type": "integer"}],
|
||||
"nullable": True,
|
||||
}
|
||||
},
|
||||
}
|
||||
expected = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"value": {
|
||||
"anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "null"}]
|
||||
}
|
||||
},
|
||||
}
|
||||
result = _handle_nullable_fields(input_schema)
|
||||
assert result == expected
|
||||
Loading…
Add table
Add a link
Reference in a new issue