""" 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, TypeAdapter, ValidationError 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, ) from tests.fixtures.json_schema_conversion_corpus import find_case, load_object_cases 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"] @pytest.mark.parametrize( ("title", "expected_model_name"), [ ({}, "GeneratedModel"), ({"title": None}, "GeneratedModel"), ({"title": ""}, ""), ], ) def test_title_uses_expected_model_name(self, title, expected_model_name): """Missing or null titles must not reach Pydantic's model factory.""" json_schema = { **title, "type": "object", "properties": { "name": {"type": "string"}, "value": {"type": "integer"}, }, "required": ["name"], } model_class = json_schema_to_model(json_schema) assert model_class.__name__ == expected_model_name instance = model_class(name="test", value=42) assert instance.name == "test" assert instance.value == 42 with pytest.raises(ValidationError): model_class() 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) def _materialized(annotation: t.Any) -> t.Any: """Strip the exact-validation `Annotated` wrapper to reach the Pydantic type.""" while t.get_origin(annotation) is t.Annotated: annotation = t.get_args(annotation)[0] return annotation 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 accepts no non-null value.""" result = _materialized( json_schema_to_pydantic_type({"anyOf": [{"type": "null"}]}) ) assert result is type(None) 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) union_4 = t.get_args(result_4)[0] assert t.get_origin(union_4) is t.Union assert len(t.get_args(union_4)) == 4 assert str in t.get_args(union_4) assert int in t.get_args(union_4) assert bool in t.get_args(union_4) assert float in t.get_args(union_4) # `integer` is a subset of `number`, so 42 matches two branches and # must be rejected by oneOf's exactly-one rule. with pytest.raises(ValidationError): TypeAdapter(result_4).validate_python(42) # 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) union_5 = t.get_args(result_5)[0] assert t.get_origin(union_5) is t.Union assert len(t.get_args(union_5)) == 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) union_6 = t.get_args(result_6)[0] assert t.get_origin(union_6) is t.Union assert len(t.get_args(union_6)) == 6 def test_oneof_single_type(self): """Test oneOf with single type returns the type directly.""" json_schema = {"oneOf": [{"type": "string"}]} result = _materialized(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 = _materialized(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") 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 instance3.flexible_field is True assert instance4.flexible_field == 3.14 # Draft 7 `oneOf` requires exactly one branch to match. An integer # satisfies both {"type": "integer"} and {"type": "number"}, so it is # rejected — the Zod and Effect converters agree. with pytest.raises(ValidationError): model_class(flexible_field=42, normal_field="world") def test_empty_schema_accepts_any_value(self): """An empty JSON Schema accepts every value.""" json_schema = {} # No type specified result = json_schema_to_pydantic_type(json_schema) assert result is t.Any 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 = _materialized(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 = _materialized(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 = _materialized(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 = _materialized(json_schema_to_pydantic_type(json_schema)) assert result is str assert not hasattr(result, "__origin__") def test_allof_single_option(self): """A single allOf option preserves its acceptance behavior.""" json_schema = { "allOf": [ { "type": "object", "properties": {"name": {"type": "string"}}, "title": "Test", } ] } adapter = TypeAdapter(json_schema_to_pydantic_type(json_schema)) result = adapter.validate_python({"name": "Ada"}) assert adapter.dump_python(result, mode="json") == {"name": "Ada"} with pytest.raises(ValidationError): adapter.validate_python({"name": 42}) 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"}] } adapter = TypeAdapter(json_schema_to_pydantic_type(json_schema)) assert adapter.validate_python("value") == "value" with pytest.raises(ValidationError): adapter.validate_python(42) def test_allof_empty_options(self): """Test allOf with empty options accepts any value.""" json_schema = {"allOf": []} adapter = TypeAdapter(json_schema_to_pydantic_type(json_schema)) assert adapter.validate_python(42) == 42 assert adapter.validate_python({"anything": True}) == {"anything": True} 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 = _materialized(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 = _materialized(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): """A true allOf branch leaves the typed branch unchanged.""" json_schema = { "allOf": [ {"type": "string"}, True, ] } adapter = TypeAdapter(json_schema_to_pydantic_type(json_schema)) assert adapter.validate_python("value") == "value" with pytest.raises(ValidationError): adapter.validate_python(42) @pytest.mark.unit @pytest.mark.schema def test_boolean_schema_in_allof_single(self): """A single true allOf branch accepts every value.""" json_schema = {"allOf": [True]} adapter = TypeAdapter(json_schema_to_pydantic_type(json_schema)) assert adapter.validate_python(42) == 42 assert adapter.validate_python({"anything": True}) == {"anything": True} @pytest.mark.unit @pytest.mark.schema def test_boolean_schema_in_oneof(self): """A true branch still participates in oneOf's exactly-one rule.""" json_schema = { "oneOf": [ {"type": "string"}, True, ] } result = json_schema_to_pydantic_type(json_schema) adapter = TypeAdapter(result) assert adapter.validate_python(42) == 42 with pytest.raises(ValidationError): adapter.validate_python("matches both branches") @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 rejects every value.""" adapter = TypeAdapter(json_schema_to_pydantic_type(False)) for value in (None, "value", 1, {}, []): with pytest.raises(ValidationError): adapter.validate_python(value) @pytest.mark.unit @pytest.mark.schema def test_only_false_schemas_in_anyof(self): """Test anyOf with only false schemas rejects every value.""" json_schema = { "anyOf": [ False, False, ] } result = json_schema_to_pydantic_type(json_schema) adapter = TypeAdapter(result) for value in (None, "some string", 123, {}, []): with pytest.raises(ValidationError): adapter.validate_python(value) @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 = _materialized(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 _parameter(schema): params = get_signature_format_from_schema_params(schema) assert len(params) == 1 return params[0] @classmethod def _annotation(cls, schema): return cls._parameter(schema).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 @pytest.mark.unit @pytest.mark.schema @pytest.mark.parametrize( ("schema_type", "expected_members"), [ (["string", "null"], {str, type(None)}), (["integer"], {int}), (["string", "integer"], {str, int}), (["file", "null"], {t.Any, type(None)}), ], ) def test_type_list_resolves(self, schema_type, expected_members): """List-valued JSON Schema types produce an annotation without raising.""" annotation = self._annotation({"properties": {"value": {"type": schema_type}}}) actual = set(t.get_args(annotation)) if t.get_args(annotation) else {annotation} assert expected_members <= actual @pytest.mark.unit @pytest.mark.schema @pytest.mark.parametrize( ("schema_type", "expected_annotation", "expected_default"), [ (["integer"], int, 0), (["boolean"], bool, False), (["array"], t.List, []), ], ) def test_single_type_list_uses_scalar_fallback( self, schema_type, expected_annotation, expected_default ): """A one-item type list keeps its scalar type's implicit default.""" parameter = self._parameter({"properties": {"value": {"type": schema_type}}}) assert parameter.annotation is expected_annotation assert parameter.default == expected_default @pytest.mark.unit @pytest.mark.schema def test_single_type_list_preserves_explicit_default(self): """An explicit default overrides the fallback for a one-item type list.""" parameter = self._parameter( {"properties": {"value": {"type": ["integer"], "default": 42}}} ) assert parameter.default == 42 @pytest.mark.unit @pytest.mark.schema def test_multi_type_list_keeps_union_fallback(self): """A multi-type list remains a union with the existing empty-string fallback.""" parameter = self._parameter( {"properties": {"value": {"type": ["integer", "null"]}}} ) assert {int, type(None)} <= set(t.get_args(parameter.annotation)) assert parameter.default == "" @pytest.mark.unit @pytest.mark.schema def test_combiner_option_with_type_list_resolves(self): """List-valued types inside combiners do not reach a scalar dict lookup.""" schema = { "properties": { "value": { "anyOf": [ {"type": ["string", "null"]}, {"type": "integer"}, ] } } } annotation = self._annotation(schema) assert {str, int, type(None)} <= set(t.get_args(annotation)) if __name__ == "__main__": pytest.main([__file__, "-v"]) class TestSharedObjectCorpusThroughJsonSchemaToModel: """`json_schema_to_model` must satisfy the shared cross-SDK object contract. This is the provider-facing entry point (`args_schema`), so a model that accepts an argument but discards it is as much a defect as one that rejects a valid payload. """ @pytest.mark.parametrize( "case,index", [ (case, index) for case in load_object_cases() for index in range(len(case.instances)) ], ids=[ f"{case.id}[{index}]" for case in load_object_cases() for index in range(len(case.instances)) ], ) def test_case(self, case, index: int) -> None: instance = case.instances[index] model = json_schema_to_model(case.schema_) if not instance.accepted_for("python"): with pytest.raises(ValidationError): model.model_validate(instance.input) return result = model.model_validate(instance.input) if instance.python is not None and instance.python.has_output: assert ( result.model_dump(mode="json", by_alias=True) == instance.python.output ) def test_free_form_content_is_readable_as_attributes(self) -> None: """Preserved dynamic keys must survive `getattr`, not just `model_dump`.""" case = find_case("root-free-form-absent-properties") model = json_schema_to_model(case.schema_) result = model.model_validate(case.instances[0].input) assert getattr(result, "anything") == {"a": 1} assert getattr(result, "other") == "x" @pytest.mark.parametrize( ("case_id", "expected"), [ ("named-properties-strict-by-default", {"additionalProperties": False}), ( "root-additional-properties-schema-valued", {"additionalProperties": {"type": "number"}}, ), ( "pattern-only-object", {"patternProperties": {"^s_": {"type": "string"}}}, ), ], ) def test_model_json_schema_preserves_root_object_policy( self, case_id: str, expected: t.Dict[str, t.Any], ) -> None: model = json_schema_to_model(find_case(case_id).schema_) advertised = model.model_json_schema() for key, value in expected.items(): assert advertised[key] == value def test_dynamic_schema_resolves_local_json_pointer(self) -> None: schema = { "$defs": {"positive": {"type": "integer", "minimum": 1}}, "type": "object", "patternProperties": {"^count_": {"$ref": "#/$defs/positive"}}, } model = json_schema_to_model(schema) assert model.model_validate({"count_a": 1}).model_dump() == {"count_a": 1} with pytest.raises(ValidationError): model.model_validate({"count_a": 0}) def test_nested_dynamic_schema_resolves_against_document_root(self) -> None: schema = { "$defs": {"positive": {"type": "integer", "minimum": 1}}, "type": "object", "properties": { "payload": { "type": "object", "patternProperties": {"^count_": {"$ref": "#/$defs/positive"}}, "additionalProperties": False, } }, "required": ["payload"], } model = json_schema_to_model(schema) result = model.model_validate({"payload": {"count_a": 1}}) assert result.model_dump() == {"payload": {"count_a": 1}} with pytest.raises(ValidationError): model.model_validate({"payload": {"count_a": 0}}) def test_dynamic_schema_materialization_cannot_reject_valid_input(self) -> None: schema = { "type": "object", "patternProperties": {"^value_": {"enum": [1], "default": 1}}, } model = json_schema_to_model(schema) assert model.model_validate({"value_a": 1}).model_dump() == {"value_a": 1} @pytest.mark.parametrize( "reference", ["#/$defs/missing", "https://example.com/schema.json"], ) def test_dynamic_schema_rejects_unresolvable_reference( self, reference: str, ) -> None: schema = { "type": "object", "patternProperties": {"^value_": {"$ref": reference}}, } with pytest.raises(ValueError, match="schema reference"): json_schema_to_model(schema) def test_dynamic_schema_rejects_invalid_pattern(self) -> None: schema = { "type": "object", "patternProperties": {"[": {"type": "string"}}, } with pytest.raises(ValueError, match="Invalid patternProperties"): json_schema_to_model(schema) def test_dynamic_schema_checks_references_inside_local_target(self) -> None: schema = { "$defs": {"nested": {"$ref": "https://example.com/external-schema.json"}}, "type": "object", "patternProperties": {"^value_": {"$ref": "#/$defs/nested"}}, } with pytest.raises(ValueError, match="must be a local JSON Pointer"): json_schema_to_model(schema) @pytest.mark.parametrize( "keyword,value,accepted", [ ( "default", {"$ref": "https://example.com/schema.json"}, {"$ref": "kept"}, ), ("const", {"$ref": "other.json#/thing"}, {"$ref": "other.json#/thing"}), ("enum", [{"$ref": "#/$defs/missing"}], {"$ref": "#/$defs/missing"}), ("examples", [{"$ref": "#anchor"}], {"$ref": "kept"}), ], ) def test_dynamic_schema_ignores_reference_shaped_instance_data( self, keyword: str, value: t.Any, accepted: t.Dict[str, t.Any], ) -> None: """`$ref`-shaped payloads are data, not references, and must not raise.""" schema = { "type": "object", "patternProperties": {"^value_": {"type": "object", keyword: value}}, } model = json_schema_to_model(schema) assert model.model_validate({"value_a": accepted}).model_dump() == { "value_a": accepted } @pytest.mark.parametrize( "dynamic_schema", [ {"type": "array", "items": {"$ref": "https://example.com/schema.json"}}, {"allOf": [{"$ref": "https://example.com/schema.json"}]}, { "type": "object", "properties": {"inner": {"$ref": "https://example.com/schema.json"}}, }, { "type": "object", "additionalProperties": { "$ref": "https://example.com/schema.json", }, }, ], ) def test_dynamic_schema_checks_references_in_nested_schema_positions( self, dynamic_schema: t.Dict[str, t.Any], ) -> None: schema = { "type": "object", "patternProperties": {"^value_": dynamic_schema}, } with pytest.raises(ValueError, match="must be a local JSON Pointer"): json_schema_to_model(schema)