Files
2026-07-13 12:38:34 +08:00

1608 lines
56 KiB
Python

"""
Comprehensive test suite for JSON schema to Pydantic conversion functions.
This module tests the core schema parsing functionality in composio.utils.shared,
particularly focusing on the required field propagation bug that was fixed.
"""
import typing as t
import pytest
from pydantic import BaseModel
from pydantic.fields import PydanticUndefined
from composio.utils.shared import (
get_signature_format_from_schema_params,
json_schema_to_fields_dict,
json_schema_to_model,
json_schema_to_pydantic_field,
json_schema_to_pydantic_type,
pydantic_model_from_param_schema,
)
class TestJsonSchemaToPydanticField:
"""Test cases for json_schema_to_pydantic_field function."""
@pytest.mark.unit
@pytest.mark.schema
def test_simple_required_field(self):
"""Test that a field in the required list is marked as required."""
name = "test_field"
json_schema = {
"type": "string",
"description": "A test field",
"title": "Test Field",
}
required = ["test_field"]
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required
)
assert field_name == "test_field"
assert field_type is str
assert field_info.default is PydanticUndefined # Required field marker
def test_simple_optional_field(self):
"""Test that a field not in the required list is marked as optional."""
name = "optional_field"
json_schema = {
"type": "string",
"description": "An optional field",
"title": "Optional Field",
"default": "default_value",
}
required = []
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required
)
assert field_name == "optional_field"
assert field_type is str
assert field_info.default == "default_value"
@pytest.mark.unit
@pytest.mark.schema
def test_nested_object_with_internal_required_not_propagated(self):
"""
CRITICAL TEST: Ensure nested object's internal required array
does NOT make the parent object required.
This tests the specific bug that was fixed.
"""
name = "nested_object"
json_schema = {
"type": "object",
"title": "NestedObject",
"properties": {"inner_field": {"type": "string", "title": "Inner Field"}},
"required": ["inner_field"], # This should NOT make nested_object required
}
required = [] # nested_object is not in parent's required list
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required
)
assert field_name == "nested_object"
assert field_info.default is not PydanticUndefined # Should NOT be required
assert field_info.default is None # Should have default value
def test_nested_object_explicitly_required(self):
"""Test that a nested object can be explicitly required via parent's required list."""
name = "nested_object"
json_schema = {
"type": "object",
"title": "NestedObject",
"properties": {"inner_field": {"type": "string", "title": "Inner Field"}},
"required": ["inner_field"],
}
required = ["nested_object"] # Explicitly in parent's required list
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required
)
assert field_name == "nested_object"
assert field_info.default is PydanticUndefined # Should be required
def test_reserved_field_name_handling(self):
"""Test that reserved Pydantic field names are properly aliased."""
name = "validate" # Reserved name
json_schema = {
"type": "string",
"description": "A field with reserved name",
"title": "Validate",
}
required = []
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required
)
assert field_name == "validate_" # Should be renamed
assert field_info.alias == "validate" # Should preserve original name
def test_reserved_required_field_name_preserves_required_status(self):
"""Test that reserved field names keep the original alias and required status."""
name = "validate"
json_schema = {
"type": "string",
"description": "A required field with reserved name",
"title": "Validate",
}
required = ["validate"]
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required
)
assert field_name == "validate_"
assert field_type is str
assert field_info.alias == "validate"
assert field_info.default is PydanticUndefined
def test_field_with_examples(self):
"""Test that examples are properly preserved in field info."""
name = "example_field"
json_schema = {
"type": "string",
"description": "A field with examples",
"title": "Example Field",
"examples": ["example1", "example2"],
}
required = []
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required
)
assert field_name == "example_field"
assert field_info.examples == ["example1", "example2"]
def test_oneof_field_description_merging(self):
"""Test that oneOf schemas have their descriptions properly merged."""
name = "oneof_field"
json_schema = {
"oneOf": [
{"type": "string", "description": "String option"},
{"type": "integer", "description": "Integer option"},
]
}
required = []
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required
)
expected_desc = "Any of the following options(separated by |): String option | Integer option"
assert field_info.description == expected_desc
def test_skip_default_parameter(self):
"""Test that skip_default parameter works correctly."""
name = "test_field"
json_schema = {"type": "string", "default": "should_be_skipped"}
required = []
field_name, field_type, field_info = json_schema_to_pydantic_field(
name, json_schema, required, skip_default=True
)
# When skip_default=True, field should be required (default=PydanticUndefined)
assert field_info.default is PydanticUndefined
class TestJsonSchemaToModel:
"""Test cases for json_schema_to_model function."""
def test_simple_model_creation(self):
"""Test creating a simple Pydantic model from JSON schema."""
json_schema = {
"title": "SimpleModel",
"type": "object",
"properties": {
"name": {"type": "string", "title": "Name"},
"age": {"type": "integer", "title": "Age"},
},
"required": ["name"],
}
model_class = json_schema_to_model(json_schema)
# Test model creation
instance = model_class(name="test")
assert instance.name == "test"
assert instance.age is None
# Test validation
with pytest.raises(Exception): # Should fail without required field
model_class()
def test_nested_object_model(self):
"""Test creating a model with nested objects."""
json_schema = {
"title": "ParentModel",
"type": "object",
"properties": {
"basic_field": {"type": "string", "title": "Basic Field"},
"nested_object": {
"type": "object",
"title": "NestedObject",
"properties": {
"inner_field": {"type": "string", "title": "Inner Field"}
},
"required": ["inner_field"],
},
},
"required": ["basic_field"],
}
model_class = json_schema_to_model(json_schema)
# Test that nested_object is optional (not required)
instance = model_class(basic_field="test")
assert instance.basic_field == "test"
assert instance.nested_object is None
# Test that nested object validation works when provided
instance_with_nested = model_class(
basic_field="test", nested_object={"inner_field": "nested_value"}
)
assert instance_with_nested.nested_object.inner_field == "nested_value"
@pytest.mark.unit
@pytest.mark.schema
def test_working_location_properties_bug_scenario(self):
"""
CRITICAL TEST: Reproduce the exact scenario that caused the bug.
This tests the workingLocationProperties scenario that was incorrectly
marked as required.
"""
json_schema = {
"title": "CreateEventRequest",
"type": "object",
"properties": {
"start_datetime": {"type": "string", "title": "Start Datetime"},
"workingLocationProperties": {
"type": "object",
"title": "WorkingLocationProperties",
"properties": {
"type": {
"type": "string",
"title": "Type",
"enum": ["homeOffice", "officeLocation", "customLocation"],
},
"customLocation": {
"type": "object",
"title": "WorkingLocationCustom",
"properties": {
"label": {"type": "string", "title": "Label"}
},
"required": ["label"],
},
},
"required": ["type"],
},
},
"required": ["start_datetime"],
}
model_class = json_schema_to_model(json_schema)
# Test that workingLocationProperties is NOT required
instance = model_class(start_datetime="2025-01-01T10:00:00")
assert instance.start_datetime == "2025-01-01T10:00:00"
assert instance.workingLocationProperties is None
# Test that nested validation works when provided
instance_with_working_location = model_class(
start_datetime="2025-01-01T10:00:00",
workingLocationProperties={
"type": "customLocation",
"customLocation": {"label": "Client Office"},
},
)
assert (
instance_with_working_location.workingLocationProperties.type
== "customLocation"
)
def test_array_type_handling(self):
"""Test handling of array types in schema."""
json_schema = {
"title": "ArrayModel",
"type": "object",
"properties": {
"tags": {"type": "array", "items": {"type": "string"}, "title": "Tags"}
},
}
model_class = json_schema_to_model(json_schema)
instance = model_class(tags=["tag1", "tag2"])
assert instance.tags == ["tag1", "tag2"]
class TestPydanticModelFromParamSchema:
"""Test cases for pydantic_model_from_param_schema function."""
def test_simple_param_schema(self):
"""Test creating a model from parameter schema format."""
param_schema = {
"title": "SimpleParam",
"type": "object",
"properties": {"name": {"type": "string", "title": "Name"}},
"required": ["name"],
}
model_class = pydantic_model_from_param_schema(param_schema)
# Should be able to create instance with required field
instance = model_class(name="test")
assert instance.name == "test"
def test_nested_object_not_making_parent_required(self):
"""
CRITICAL TEST: Ensure nested objects with internal required fields
don't make the parent object required in pydantic_model_from_param_schema.
"""
param_schema = {
"title": "ParentParam",
"type": "object",
"properties": {
"required_field": {"type": "string", "title": "Required Field"},
"optional_nested": {
"type": "object",
"title": "Optional Nested",
"properties": {
"inner_required": {"type": "string", "title": "Inner Required"}
},
"required": [
"inner_required"
], # Should NOT make optional_nested required
},
},
"required": ["required_field"],
}
model_class = pydantic_model_from_param_schema(param_schema)
# Should work with just the required field
instance = model_class(required_field="test")
assert instance.required_field == "test"
# optional_nested should be optional (None or default value)
def test_array_type_param_schema(self):
"""Test array type handling in parameter schema."""
param_schema = {
"title": "ArrayParam",
"type": "array",
"items": {"type": "string", "title": "String Item"},
}
result = pydantic_model_from_param_schema(param_schema)
# Should return List[str] type
assert hasattr(result, "__origin__") # Generic type
assert result.__origin__ is list
def test_missing_title_error(self):
"""Test that missing title raises appropriate error."""
param_schema = {
"type": "object",
"properties": {},
# Missing "title"
}
with pytest.raises(ValueError, match="Missing 'title' in param_schema"):
pydantic_model_from_param_schema(param_schema)
class TestJsonSchemaToPydanticType:
"""Test cases for json_schema_to_pydantic_type function."""
def test_basic_types(self):
"""Test conversion of basic JSON schema types to Python types."""
test_cases = [
({"type": "string"}, str),
({"type": "integer"}, int),
({"type": "number"}, float),
({"type": "boolean"}, bool),
]
for json_schema, expected_type in test_cases:
result = json_schema_to_pydantic_type(json_schema)
assert result == expected_type
def test_anyof_null_only_preserves_nullability(self):
"""Test that anyOf with only null maps to Optional[Any] instead of str."""
result = json_schema_to_pydantic_type({"anyOf": [{"type": "null"}]})
assert t.get_origin(result) is t.Union
assert t.Any in t.get_args(result)
assert type(None) in t.get_args(result)
def test_array_type(self):
"""Test array type conversion."""
json_schema = {"type": "array", "items": {"type": "string"}}
result = json_schema_to_pydantic_type(json_schema)
assert hasattr(result, "__origin__")
assert result.__origin__ is list
def test_object_type_creates_nested_model(self):
"""Test that object types create nested Pydantic models."""
json_schema = {
"type": "object",
"title": "NestedModel",
"properties": {"field": {"type": "string", "title": "Field"}},
}
result = json_schema_to_pydantic_type(json_schema)
assert isinstance(result, type)
assert issubclass(result, BaseModel)
def test_oneof_union_types(self):
"""Test oneOf schemas create union types."""
json_schema = {"oneOf": [{"type": "string"}, {"type": "integer"}]}
result = json_schema_to_pydantic_type(json_schema)
# Should create a Union type
assert hasattr(result, "__origin__")
def test_oneof_unlimited_types(self):
"""Test oneOf schemas with unlimited number of types (fixes the 3-type limit bug)."""
# Test 4 types (previously would fail)
json_schema_4 = {
"oneOf": [
{"type": "string"},
{"type": "integer"},
{"type": "boolean"},
{"type": "number"},
]
}
result_4 = json_schema_to_pydantic_type(json_schema_4)
assert hasattr(result_4, "__origin__")
assert result_4.__origin__ is t.Union
assert len(result_4.__args__) == 4
assert str in result_4.__args__
assert int in result_4.__args__
assert bool in result_4.__args__
assert float in result_4.__args__
# Test 5 types
json_schema_5 = {
"oneOf": [
{"type": "string"},
{"type": "integer"},
{"type": "boolean"},
{"type": "number"},
{"type": "array"},
]
}
result_5 = json_schema_to_pydantic_type(json_schema_5)
assert hasattr(result_5, "__origin__")
assert result_5.__origin__ is t.Union
assert len(result_5.__args__) == 5
# Test 6 types (stress test, avoiding null which expands to Optional[Any])
json_schema_6 = {
"oneOf": [
{"type": "string"},
{"type": "integer"},
{"type": "boolean"},
{"type": "number"},
{"type": "array"},
{"type": "object"},
]
}
result_6 = json_schema_to_pydantic_type(json_schema_6)
assert hasattr(result_6, "__origin__")
assert result_6.__origin__ is t.Union
assert len(result_6.__args__) == 6
def test_oneof_single_type(self):
"""Test oneOf with single type returns the type directly."""
json_schema = {"oneOf": [{"type": "string"}]}
result = json_schema_to_pydantic_type(json_schema)
assert result is str
# Single type should not create a Union
assert not hasattr(result, "__origin__")
def test_oneof_with_complex_types(self):
"""Test oneOf with complex types like objects and arrays."""
json_schema = {
"oneOf": [
{"type": "string"},
{"type": "array", "items": {"type": "integer"}},
{
"type": "object",
"title": "ComplexObject",
"properties": {"field": {"type": "string"}},
},
]
}
result = json_schema_to_pydantic_type(json_schema)
assert hasattr(result, "__origin__")
assert result.__origin__ is t.Union
assert len(result.__args__) == 3
# Check that we have string, List[int], and a BaseModel subclass
args = result.__args__
assert str in args
# One should be a List type
list_types = [
arg for arg in args if hasattr(arg, "__origin__") and arg.__origin__ is list
]
assert len(list_types) == 1
# One should be a BaseModel subclass
model_types = [
arg for arg in args if isinstance(arg, type) and issubclass(arg, BaseModel)
]
assert len(model_types) >= 1
def test_oneof_nested_in_object(self):
"""Test oneOf field within an object schema."""
json_schema = {
"type": "object",
"title": "ObjectWithOneOf",
"properties": {
"flexible_field": {
"oneOf": [
{"type": "string"},
{"type": "integer"},
{"type": "boolean"},
{"type": "number"},
]
},
"normal_field": {"type": "string"},
},
"required": ["flexible_field"],
}
# Test that the model can be created
model_class = json_schema_to_model(json_schema)
# Test with different oneOf values
instance1 = model_class(flexible_field="hello", normal_field="world")
instance2 = model_class(flexible_field=42, normal_field="world")
instance3 = model_class(flexible_field=True, normal_field="world")
instance4 = model_class(flexible_field=3.14, normal_field="world")
assert instance1.flexible_field == "hello"
assert instance2.flexible_field == 42
assert instance3.flexible_field is True
assert instance4.flexible_field == 3.14
def test_fallback_to_string(self):
"""Test that missing type defaults to string."""
json_schema = {} # No type specified
result = json_schema_to_pydantic_type(json_schema)
assert result is str
def test_unsupported_type_fallback(self):
"""Test that unsupported types fall back to string (graceful degradation)."""
json_schema = {"type": "unsupported_type"}
# The library gracefully falls back to string instead of raising an error
result = json_schema_to_pydantic_type(json_schema)
assert result is str
def test_anyof_nullable_object(self):
"""Test anyOf with object and null types (common for nullable fields)."""
json_schema = {
"anyOf": [{"type": "object", "additionalProperties": {}}, {"type": "null"}]
}
result = json_schema_to_pydantic_type(json_schema)
# Should return Optional[dict], not str (to allow both dict and None)
assert hasattr(result, "__origin__")
assert result.__origin__ is t.Union
# The library returns `dict` (the concrete type) instead of `typing.Dict`
assert dict in result.__args__ or t.Dict in result.__args__
assert type(None) in result.__args__
def test_anyof_nullable_object_with_properties(self):
"""Test anyOf with object (with properties) and null types."""
json_schema = {
"anyOf": [
{
"type": "object",
"title": "CustomFields",
"properties": {"field1": {"type": "string"}},
},
{"type": "null"},
]
}
result = json_schema_to_pydantic_type(json_schema)
# Should return Optional[BaseModel subclass] (Union with None)
assert hasattr(result, "__origin__")
assert result.__origin__ is t.Union
# One of the args should be a BaseModel subclass
model_types = [
arg
for arg in result.__args__
if isinstance(arg, type) and issubclass(arg, BaseModel)
]
assert len(model_types) == 1
# None should also be in the union
assert type(None) in result.__args__
def test_anyof_multiple_types(self):
"""Test anyOf with multiple non-null types."""
json_schema = {
"anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "object"}]
}
result = json_schema_to_pydantic_type(json_schema)
assert hasattr(result, "__origin__")
assert result.__origin__ is t.Union
def test_anyof_single_type(self):
"""Test anyOf with single type returns the type directly."""
json_schema = {"anyOf": [{"type": "string"}]}
result = json_schema_to_pydantic_type(json_schema)
assert result is str
assert not hasattr(result, "__origin__")
def test_allof_single_option(self):
"""Test allOf with single option."""
json_schema = {
"allOf": [
{
"type": "object",
"properties": {"name": {"type": "string"}},
"title": "Test",
}
]
}
result = json_schema_to_pydantic_type(json_schema)
assert isinstance(result, type)
assert issubclass(result, BaseModel)
def test_allof_multiple_options_with_type(self):
"""Test allOf with multiple options where one has type."""
json_schema = {
"allOf": [{"description": "Some description"}, {"type": "string"}]
}
result = json_schema_to_pydantic_type(json_schema)
# The library creates an AllOfModel class that combines the schemas
# or returns str depending on the schema structure
assert result is not None
# Either it's str or a model class (both are valid)
assert result is str or (
isinstance(result, type) and issubclass(result, BaseModel)
)
def test_allof_empty_options(self):
"""Test allOf with empty options falls back to string."""
json_schema = {"allOf": []}
result = json_schema_to_pydantic_type(json_schema)
assert result is str
class TestJsonSchemaToFieldsDict:
"""Test cases for json_schema_to_fields_dict function."""
def test_basic_fields_dict(self):
"""Test creating fields dictionary from JSON schema."""
json_schema = {
"properties": {
"name": {"type": "string", "title": "Name"},
"age": {"type": "integer", "title": "Age"},
},
"required": ["name"],
}
fields_dict = json_schema_to_fields_dict(json_schema)
assert "name" in fields_dict
assert "age" in fields_dict
# Check field types and info
name_type, name_field = fields_dict["name"]
age_type, age_field = fields_dict["age"]
assert name_type is str
assert age_type is int
assert name_field.default is PydanticUndefined # Required
assert age_field.default is None # Optional
class TestRegressionScenarios:
"""Test cases for specific regression scenarios and edge cases."""
def test_deeply_nested_objects_required_propagation(self):
"""Test deeply nested objects don't propagate required fields incorrectly."""
json_schema = {
"title": "DeeplyNested",
"type": "object",
"properties": {
"level1": {
"type": "object",
"title": "Level1",
"properties": {
"level2": {
"type": "object",
"title": "Level2",
"properties": {
"level3": {
"type": "object",
"title": "Level3",
"properties": {
"deep_field": {
"type": "string",
"title": "Deep Field",
}
},
"required": ["deep_field"],
}
},
"required": ["level3"],
}
},
"required": ["level2"],
}
},
"required": [], # level1 should NOT be required
}
model_class = json_schema_to_model(json_schema)
# Should be able to create instance without level1
instance = model_class()
assert instance.level1 is None
def test_multiple_nested_objects_same_level(self):
"""Test multiple nested objects at same level with different required fields."""
json_schema = {
"title": "MultipleNested",
"type": "object",
"properties": {
"config1": {
"type": "object",
"title": "Config1",
"properties": {"setting1": {"type": "string", "title": "Setting1"}},
"required": ["setting1"],
},
"config2": {
"type": "object",
"title": "Config2",
"properties": {"setting2": {"type": "string", "title": "Setting2"}},
"required": ["setting2"],
},
"required_field": {"type": "string", "title": "Required Field"},
},
"required": ["required_field"],
}
model_class = json_schema_to_model(json_schema)
# Should work with just required_field
instance = model_class(required_field="test")
assert instance.required_field == "test"
assert instance.config1 is None
assert instance.config2 is None
def test_empty_required_array_handling(self):
"""Test handling of empty required arrays."""
json_schema = {
"title": "EmptyRequired",
"type": "object",
"properties": {
"optional1": {"type": "string", "title": "Optional1"},
"optional2": {"type": "string", "title": "Optional2"},
},
"required": [],
}
model_class = json_schema_to_model(json_schema)
# Should work with no fields
instance = model_class()
assert instance.optional1 is None
assert instance.optional2 is None
def test_missing_required_array_handling(self):
"""Test handling when required array is missing entirely."""
json_schema = {
"title": "NoRequired",
"type": "object",
"properties": {
"field1": {"type": "string", "title": "Field1"},
"field2": {"type": "string", "title": "Field2"},
},
# No "required" key at all
}
model_class = json_schema_to_model(json_schema)
# Should work with no fields (all optional)
instance = model_class()
assert instance.field1 is None
assert instance.field2 is None
class TestEdgeCases:
"""Test cases for edge cases and error conditions."""
def test_none_schema_handling(self):
"""Test handling of None or empty schemas."""
with pytest.raises((TypeError, AttributeError, KeyError)):
json_schema_to_model(None)
def test_malformed_schema_handling(self):
"""Test handling of malformed schemas."""
malformed_schemas = [
{"type": "object"}, # Missing properties
{"properties": {}}, # Missing type and title
{"title": "Test", "type": "invalid_type"}, # Invalid type
]
for schema in malformed_schemas:
# Should either handle gracefully or raise appropriate error
try:
result = json_schema_to_model(schema)
# If it doesn't raise an error, it should at least return something
assert result is not None
except (ValueError, KeyError, TypeError):
# These are acceptable errors for malformed schemas
pass
def test_circular_reference_protection(self):
"""Test that circular references don't cause infinite recursion."""
# This is a complex scenario that would require special handling
# For now, we just ensure it doesn't crash
json_schema = {
"title": "SelfReference",
"type": "object",
"properties": {
"name": {"type": "string", "title": "Name"},
"children": {
"type": "array",
"items": {"$ref": "#"}, # Self-reference
},
},
}
# This might not work perfectly but shouldn't crash
try:
model_class = json_schema_to_model(json_schema)
# If successful, test basic functionality
instance = model_class(name="test")
assert instance.name == "test"
except (RecursionError, ValueError):
# Acceptable for now - circular references are complex
pass
class TestCrewAICustomFieldsBug:
"""Regression tests for PLEN-1177 - CustomFields type error with CrewAI."""
@pytest.mark.unit
@pytest.mark.schema
def test_custom_fields_object_type_preserved(self):
"""
Test that CustomFields with anyOf [object, null] returns Dict, not str.
This is the exact bug scenario from PLEN-1177.
"""
# Schema similar to what Salesforce tools return for CustomFields
json_schema = {
"title": "CreateLeadRequest",
"type": "object",
"properties": {
"Company": {"type": "string", "title": "Company"},
"LastName": {"type": "string", "title": "LastName"},
"CustomFields": {
"anyOf": [
{"type": "object", "additionalProperties": {}},
{"type": "null"},
],
"default": None,
"description": "Dictionary of custom field API names and their values.",
"title": "CustomFields",
},
},
"required": ["Company", "LastName"],
}
model_class = json_schema_to_model(json_schema)
# The model should accept a dict for CustomFields
instance = model_class(
Company="Test Corp",
LastName="Smith",
CustomFields={"Custom_Field__c": "Value"},
)
assert instance.CustomFields == {"Custom_Field__c": "Value"}
# The model should also accept None
instance_none = model_class(
Company="Test Corp",
LastName="Smith",
CustomFields=None,
)
assert instance_none.CustomFields is None
@pytest.mark.unit
@pytest.mark.schema
def test_pydantic_model_json_schema_preserves_object_type(self):
"""
Test that when Pydantic model is converted back to JSON schema,
the object type is preserved (not converted to string).
"""
json_schema = {
"title": "TestModel",
"type": "object",
"properties": {
"custom_dict": {
"anyOf": [{"type": "object"}, {"type": "null"}],
"default": None,
}
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
# The generated schema should have object type, not string
custom_dict_schema = generated_schema["properties"]["custom_dict"]
# It might be wrapped in anyOf or have direct type
if "anyOf" in custom_dict_schema:
types = [opt.get("type") for opt in custom_dict_schema["anyOf"]]
assert "object" in types
assert "string" not in types
else:
assert (
custom_dict_schema.get("type") == "object"
or custom_dict_schema.get("additionalProperties") is not None
)
@pytest.mark.unit
@pytest.mark.schema
def test_anyof_with_nested_custom_fields(self):
"""
Test anyOf handling with more complex nested CustomFields scenario.
"""
json_schema = {
"title": "SalesforceRequest",
"type": "object",
"properties": {
"leadData": {
"type": "object",
"title": "LeadData",
"properties": {
"Name": {"type": "string", "title": "Name"},
"CustomFields": {
"anyOf": [
{"type": "object", "additionalProperties": {}},
{"type": "null"},
],
"default": None,
},
},
"required": ["Name"],
}
},
"required": ["leadData"],
}
model_class = json_schema_to_model(json_schema)
# Test with CustomFields as dict
instance = model_class(
leadData={"Name": "John", "CustomFields": {"Industry__c": "Tech"}}
)
assert instance.leadData.CustomFields == {"Industry__c": "Tech"}
# Test with CustomFields as None
instance_none = model_class(leadData={"Name": "John", "CustomFields": None})
assert instance_none.leadData.CustomFields is None
class TestBooleanSchemas:
"""Test cases for JSON Schema boolean schema handling (draft-06+)."""
@pytest.mark.unit
@pytest.mark.schema
def test_true_schema_in_anyof(self):
"""Test that true boolean schema in anyOf doesn't crash."""
json_schema = {
"anyOf": [
{"type": "string"},
True, # Boolean schema
]
}
result = json_schema_to_pydantic_type(json_schema)
# Should handle gracefully, creating a union including Any
assert result is not None
# The result should include Any type from the true schema
assert hasattr(result, "__origin__")
assert result.__origin__ is t.Union
@pytest.mark.unit
@pytest.mark.schema
def test_false_schema_in_anyof(self):
"""Test that false boolean schema in anyOf doesn't crash."""
json_schema = {
"anyOf": [
{"type": "string"},
False, # Boolean schema - should be filtered out
]
}
result = json_schema_to_pydantic_type(json_schema)
# Should handle gracefully, false schema filtered out leaving just string
assert result is str
@pytest.mark.unit
@pytest.mark.schema
def test_boolean_schema_in_allof_with_type(self):
"""Test that boolean schemas in allOf don't crash when combined with typed schema."""
json_schema = {
"allOf": [
{"type": "string"},
True,
]
}
result = json_schema_to_pydantic_type(json_schema)
# Should not crash - may return str or an AllOf model
assert result is not None
# Either it's str or a model class (library creates AllOfModel)
assert result is str or (
isinstance(result, type) and issubclass(result, BaseModel)
)
@pytest.mark.unit
@pytest.mark.schema
def test_boolean_schema_in_allof_single(self):
"""Test that single boolean schema in allOf returns appropriate type."""
json_schema = {"allOf": [True]}
result = json_schema_to_pydantic_type(json_schema)
# Single true schema filtered to {} - library may return Any or a model
assert result is not None
# Could be Any, or a generated model (both are valid)
assert result is t.Any or (
isinstance(result, type) and issubclass(result, BaseModel)
)
@pytest.mark.unit
@pytest.mark.schema
def test_boolean_schema_in_oneof(self):
"""Test that boolean schemas in oneOf don't crash."""
json_schema = {
"oneOf": [
{"type": "string"},
True,
]
}
result = json_schema_to_pydantic_type(json_schema)
# Should handle gracefully
assert result is not None
assert hasattr(result, "__origin__")
assert result.__origin__ is t.Union
@pytest.mark.unit
@pytest.mark.schema
def test_standalone_true_schema(self):
"""Test that standalone true schema returns Any."""
result = json_schema_to_pydantic_type(True)
assert result is t.Any
@pytest.mark.unit
@pytest.mark.schema
def test_standalone_false_schema(self):
"""Test that standalone false schema returns None (to be filtered out)."""
result = json_schema_to_pydantic_type(False)
assert result is None
@pytest.mark.unit
@pytest.mark.schema
def test_only_false_schemas_in_anyof(self):
"""Test anyOf with only false schemas falls back to string."""
json_schema = {
"anyOf": [
False,
False,
]
}
result = json_schema_to_pydantic_type(json_schema)
# All false schemas filtered out, should fall back to string
assert result is str
@pytest.mark.unit
@pytest.mark.schema
def test_mixed_boolean_schemas_in_anyof(self):
"""Test anyOf with mixed true and false schemas."""
json_schema = {
"anyOf": [
True,
False,
{"type": "integer"},
]
}
result = json_schema_to_pydantic_type(json_schema)
# Should create union of Any and int (false filtered out)
assert result is not None
assert hasattr(result, "__origin__")
assert result.__origin__ is t.Union
class TestBooleanDefaultCoercion:
"""Regression tests for PLEN-1311 - Boolean default type mismatch in LangchainProvider."""
@pytest.mark.unit
@pytest.mark.schema
def test_anyof_boolean_null_with_boolean_default(self):
"""
Test that anyOf [boolean, null] with boolean default preserves types.
Regression test for PLEN-1311: GOOGLEDRIVE_FIND_FILE supportsAllDrives
field was incorrectly converted to string type with string default.
"""
json_schema = {
"title": "FindFileRequest",
"type": "object",
"properties": {
"supportsAllDrives": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"default": True,
"description": "Whether to search all drives.",
},
},
}
model_class = json_schema_to_model(json_schema)
# Verify the model accepts boolean values
instance = model_class(supportsAllDrives=True)
assert instance.supportsAllDrives is True
instance_false = model_class(supportsAllDrives=False)
assert instance_false.supportsAllDrives is False
# Verify the generated JSON schema preserves boolean type
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["supportsAllDrives"]
# Should NOT be string type
assert prop.get("type") != "string"
# Default should be boolean True, not string "true"
assert prop.get("default") is True
@pytest.mark.unit
@pytest.mark.schema
def test_anyof_boolean_null_with_string_default_coerced(self):
"""
Test that string "true"/"false" defaults are coerced to boolean.
This handles cases where the API returns stringified boolean defaults.
"""
json_schema = {
"title": "FindFileRequest",
"type": "object",
"properties": {
"supportsAllDrives": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"default": "true", # String, should be coerced to True
"description": "Whether to search all drives.",
},
},
}
model_class = json_schema_to_model(json_schema)
# Verify the model accepts boolean values
instance = model_class(supportsAllDrives=True)
assert instance.supportsAllDrives is True
# Verify the generated JSON schema has coerced default
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["supportsAllDrives"]
# Should NOT have string type
assert prop.get("type") != "string"
# Default should be coerced to boolean True
assert prop.get("default") is True
@pytest.mark.unit
@pytest.mark.schema
def test_anyof_boolean_null_with_string_false_default_coerced(self):
"""
Test that string "false" default is coerced to boolean False.
"""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"enabled": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"default": "false", # String, should be coerced to False
"description": "Enable feature.",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["enabled"]
# Default should be coerced to boolean False
assert prop.get("default") is False
@pytest.mark.unit
@pytest.mark.schema
def test_integer_with_string_default_coerced(self):
"""Test that string integer defaults are coerced."""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"page": {
"type": "integer",
"default": "1", # String, should be coerced to 1
"description": "Page number",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["page"]
# Default should be integer 1, not string "1"
assert prop.get("default") == 1
assert isinstance(prop.get("default"), int)
@pytest.mark.unit
@pytest.mark.schema
def test_float_with_string_default_coerced(self):
"""Test that string float defaults are coerced."""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"rate": {
"type": "number",
"default": "3.14", # String, should be coerced to 3.14
"description": "Rate value",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["rate"]
# Default should be float 3.14, not string "3.14"
assert prop.get("default") == 3.14
assert isinstance(prop.get("default"), float)
@pytest.mark.unit
@pytest.mark.schema
def test_boolean_default_not_coerced_when_already_correct(self):
"""Test that boolean defaults that are already correct are not modified."""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"enabled": {
"type": "boolean",
"default": True, # Already boolean
"description": "Enable feature.",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["enabled"]
assert prop.get("default") is True
assert isinstance(prop.get("default"), bool)
@pytest.mark.unit
@pytest.mark.schema
def test_string_default_not_coerced_for_string_type(self):
"""Test that string defaults for string type fields are preserved."""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"name": {
"type": "string",
"default": "true", # String value, should stay as string
"description": "Name field",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["name"]
# Default should stay as string "true"
assert prop.get("default") == "true"
assert isinstance(prop.get("default"), str)
@pytest.mark.unit
@pytest.mark.schema
def test_allof_boolean_with_string_default_coerced(self):
"""Test that allOf with boolean type coerces string default."""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"flag": {
"allOf": [{"type": "boolean"}],
"default": "true",
"description": "Flag field",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["flag"]
# Default should be coerced to boolean True
assert prop.get("default") is True
@pytest.mark.unit
@pytest.mark.schema
def test_invalid_boolean_string_not_coerced(self):
"""Test that invalid boolean strings are not coerced and return as-is."""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"flag": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"default": "invalid", # Not a valid boolean string
"description": "Flag field",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["flag"]
# Default should remain as string "invalid" since it can't be coerced
assert prop.get("default") == "invalid"
@pytest.mark.unit
@pytest.mark.schema
def test_empty_string_default_not_coerced(self):
"""Test that empty string defaults are not coerced."""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"value": {
"anyOf": [{"type": "integer"}, {"type": "null"}],
"default": "", # Empty string
"description": "Value field",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["value"]
# Default should remain as empty string since it can't be coerced to int
assert prop.get("default") == ""
@pytest.mark.unit
@pytest.mark.schema
def test_non_string_default_not_coerced(self):
"""Test that non-string defaults (like int, list) are returned as-is."""
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"count": {
"type": "integer",
"default": 42, # Already an integer
"description": "Count field",
},
"items": {
"type": "array",
"default": [1, 2, 3], # Already a list
"description": "Items field",
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
count_prop = generated_schema["properties"]["count"]
assert count_prop.get("default") == 42
assert isinstance(count_prop.get("default"), int)
items_prop = generated_schema["properties"]["items"]
assert items_prop.get("default") == [1, 2, 3]
assert isinstance(items_prop.get("default"), list)
@pytest.mark.unit
@pytest.mark.schema
def test_boolean_coercion_case_insensitive(self):
"""Test that boolean coercion handles various case combinations."""
test_cases = [
("TRUE", True),
("True", True),
("FALSE", False),
("False", False),
("YES", True),
("Yes", True),
("NO", False),
("No", False),
]
for string_value, expected_bool in test_cases:
json_schema = {
"title": "TestRequest",
"type": "object",
"properties": {
"flag": {
"type": "boolean",
"default": string_value,
},
},
}
model_class = json_schema_to_model(json_schema)
generated_schema = model_class.model_json_schema()
prop = generated_schema["properties"]["flag"]
assert prop.get("default") is expected_bool, (
f"Expected '{string_value}' to coerce to {expected_bool}"
)
class TestGetSignatureFormatFromSchemaParams:
"""Test cases for get_signature_format_from_schema_params union handling."""
@staticmethod
def _annotation(schema):
params = get_signature_format_from_schema_params(schema)
assert len(params) == 1
return params[0].annotation
@pytest.mark.unit
@pytest.mark.schema
def test_oneof_four_members(self):
"""oneOf with 4 options builds a Union instead of raising ValueError."""
schema = {
"properties": {
"value": {
"oneOf": [
{"type": "string"},
{"type": "integer"},
{"type": "boolean"},
{"type": "number"},
]
}
}
}
annotation = self._annotation(schema)
assert t.get_origin(annotation) is t.Union
assert set(t.get_args(annotation)) == {str, int, bool, float}
@pytest.mark.unit
@pytest.mark.schema
def test_oneof_five_members(self):
"""oneOf with more than four options is also supported (no 3-member cap)."""
schema = {
"properties": {
"value": {
"oneOf": [
{"type": "string"},
{"type": "integer"},
{"type": "boolean"},
{"type": "number"},
{"type": "array"},
]
}
}
}
annotation = self._annotation(schema)
assert t.get_origin(annotation) is t.Union
assert len(t.get_args(annotation)) == 5
@pytest.mark.unit
@pytest.mark.schema
def test_anyof_option_missing_type(self):
"""An anyOf option without a 'type' key maps to Any instead of raising KeyError."""
schema = {
"properties": {
"value": {
"anyOf": [
{"description": "free-form value"},
{"type": "string"},
]
}
}
}
annotation = self._annotation(schema)
assert t.get_origin(annotation) is t.Union
args = t.get_args(annotation)
assert t.Any in args
assert str in args
@pytest.mark.unit
@pytest.mark.schema
def test_anyof_all_options_missing_type(self):
"""anyOf where every option lacks a 'type' collapses to a single Any annotation."""
schema = {
"properties": {
"value": {
"anyOf": [
{"description": "a"},
{"description": "b"},
]
}
}
}
assert self._annotation(schema) is t.Any
@pytest.mark.unit
@pytest.mark.schema
def test_oneof_single_member(self):
"""oneOf with a single option resolves to that type directly (unchanged)."""
schema = {"properties": {"value": {"oneOf": [{"type": "string"}]}}}
assert self._annotation(schema) is str
@pytest.mark.unit
@pytest.mark.schema
def test_oneof_two_members(self):
"""Two-member oneOf behavior is preserved (Union of the two types)."""
schema = {
"properties": {
"value": {"oneOf": [{"type": "string"}, {"type": "integer"}]}
}
}
assert self._annotation(schema) == t.Union[str, int]
@pytest.mark.unit
@pytest.mark.schema
def test_oneof_three_members(self):
"""Three-member oneOf behavior is preserved (Union of the three types)."""
schema = {
"properties": {
"value": {
"oneOf": [
{"type": "string"},
{"type": "integer"},
{"type": "boolean"},
]
}
}
}
assert self._annotation(schema) == t.Union[str, int, bool]
@pytest.mark.unit
@pytest.mark.schema
def test_anyof_with_null_member_still_resolves(self):
"""Nullable anyOf [type, null] continues to resolve without raising."""
schema = {
"properties": {"value": {"anyOf": [{"type": "string"}, {"type": "null"}]}}
}
annotation = self._annotation(schema)
assert t.get_origin(annotation) is t.Union
args = t.get_args(annotation)
assert str in args
assert type(None) in args
if __name__ == "__main__":
pytest.main([__file__, "-v"])