|
|
|
|
|
|
|
|
from typing import Any |
|
|
|
|
|
import pytest |
|
|
from langflow.helpers.base_model import build_model_from_schema |
|
|
from pydantic import BaseModel |
|
|
from pydantic_core import PydanticUndefined |
|
|
|
|
|
|
|
|
class TestBuildModelFromSchema: |
|
|
|
|
|
def test_create_model_from_valid_schema(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value", "description": "A string field"}, |
|
|
{"name": "field2", "type": "int", "default": 0, "description": "An integer field"}, |
|
|
{"name": "field3", "type": "bool", "default": False, "description": "A boolean field"}, |
|
|
] |
|
|
model = build_model_from_schema(schema) |
|
|
instance = model(field1="test", field2=123, field3=True) |
|
|
assert instance.field1 == "test" |
|
|
assert instance.field2 == 123 |
|
|
assert instance.field3 is True |
|
|
|
|
|
|
|
|
def test_handle_empty_schema(self): |
|
|
schema = [] |
|
|
model = build_model_from_schema(schema) |
|
|
instance = model() |
|
|
assert instance is not None |
|
|
|
|
|
|
|
|
def test_handles_multiple_fields_fixed_with_instance_check(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value1"}, |
|
|
{"name": "field2", "type": "int", "default": 42}, |
|
|
{"name": "field3", "type": "list", "default": [1, 2, 3]}, |
|
|
{"name": "field4", "type": "dict", "default": {"key": "value"}}, |
|
|
] |
|
|
|
|
|
model = build_model_from_schema(schema) |
|
|
model_instance = model(field1="test", field2=123, field3=[1, 2, 3], field4={"key": "value"}) |
|
|
|
|
|
assert issubclass(model, BaseModel) |
|
|
assert hasattr(model_instance, "field1") |
|
|
assert hasattr(model_instance, "field2") |
|
|
assert hasattr(model_instance, "field3") |
|
|
assert hasattr(model_instance, "field4") |
|
|
|
|
|
|
|
|
def test_correctly_accesses_descriptions_recommended_fix(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value1", "description": "Description for field1"}, |
|
|
{"name": "field2", "type": "int", "default": 42, "description": "Description for field2"}, |
|
|
{"name": "field3", "type": "list", "default": [1, 2, 3], "description": "Description for field3"}, |
|
|
{"name": "field4", "type": "dict", "default": {"key": "value"}, "description": "Description for field4"}, |
|
|
] |
|
|
|
|
|
model = build_model_from_schema(schema) |
|
|
|
|
|
assert model.model_fields["field1"].description == "Description for field1" |
|
|
assert model.model_fields["field2"].description == "Description for field2" |
|
|
assert model.model_fields["field3"].description == "Description for field3" |
|
|
assert model.model_fields["field4"].description == "Description for field4" |
|
|
|
|
|
|
|
|
def test_supports_single_and_multiple_type_annotations(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value1", "description": "Description 1"}, |
|
|
{"name": "field2", "type": "list", "default": [1, 2, 3], "description": "Description 2", "multiple": True}, |
|
|
{"name": "field3", "type": "int", "default": 100, "description": "Description 3"}, |
|
|
] |
|
|
model_type = build_model_from_schema(schema) |
|
|
assert issubclass(model_type, BaseModel) |
|
|
|
|
|
|
|
|
def test_manages_unknown_field_types(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value1"}, |
|
|
{"name": "field2", "type": "unknown_type", "default": "default_value2"}, |
|
|
] |
|
|
with pytest.raises(ValueError, match="Invalid type: unknown_type"): |
|
|
build_model_from_schema(schema) |
|
|
|
|
|
|
|
|
def test_raises_error_for_invalid_input_different_exception_with_specific_exception(self): |
|
|
schema = [{"name": "field1", "type": "invalid_type", "default": "default_value"}] |
|
|
with pytest.raises(ValueError, match="Invalid type: invalid_type"): |
|
|
build_model_from_schema(schema) |
|
|
|
|
|
|
|
|
def test_process_schema_missing_optional_keys_updated(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value1"}, |
|
|
{"name": "field2", "type": "int", "default": 0, "description": "Field 2 description"}, |
|
|
{"name": "field3", "type": "list", "default": [], "multiple": True}, |
|
|
{"name": "field4", "type": "dict", "default": {}, "description": "Field 4 description", "multiple": True}, |
|
|
] |
|
|
result_model = build_model_from_schema(schema) |
|
|
assert result_model.__annotations__["field1"] == str |
|
|
assert result_model.model_fields["field1"].description == "" |
|
|
assert result_model.__annotations__["field2"] == int |
|
|
assert result_model.model_fields["field2"].description == "Field 2 description" |
|
|
assert result_model.__annotations__["field3"] == list[list[Any]] |
|
|
assert result_model.model_fields["field3"].description == "" |
|
|
assert result_model.__annotations__["field4"] == list[dict[str, Any]] |
|
|
assert result_model.model_fields["field4"].description == "Field 4 description" |
|
|
|
|
|
|
|
|
def test_schema_fields_with_none_default(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": None, "description": "Field 1 description"}, |
|
|
{"name": "field2", "type": "int", "default": None, "description": "Field 2 description"}, |
|
|
{"name": "field3", "type": "list", "default": None, "description": "Field 3 description", "multiple": True}, |
|
|
] |
|
|
model = build_model_from_schema(schema) |
|
|
assert model.model_fields["field1"].default == PydanticUndefined |
|
|
assert model.model_fields["field2"].default == PydanticUndefined |
|
|
assert model.model_fields["field3"].default == PydanticUndefined |
|
|
|
|
|
|
|
|
def test_nested_list_and_dict_types_handling(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "list", "default": [], "description": "list field", "multiple": True}, |
|
|
{"name": "field2", "type": "dict", "default": {}, "description": "Dict field"}, |
|
|
] |
|
|
model_type = build_model_from_schema(schema) |
|
|
assert issubclass(model_type, BaseModel) |
|
|
|
|
|
|
|
|
def test_handle_large_schemas_efficiently(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value1", "description": "Description 1"}, |
|
|
{"name": "field2", "type": "int", "default": 100, "description": "Description 2"}, |
|
|
{"name": "field3", "type": "list", "default": [1, 2, 3], "description": "Description 3", "multiple": True}, |
|
|
{"name": "field4", "type": "dict", "default": {"key": "value"}, "description": "Description 4"}, |
|
|
] |
|
|
model_type = build_model_from_schema(schema) |
|
|
assert issubclass(model_type, BaseModel) |
|
|
|
|
|
|
|
|
def test_returns_valid_model_class(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value1", "description": "Description for field1"}, |
|
|
{"name": "field2", "type": "int", "default": 42, "description": "Description for field2", "multiple": True}, |
|
|
] |
|
|
model_class = build_model_from_schema(schema) |
|
|
assert issubclass(model_class, BaseModel) |
|
|
|
|
|
|
|
|
def test_no_duplicate_field_names_fixed_fixed(self): |
|
|
schema = [ |
|
|
{"name": "field1", "type": "str", "default": "default_value1"}, |
|
|
{"name": "field2", "type": "int", "default": 0}, |
|
|
{"name": "field1", "type": "float", "default": 0.0}, |
|
|
] |
|
|
model = build_model_from_schema(schema) |
|
|
assert model.__annotations__["field1"] == float |
|
|
assert model.__annotations__["field2"] == int |
|
|
|