llm

tests/test_test_orchestrator.py

@@ -0,0 +1,438 @@
+import unittest
+import logging
+from typing import Optional, List, Dict, Any, Type, Union
+from uuid import UUID
+import datetime as dt
+
+# 调整导入路径以适应测试文件在 tests/ 目录下的情况
+import sys
+import os
+current_file_dir = os.path.dirname(os.path.abspath(__file__))
+project_root = os.path.dirname(current_file_dir)
+if project_root not in sys.path:
+    sys.path.insert(0, project_root)
+
+from pydantic import BaseModel, Field, ValidationError
+from pydantic.networks import EmailStr
+
+from ddms_compliance_suite.test_orchestrator import APITestOrchestrator, _dynamic_model_cache
+from ddms_compliance_suite.llm_utils.llm_service import LLMService # For orchestrator init if needed
+
+# 基本的 Orchestrator 初始化参数，如果测试中需要实例化 Orchestrator
+BASE_URL_FOR_TEST = "http://fakeapi.com"
+
+# 全局禁用或设置较低级别的日志，以便测试输出更干净
+# logging.basicConfig(level=logging.ERROR)
+# logging.getLogger(\"ddms_compliance_suite.test_orchestrator\").setLevel(logging.WARNING)
+
+# Helper functions to extract constraint values from FieldInfo.metadata
+def get_metadata_constraint_value(metadata_list: list, constraint_attr_name: str) -> Any:
+    for m_obj in metadata_list:
+        if hasattr(m_obj, constraint_attr_name):
+            return getattr(m_obj, constraint_attr_name)
+    return None
+
+class TestDynamicModelCreation(unittest.TestCase):
+    """
+    专门测试 APITestOrchestrator._create_pydantic_model_from_schema 方法。
+    """
+
+    def setUp(self):
+        """清除动态模型缓存，确保每个测试的独立性。"""
+        _dynamic_model_cache.clear()
+        # 创建一个Orchestrator实例，_create_pydantic_model_from_schema是它的方法
+        # 对于仅测试 _create_pydantic_model_from_schema，LLM配置可以为None
+        self.orchestrator = APITestOrchestrator(base_url=BASE_URL_FOR_TEST)
+        # 可以通过 self.orchestrator._create_pydantic_model_from_schema 调用
+
+    def tearDown(self):
+        """再次清除缓存，以防万一。"""
+        _dynamic_model_cache.clear()
+
+    def test_simple_object(self):
+        """测试基本对象创建，包含不同类型的字段和必需字段。"""
+        schema = {
+            "type": "object",
+            "properties": {
+                "name": {"type": "string", "description": "User name"},
+                "age": {"type": "integer", "minimum": 0},
+                "email": {"type": "string", "format": "email"},
+                "is_active": {"type": "boolean", "default": True},
+                "height": {"type": "number"}
+            },
+            "required": ["name", "age"]
+        }
+        model_name = "SimpleUserModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        self.assertTrue(issubclass(DynamicModel, BaseModel))
+        self.assertEqual(DynamicModel.__name__, model_name)
+
+        fields = DynamicModel.model_fields
+        self.assertIn("name", fields)
+        self.assertEqual(fields["name"].annotation, str)
+        self.assertTrue(fields["name"].is_required())
+        self.assertEqual(fields["name"].description, "User name")
+
+        self.assertIn("age", fields)
+        age_field_info = fields["age"]
+        self.assertEqual(age_field_info.annotation, int)
+        self.assertTrue(age_field_info.is_required())
+        self.assertEqual(get_metadata_constraint_value(age_field_info.metadata, 'ge'), 0)
+
+        self.assertIn("email", fields)
+        self.assertEqual(fields["email"].annotation, Optional[EmailStr]) # Not required, so Optional
+        self.assertFalse(fields["email"].is_required())
+
+        self.assertIn("is_active", fields)
+        self.assertEqual(fields["is_active"].annotation, bool) # Corrected: Has default, so it's bool
+        self.assertEqual(fields["is_active"].default, True)
+        self.assertFalse(fields["is_active"].is_required()) # Fields with defaults are not strictly required from user input
+
+        self.assertIn("height", fields)
+        self.assertEqual(fields["height"].annotation, Optional[float]) # Not required
+
+        # 测试实例化和验证
+        valid_data = {"name": "Test", "age": 30, "email": "test@example.com", "height": 1.75}
+        instance = DynamicModel(**valid_data)
+        self.assertEqual(instance.name, "Test")
+        self.assertEqual(instance.is_active, True) # Default value
+
+        with self.assertRaises(ValidationError):
+            DynamicModel(age=-5, email="bademail") # name missing, age invalid
+
+    def test_nested_object(self):
+        """测试嵌套对象的创建。"""
+        schema = {
+            "type": "object",
+            "properties": {
+                "id": {"type": "string"},
+                "profile": {
+                    "type": "object",
+                    "properties": {
+                        "user_email": {"type": "string", "format": "email"},
+                        "score": {"type": "integer", "default": 0}
+                    },
+                    "required": ["user_email"]
+                }
+            },
+            "required": ["id"]
+        }
+        model_name = "NestedOuterModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        fields = DynamicModel.model_fields
+        self.assertIn("profile", fields)
+        
+        ProfileModel = fields["profile"].annotation
+        self.assertTrue(hasattr(ProfileModel, '__origin__') and ProfileModel.__origin__ is Union)
+        self.assertIn(type(None), ProfileModel.__args__)
+        NestedProfileModel = [arg for arg in ProfileModel.__args__ if arg is not type(None)][0]
+
+        self.assertTrue(issubclass(NestedProfileModel, BaseModel))
+        self.assertEqual(NestedProfileModel.__name__, f"{model_name}_profile")
+
+        nested_fields = NestedProfileModel.model_fields
+        self.assertIn("user_email", nested_fields)
+        self.assertEqual(nested_fields["user_email"].annotation, EmailStr)
+        self.assertTrue(nested_fields["user_email"].is_required())
+
+        self.assertIn("score", nested_fields)
+        self.assertEqual(nested_fields["score"].annotation, int)
+        self.assertEqual(nested_fields["score"].default, 0)
+        
+        # Test instantiation
+        valid_data = {"id": "abc", "profile": {"user_email": "nested@example.com"}}
+        instance = DynamicModel(**valid_data)
+        self.assertEqual(instance.id, "abc")
+        self.assertEqual(instance.profile.user_email, "nested@example.com")
+        self.assertEqual(instance.profile.score, 0)
+
+    def test_array_of_simple_types(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "tags": {"type": "array", "items": {"type": "string"}},
+                "scores": {"type": "array", "items": {"type": "integer"}, "default": []}
+            }
+        }
+        model_name = "ArraySimpleModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        fields = DynamicModel.model_fields
+
+        self.assertIn("tags", fields)
+        self.assertEqual(fields["tags"].annotation, Optional[List[str]])
+
+        self.assertIn("scores", fields)
+        self.assertEqual(fields["scores"].annotation, List[int])
+        self.assertEqual(fields["scores"].default, [])
+
+        valid_data = {"tags": ["a", "b"], "scores": [1,2,3]}
+        instance = DynamicModel(**valid_data)
+        self.assertEqual(instance.tags, ["a", "b"])
+
+        # Test default for scores when tags is provided
+        instance2 = DynamicModel(tags=["c"])
+        self.assertEqual(instance2.scores, [])
+
+
+    def test_array_of_objects(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "users": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "username": {"type": "string"},
+                            "user_id": {"type": "integer"}
+                        },
+                        "required": ["username"]
+                    }
+                }
+            }
+        }
+        model_name = "ArrayObjectModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        fields = DynamicModel.model_fields
+        self.assertIn("users", fields)
+        
+        # users is Optional[List[UserModel_users_Item]]
+        UserListItemType = fields["users"].annotation
+        self.assertTrue(hasattr(UserListItemType, '__origin__') and UserListItemType.__origin__ is Union)
+        UserListType = [arg for arg in UserListItemType.__args__ if arg is not type(None)][0]
+
+        self.assertEqual(UserListType.__origin__, list) # Check it's a List
+        ItemModel = UserListType.__args__[0] # Get the item type from List[ItemType]
+        
+        self.assertTrue(issubclass(ItemModel, BaseModel))
+        self.assertEqual(ItemModel.__name__, f"{model_name}_users_Item")
+
+        item_fields = ItemModel.model_fields
+        self.assertEqual(item_fields["username"].annotation, str)
+        self.assertTrue(item_fields["username"].is_required())
+        self.assertEqual(item_fields["user_id"].annotation, Optional[int])
+
+        valid_data = {"users": [{"username": "a", "user_id":1}, {"username": "b"}]}
+        instance = DynamicModel(**valid_data)
+        self.assertEqual(len(instance.users), 2)
+        self.assertEqual(instance.users[0].username, "a")
+
+
+    def test_field_constraints(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "quantity": {"type": "integer", "minimum": 1, "maximum": 100},
+                "code": {"type": "string", "minLength": 3, "maxLength": 5, "pattern": "^[A-Z]+$"},
+                "percentage": {"type": "number", "minimum": 0.0, "maximum": 1.0}
+            }
+        }
+        model_name = "ConstraintsModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        fields = DynamicModel.model_fields
+
+        # Quantity (int)
+        quantity_field_info = fields["quantity"]
+        self.assertEqual(get_metadata_constraint_value(quantity_field_info.metadata, 'ge'), 1)
+        self.assertEqual(get_metadata_constraint_value(quantity_field_info.metadata, 'le'), 100)
+        
+        # Code (str)
+        code_field_info = fields["code"]
+        self.assertEqual(get_metadata_constraint_value(code_field_info.metadata, 'min_length'), 3)
+        self.assertEqual(get_metadata_constraint_value(code_field_info.metadata, 'max_length'), 5)
+        self.assertEqual(get_metadata_constraint_value(code_field_info.metadata, 'pattern'), "^[A-Z]+$")
+
+        # Percentage (float/number)
+        percentage_field_info = fields["percentage"]
+        self.assertEqual(get_metadata_constraint_value(percentage_field_info.metadata, 'ge'), 0.0)
+        self.assertEqual(get_metadata_constraint_value(percentage_field_info.metadata, 'le'), 1.0)
+
+        # Test validation
+        with self.assertRaises(ValidationError): DynamicModel(quantity=0)
+        with self.assertRaises(ValidationError): DynamicModel(code="ab")
+        with self.assertRaises(ValidationError): DynamicModel(code="ABCDEF")
+        with self.assertRaises(ValidationError): DynamicModel(code="ab1")
+        with self.assertRaises(ValidationError): DynamicModel(percentage=1.1)
+        DynamicModel(quantity=50, code="XYZ", percentage=0.5) # Should be valid
+
+
+    def test_enum_in_description(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "status": {"type": "string", "enum": ["active", "inactive", "pending"], "description": "Current status."}
+            }
+        }
+        model_name = "EnumDescModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        fields = DynamicModel.model_fields
+        self.assertIn("status", fields)
+        self.assertIn("Enum values: active, inactive, pending", fields["status"].description)
+        self.assertIn("Current status.", fields["status"].description)
+
+
+    def test_datetime_formats(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "created_at": {"type": "string", "format": "date-time"},
+                "event_date": {"type": "string", "format": "date"},
+                "uid": {"type": "string", "format": "uuid"}
+            }
+        }
+        model_name = "DateTimeUUIDModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        fields = DynamicModel.model_fields
+
+        self.assertEqual(fields["created_at"].annotation, Optional[dt.datetime])
+        self.assertEqual(fields["event_date"].annotation, Optional[dt.date])
+        self.assertEqual(fields["uid"].annotation, Optional[UUID])
+
+        valid_data = {
+            "created_at": "2024-01-15T10:30:00Z",
+            "event_date": "2024-01-15",
+            "uid": "a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"
+        }
+        instance = DynamicModel(**valid_data)
+        self.assertIsInstance(instance.created_at, dt.datetime)
+        self.assertIsInstance(instance.event_date, dt.date)
+        self.assertIsInstance(instance.uid, UUID)
+
+
+    def test_empty_object_schema(self):
+        schema = {"type": "object", "properties": {}} # Empty properties
+        model_name = "EmptyPropertiesModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        self.assertEqual(len(DynamicModel.model_fields), 0)
+        DynamicModel() # Should instantiate
+
+        schema2 = {"type": "object"} # No properties field at all
+        model_name2 = "NoPropertiesFieldModel"
+        DynamicModel2 = self.orchestrator._create_pydantic_model_from_schema(schema2, model_name2)
+        self.assertIsNotNone(DynamicModel2)
+        self.assertEqual(len(DynamicModel2.model_fields), 0)
+        DynamicModel2()
+
+    def test_invalid_top_level_schema(self):
+        schema = {"type": "string"} # Not an object
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, "InvalidSchemaModel")
+        self.assertIsNone(DynamicModel)
+
+        schema2 = [{"type": "object"}] # Not a dict
+        DynamicModel2 = self.orchestrator._create_pydantic_model_from_schema(schema2, "InvalidSchemaModel2")
+        self.assertIsNone(DynamicModel2)
+
+
+    def test_model_caching(self):
+        schema = {"type": "object", "properties": {"name": {"type": "string"}}}
+        model_name = "CachedModel"
+        
+        Model1 = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(Model1)
+        self.assertIn(model_name, _dynamic_model_cache)
+        
+        Model2 = self.orchestrator._create_pydantic_model_from_schema(schema, model_name) # Should be from cache
+        self.assertIs(Model1, Model2) # Check they are the same object
+
+    def test_recursion_depth_limit(self):
+        # Construct a schema that would recurse indefinitely if not limited
+        # A: { "prop_b": B }, B: { "prop_a": A } - this is hard with current naming
+        # Easier: A: { "prop_a": A_prop_a }
+        # Let's try A: { "next": A }
+        # The _create_pydantic_model_from_schema method itself uses model_name + prop_name for nested models,
+        # so a direct self-reference in schema like {"type": "object", "properties": {"self": {"$ref": "#/"}}}
+        # is not fully handled yet and would rely on ForwardRef if schema was static.
+        # For dynamic creation, the depth limit is the main guard.
+        
+        # Create a schema that nests deeply
+        deep_schema: Dict[str, Any] = {"type": "object", "properties": {}}
+        current_level = deep_schema["properties"]
+        
+        # MAX_RECURSION_DEPTH in APITestOrchestrator is 10
+        # We create a schema of depth 11 (0 to 10 for properties)
+        # property name level_0 contains object with property level_1 etc.
+        for i in range(12): # Go a bit beyond the limit
+            current_level[f"level_{i}"] = {"type": "object", "properties": {}}
+            if i < 11: # Don't add properties to the very last one
+                 current_level = current_level[f"level_{i}"]["properties"]
+            
+        with self.assertLogs(level='ERROR') as log_watcher:
+            GeneratedModel = self.orchestrator._create_pydantic_model_from_schema(deep_schema, "DeepRecursiveModel")
+            self.assertTrue(any("达到最大递归深度" in msg for msg in log_watcher.output))
+            self.assertIsNotNone(GeneratedModel)
+
+    def test_name_sanitization(self):
+        schema = {"type": "object", "properties": {"test": {"type": "string"}}}
+        # Valid name
+        Model1 = self.orchestrator._create_pydantic_model_from_schema(schema, "ValidName123")
+        self.assertIsNotNone(Model1)
+        self.assertEqual(Model1.__name__, "ValidName123")
+
+        # Name with spaces and hyphens
+        Model2 = self.orchestrator._create_pydantic_model_from_schema(schema, "Invalid Name-Test")
+        self.assertIsNotNone(Model2)
+        self.assertEqual(Model2.__name__, "Invalid_Name_Test") # Check sanitized name
+
+        # Name starting with number
+        Model3 = self.orchestrator._create_pydantic_model_from_schema(schema, "123InvalidStart")
+        self.assertIsNotNone(Model3)
+        self.assertEqual(Model3.__name__, "DynamicModel_123InvalidStart")
+
+        # Empty name - should get a default prefix
+        Model4 = self.orchestrator._create_pydantic_model_from_schema(schema, "")
+        self.assertIsNotNone(Model4)
+        self.assertTrue(Model4.__name__.startswith("DynamicModel_"))
+
+        # Name that is just underscores
+        Model5 = self.orchestrator._create_pydantic_model_from_schema(schema, "___")
+        self.assertIsNotNone(Model5)
+        self.assertEqual(Model5.__name__, "___") # Underscores are valid but Pydantic might mangle if it's a dunder name. create_model seems to keep it.
+
+    def test_optional_logic_for_fields(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "required_field": {"type": "string"},
+                "optional_field_no_default": {"type": "integer"},
+                "optional_field_with_default": {"type": "boolean", "default": False},
+                "optional_nested_object": {
+                    "type": "object",
+                    "properties": {"value": {"type": "string"}}
+                }
+            },
+            "required": ["required_field"]
+        }
+        model_name = "OptionalFieldsModel"
+        DynamicModel = self.orchestrator._create_pydantic_model_from_schema(schema, model_name)
+        self.assertIsNotNone(DynamicModel)
+        fields = DynamicModel.model_fields
+
+        self.assertEqual(fields["required_field"].annotation, str)
+        self.assertTrue(fields["required_field"].is_required())
+
+        self.assertEqual(fields["optional_field_no_default"].annotation, Optional[int])
+        self.assertFalse(fields["optional_field_no_default"].is_required())
+        self.assertEqual(fields["optional_field_no_default"].default, None) # Pydantic default for Optional[T] is None
+
+        self.assertEqual(fields["optional_field_with_default"].annotation, bool)
+        self.assertFalse(fields["optional_field_with_default"].is_required())
+        self.assertEqual(fields["optional_field_with_default"].default, False)
+        
+        # optional_nested_object is not required
+        NestedType = fields["optional_nested_object"].annotation
+        self.assertTrue(hasattr(NestedType, '__origin__') and NestedType.__origin__ is Union)
+        self.assertIn(type(None), NestedType.__args__)
+        ActualNestedModel = [arg for arg in NestedType.__args__ if arg is not type(None)][0]
+        self.assertTrue(issubclass(ActualNestedModel, BaseModel))
+
+
+if __name__ == '__main__':
+    unittest.main()