ruoyunbai/compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mvp

Browse Source
This commit is contained in:
gongwenxin
2025-05-28 15:55:46 +08:00
parent f0cc525141
commit 936714242f
313 changed files with 345685 additions and 7847 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__pycache__/test_test_case_registry.cpython-312.pyc
BIN
View File
Binary file not shown.
tests/__pycache__/test_test_orchestrator.cpython-312.pyc
BIN
View File
Binary file not shown.
tests/test_test_case_registry.py
-325
View File
@@ -1,325 +0,0 @@
import unittest
import os
import shutil
import logging
from typing import List
# 调整导入路径以适应测试文件在 tests/ 目录下的情况
# 我们假设 tests/ 和 ddms_compliance_suite/ 在同一级别 (项目根目录下)
import sys
# 获取当前文件 (test_test_case_registry.py) 的目录 (tests/)
current_file_dir = os.path.dirname(os.path.abspath(__file__))
# 获取项目根目录 (tests/ 的上一级)
project_root = os.path.dirname(current_file_dir)
# 将项目根目录添加到 sys.path 中，以便可以找到 ddms_compliance_suite 包
if project_root not in sys.path:
sys.path.insert(0, project_root)
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
from ddms_compliance_suite.test_case_registry import TestCaseRegistry
# 为了测试，我们需要一个临时的测试用例目录
TEMP_TEST_CASES_DIR = os.path.join(current_file_dir, "temp_custom_testcases_for_registry_test")
# 禁用 TestCaseRegistry 和 BaseAPITestCase 在测试期间的 INFO 和 DEBUG 日志，除非特意捕获
# logging.getLogger("ddms_compliance_suite.test_case_registry").setLevel(logging.WARNING)
# logging.getLogger("testcase").setLevel(logging.WARNING) # BaseAPITestCase uses testcase.<id>
class TestTestCaseRegistry(unittest.TestCase):
def setUp(self):
"""在每个测试方法运行前创建临时测试用例目录。"""
if os.path.exists(TEMP_TEST_CASES_DIR):
shutil.rmtree(TEMP_TEST_CASES_DIR)
os.makedirs(TEMP_TEST_CASES_DIR)
self.registry = None # 确保每个测试都重新初始化registry
def tearDown(self):
"""在每个测试方法运行后清理临时测试用例目录。"""
if os.path.exists(TEMP_TEST_CASES_DIR):
shutil.rmtree(TEMP_TEST_CASES_DIR)
def _create_test_case_file(self, filename: str, content: str):
"""辅助方法，在临时目录中创建测试用例文件。"""
with open(os.path.join(TEMP_TEST_CASES_DIR, filename), "w", encoding="utf-8") as f:
f.write(content)
def test_init_with_non_existent_dir(self):
"""测试使用不存在的目录初始化 TestCaseRegistry。"""
non_existent_dir = os.path.join(TEMP_TEST_CASES_DIR, "_i_do_not_exist_")
with self.assertLogs(level='WARNING') as log_watcher:
registry = TestCaseRegistry(test_cases_dir=non_existent_dir)
self.assertTrue(any(f"测试用例目录不存在或不是一个目录: {non_existent_dir}" in msg for msg in log_watcher.output))
self.assertEqual(len(registry.get_all_test_case_classes()), 0)
def test_init_with_empty_dir(self):
"""测试使用空的目录初始化 TestCaseRegistry。"""
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
self.assertEqual(len(registry.get_all_test_case_classes()), 0)
# 应该有一条INFO日志表明发现完成且数量为0
def test_discover_single_valid_test_case(self):
"""测试发现单个有效的测试用例。"""
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class MyTest(BaseAPITestCase):
id = "TC-001"
name = "Test Case 1"
description = "Desc 1"
severity = TestSeverity.HIGH
tags = ["tag1"]
"""
self._create_test_case_file("test_001.py", content)
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
all_cases = registry.get_all_test_case_classes()
self.assertEqual(len(all_cases), 1)
self.assertEqual(all_cases[0].id, "TC-001")
self.assertIsNotNone(registry.get_test_case_by_id("TC-001"))
def test_discover_multiple_test_cases_in_one_file(self):
"""测试在单个文件中发现多个测试用例。"""
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class TestA(BaseAPITestCase):
id = "TC-A"
name = "A"
class TestB(BaseAPITestCase):
id = "TC-B"
name = "B"
"""
self._create_test_case_file("test_ab.py", content)
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
all_cases = registry.get_all_test_case_classes()
self.assertEqual(len(all_cases), 2)
self.assertIsNotNone(registry.get_test_case_by_id("TC-A"))
self.assertIsNotNone(registry.get_test_case_by_id("TC-B"))
# 确保顺序与文件中定义的顺序（或至少是可预测的）一致，inspect.getmembers 通常按字母顺序
ids = sorted([case.id for case in all_cases])
self.assertEqual(ids, ["TC-A", "TC-B"])
def test_discover_test_cases_in_multiple_files(self):
"""测试在多个文件中发现测试用例。"""
content1 = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class Test1(BaseAPITestCase):
id = "TC-1"
"""
content2 = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class Test2(BaseAPITestCase):
id = "TC-2"
"""
self._create_test_case_file("file1.py", content1)
self._create_test_case_file("file2.py", content2)
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
all_cases = registry.get_all_test_case_classes()
self.assertEqual(len(all_cases), 2)
self.assertIsNotNone(registry.get_test_case_by_id("TC-1"))
self.assertIsNotNone(registry.get_test_case_by_id("TC-2"))
def test_ignore_non_py_files_and_dunder_files(self):
"""测试忽略非.py文件和以__开头的文件。"""
self._create_test_case_file("not_a_test.txt", "text content")
self._create_test_case_file("__init__.py", "# I am an init file")
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class RealTest(BaseAPITestCase):
id = "TC-REAL"
"""
self._create_test_case_file("real_test.py", content)
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
all_cases = registry.get_all_test_case_classes()
self.assertEqual(len(all_cases), 1)
self.assertEqual(all_cases[0].id, "TC-REAL")
def test_handle_import_error_in_test_file(self):
"""测试处理测试用例文件中的导入错误。"""
content = "import non_existent_module\n" # This will cause ImportError
self._create_test_case_file("importerror_test.py", content)
with self.assertLogs(level='ERROR') as log_watcher:
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
self.assertTrue(any("导入模块 'importerror_test' 从" in msg and "失败" in msg for msg in log_watcher.output))
self.assertEqual(len(registry.get_all_test_case_classes()), 0)
def test_handle_attribute_error_missing_id(self):
"""测试处理测试用例类缺少 'id' 属性的情况。"""
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class MissingIdTest(BaseAPITestCase):
name = "Missing ID"
# id is missing
"""
self._create_test_case_file("missing_id.py", content)
# AttributeError is caught by the generic Exception in discover_test_cases if not directly handled
# It might also depend on when/how inspect.getmembers tries to access obj.id
# Forcing access here to ensure the test scenario is valid if discovery itself doesn't raise it immediately
with self.assertLogs(level='ERROR') as log_watcher:
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
# The error log in discover_test_cases for AttributeError on obj.id might be tricky to assert precisely
# We'll check that no test cases were loaded from this problematic file, but other files might load.
self.assertTrue(any("在模块 'missing_id'" in msg and "查找测试用例时出错" in msg for msg in log_watcher.output),
msg=f"Did not find expected error log. Logs: {log_watcher.output}")
# Ensure no test cases are registered if the only file has this error
self.assertEqual(len(registry.get_all_test_case_classes()), 0)
def test_duplicate_test_case_id(self):
"""测试发现重复的测试用例 ID。"""
content1 = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class TestOne(BaseAPITestCase):
id = "DUPLICATE-ID-001"
name = "First with ID"
"""
content2 = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class TestTwo(BaseAPITestCase):
id = "DUPLICATE-ID-001" # Same ID
name = "Second with ID"
"""
self._create_test_case_file("file_one.py", content1)
self._create_test_case_file("file_two.py", content2)
with self.assertLogs(level='WARNING') as log_watcher:
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
self.assertTrue(any("发现重复的测试用例 ID: 'DUPLICATE-ID-001'" in msg for msg in log_watcher.output))
all_cases = registry.get_all_test_case_classes()
# inspect.getmembers order is not guaranteed across files, so we can't be sure which one is kept.
# However, the _registry (by ID) should have only one entry.
self.assertEqual(len(registry._registry), 1)
# The _test_case_classes list might have two if they are distinct class objects, but one overrides in _registry
# Depending on load order, one will overwrite the other in _registry. Let's check the final one.
registered_case = registry.get_test_case_by_id("DUPLICATE-ID-001")
self.assertIsNotNone(registered_case)
# We cannot reliably assert which name ('First with ID' or 'Second with ID') is kept due to file load order.
# Check that _test_case_classes might have more if classes are distinct but _registry has one.
# If the class objects are truly distinct, len(all_cases) could be 2. The important part is that by ID, only one is retrievable.
# A more robust check for `_test_case_classes` would be to ensure it contains the class that `_registry` points to.
self.assertIn(registered_case, all_cases)
# If the goal is that _test_case_classes should also be unique by some criteria after discovery, that logic would need adjustment.
# For now, get_all_test_case_classes returns all *discovered* classes that are BaseAPITestCase subclasses.
# And get_test_case_by_id returns the one that won the ID race.
# A typical use case iterates get_all_test_case_classes() for filtering, so this list should ideally be clean or documented.
# For now, we accept it might contain classes whose IDs were superseded if they are distinct objects.
def test_get_applicable_test_cases_no_restrictions(self):
"""测试 get_applicable_test_cases，当测试用例没有适用性限制时。"""
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class NoRestrictionTest(BaseAPITestCase):
id = "TC-NR-001"
name = "No Restrictions"
"""
self._create_test_case_file("no_restriction.py", content)
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
applicable = registry.get_applicable_test_cases("GET", "/api/items")
self.assertEqual(len(applicable), 1)
self.assertEqual(applicable[0].id, "TC-NR-001")
def test_get_applicable_by_method(self):
"""测试根据 applicable_methods 进行筛选。"""
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class GetOnlyTest(BaseAPITestCase):
id = "TC-GET-ONLY"
name = "GET Only"
applicable_methods = ["GET", "HEAD"]
class PostOnlyTest(BaseAPITestCase):
id = "TC-POST-ONLY"
name = "POST Only"
applicable_methods = ["POST"]
"""
self._create_test_case_file("method_tests.py", content)
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
applicable_get = registry.get_applicable_test_cases("GET", "/api/data")
self.assertEqual(len(applicable_get), 1)
self.assertEqual(applicable_get[0].id, "TC-GET-ONLY")
applicable_post = registry.get_applicable_test_cases("POST", "/api/data")
self.assertEqual(len(applicable_post), 1)
self.assertEqual(applicable_post[0].id, "TC-POST-ONLY")
applicable_put = registry.get_applicable_test_cases("PUT", "/api/data")
self.assertEqual(len(applicable_put), 0)
applicable_head = registry.get_applicable_test_cases("HEAD", "/api/data") # Case sensitive check for methods in list
self.assertEqual(len(applicable_head), 1)
self.assertEqual(applicable_head[0].id, "TC-GET-ONLY")
def test_get_applicable_by_path_regex(self):
"""测试根据 applicable_paths_regex 进行筛选。"""
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class UserPathTest(BaseAPITestCase):
id = "TC-USER-PATH"
name = "User Path"
applicable_paths_regex = r"^/api/users/\\d+$" # Matches /api/users/<number>
class OrderPathTest(BaseAPITestCase):
id = "TC-ORDER-PATH"
name = "Order Path"
applicable_paths_regex = r"^/api/orders"
"""
self._create_test_case_file("path_tests.py", content)
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
applicable_user = registry.get_applicable_test_cases("GET", "/api/users/123")
self.assertEqual(len(applicable_user), 1)
self.assertEqual(applicable_user[0].id, "TC-USER-PATH")
applicable_order = registry.get_applicable_test_cases("POST", "/api/orders/new")
self.assertEqual(len(applicable_order), 1)
self.assertEqual(applicable_order[0].id, "TC-ORDER-PATH")
applicable_none1 = registry.get_applicable_test_cases("GET", "/api/products/789")
self.assertEqual(len(applicable_none1), 0)
applicable_none2 = registry.get_applicable_test_cases("GET", "/api/users/profile") # Does not match \d+
self.assertEqual(len(applicable_none2), 0)
def test_get_applicable_by_method_and_path(self):
"""测试同时根据方法和路径进行筛选。"""
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class SpecificGetTest(BaseAPITestCase):
id = "TC-SPECIFIC-GET"
name = "Specific GET"
applicable_methods = ["GET"]
applicable_paths_regex = r"^/data/\\w+$"
"""
self._create_test_case_file("specific_get.py", content)
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
applicable = registry.get_applicable_test_cases("GET", "/data/item1")
self.assertEqual(len(applicable), 1)
self.assertEqual(applicable[0].id, "TC-SPECIFIC-GET")
not_applicable_method = registry.get_applicable_test_cases("POST", "/data/item1")
self.assertEqual(len(not_applicable_method), 0)
not_applicable_path = registry.get_applicable_test_cases("GET", "/data/item1/details")
self.assertEqual(len(not_applicable_path), 0)
def test_invalid_path_regex_handling(self):
"""测试处理无效的路径正则表达式。"""
content = """
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity
class InvalidRegexTest(BaseAPITestCase):
id = "TC-INVALID-REGEX"
name = "Invalid Regex"
applicable_paths_regex = r"^/api/path_(" # Unbalanced parenthesis
"""
self._create_test_case_file("invalid_regex.py", content)
with self.assertLogs(level='ERROR') as log_watcher:
registry = TestCaseRegistry(test_cases_dir=TEMP_TEST_CASES_DIR)
self.assertTrue(any("中的路径正则表达式 'invalid_regex.InvalidRegexTest' 无效" in msg for msg in log_watcher.output))
# The test case with invalid regex should not match any path
applicable = registry.get_applicable_test_cases("GET", "/api/path_something")
self.assertEqual(len(applicable), 0)
if __name__ == '__main__':
# Configure logging for detailed output when running directly
# logging.basicConfig(stream=sys.stdout, level=logging.DEBUG,
# format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
unittest.main()
tests/test_test_orchestrator.py
-438
View File
@@ -1,438 +0,0 @@
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()