import os import importlib.util import inspect import logging import re from typing import List, Type, Optional, Dict # 确保可以从 sibling 模块导入 from .test_framework_core import BaseAPITestCase class TestCaseRegistry: """ 负责发现、加载和管理所有自定义的APITestCase类。 """ def __init__(self, test_cases_dir: str): """ 初始化 TestCaseRegistry。 Args: test_cases_dir: 存放自定义测试用例 (.py 文件) 的目录路径。 """ self.test_cases_dir = test_cases_dir self.logger = logging.getLogger(__name__) self._registry: Dict[str, Type[BaseAPITestCase]] = {} self._test_case_classes: List[Type[BaseAPITestCase]] = [] self.discover_test_cases() def discover_test_cases(self): """ 扫描指定目录，动态导入模块，并注册所有继承自 BaseAPITestCase 的类。 """ if not os.path.isdir(self.test_cases_dir): self.logger.warning(f"测试用例目录不存在或不是一个目录: {self.test_cases_dir}") return self.logger.info(f"开始从目录 '{self.test_cases_dir}' 发现测试用例...") found_count = 0 for filename in os.listdir(self.test_cases_dir): if filename.endswith(".py") and not filename.startswith("__"): module_name = filename[:-3] file_path = os.path.join(self.test_cases_dir, filename) try: # 动态导入模块 spec = importlib.util.spec_from_file_location(module_name, file_path) if spec and spec.loader: module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) self.logger.debug(f"成功导入模块: {module_name} 从 {file_path}") # 在模块中查找 BaseAPITestCase 的子类 for name, obj in inspect.getmembers(module): if inspect.isclass(obj) and issubclass(obj, BaseAPITestCase) and obj is not BaseAPITestCase: if obj.id in self._registry: self.logger.warning(f"发现重复的测试用例 ID: '{obj.id}' (来自类 '{obj.__name__}' in {file_path})。之前的定义将被覆盖。") self._registry[obj.id] = obj if obj not in self._test_case_classes: # 避免重复添加同一个类对象 self._test_case_classes.append(obj) found_count += 1 self.logger.info(f"已注册测试用例: '{obj.id}' ({obj.name}) 来自类 '{obj.__name__}'") else: self.logger.error(f"无法为文件 '{file_path}' 创建模块规范。") except ImportError as e: self.logger.error(f"导入模块 '{module_name}' 从 '{file_path}' 失败: {e}", exc_info=True) except AttributeError as e: self.logger.error(f"在模块 '{module_name}' ({file_path}) 中查找测试用例时出错 (可能是缺少必要的元数据如 'id'): {e}", exc_info=True) except Exception as e: self.logger.error(f"处理文件 '{file_path}' 时发生未知错误: {e}", exc_info=True) # 根据 execution_order 对收集到的测试用例类进行排序 try: self._test_case_classes.sort(key=lambda tc_class: (getattr(tc_class, 'execution_order', 100), tc_class.__name__)) self.logger.info(f"已根据 execution_order (主要) 和类名 (次要) 对 {len(self._test_case_classes)} 个测试用例类进行了排序。") except AttributeError as e_sort: self.logger.error(f"对测试用例类进行排序时发生 AttributeError (可能部分类缺少 execution_order): {e_sort}", exc_info=True) except Exception as e_sort_general: self.logger.error(f"对测试用例类进行排序时发生未知错误: {e_sort_general}", exc_info=True) self.logger.info(f"测试用例发现完成。总共注册了 {len(self._registry)} 个独特的测试用例 (基于ID)。发现并排序了 {len(self._test_case_classes)} 个测试用例类。") def get_test_case_by_id(self, case_id: str) -> Optional[Type[BaseAPITestCase]]: """根据ID获取已注册的测试用例类。""" return self._registry.get(case_id) def get_all_test_case_classes(self) -> List[Type[BaseAPITestCase]]: """获取所有已注册的测试用例类列表。""" return list(self._test_case_classes) # 返回副本 def get_applicable_test_cases(self, endpoint_method: str, endpoint_path: str) -> List[Type[BaseAPITestCase]]: """ 根据API端点的方法和路径，筛选出适用的测试用例类。 Args: endpoint_method: API端点的方法 (例如 "GET", "POST")。 endpoint_path: API端点的路径 (例如 "/users/{id}")。 Returns: 一个包含适用测试用例类的列表。 """ applicable_cases: List[Type[BaseAPITestCase]] = [] for tc_class in self._test_case_classes: # 1. 检查 applicable_methods if tc_class.applicable_methods is not None: if endpoint_method.upper() not in [m.upper() for m in tc_class.applicable_methods]: self.logger.debug(f"测试用例 '{tc_class.id}' 不适用于方法 '{endpoint_method}' (期望: {tc_class.applicable_methods})，已跳过。") continue # 方法不匹配，跳过此测试用例 # 2. 检查 applicable_paths_regex if tc_class.applicable_paths_regex is not None: try: if not re.match(tc_class.applicable_paths_regex, endpoint_path): self.logger.debug(f"测试用例 '{tc_class.id}' 不适用于路径 '{endpoint_path}' (正则: '{tc_class.applicable_paths_regex}')，已跳过。") continue # 路径正则不匹配，跳过此测试用例 except re.error as e: self.logger.error(f"测试用例 '{tc_class.id}' 中的路径正则表达式 '{tc_class.applicable_paths_regex}' 无效: {e}。此测试用例将不匹配任何路径。") continue # 如果通过了所有检查，则认为适用 applicable_cases.append(tc_class) self.logger.debug(f"测试用例 '{tc_class.id}' 适用于端点 '{endpoint_method} {endpoint_path}'。") return applicable_cases # 示例用法 (用于测试此模块，实际使用时由编排器调用) if __name__ == '__main__': logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) # 假设项目结构如下: # your_project_root/ # ddms_compliance_suite/ # test_framework_core.py # test_case_registry.py # custom_testcases/ <-- 测试用例存放目录 # example_checks.py # another_set_of_checks.py # 创建一个临时的 custom_testcases 目录和一些示例测试用例文件用于测试 current_dir = os.path.dirname(os.path.abspath(__file__)) custom_testcases_path = os.path.join(os.path.dirname(current_dir), "custom_testcases") # 假设custom_testcases在ddms_compliance_suite的父目录 if not os.path.exists(custom_testcases_path): os.makedirs(custom_testcases_path) logger.info(f"创建临时目录: {custom_testcases_path}") # 示例测试用例文件1: example_checks.py example_checks_content = """ from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity class MyFirstTest(BaseAPITestCase): id = "TC-EXAMPLE-001" name = "我的第一个测试" description = "一个简单的示例测试。" severity = TestSeverity.INFO tags = ["example"] applicable_methods = ["GET"] class MySecondTest(BaseAPITestCase): id = "TC-EXAMPLE-002" name = "我的第二个测试" description = "另一个示例测试，适用于所有方法和特定路径。" severity = TestSeverity.MEDIUM tags = ["example", "path_specific"] applicable_paths_regex = r"/api/users/.+" """ with open(os.path.join(custom_testcases_path, "example_checks.py"), "w", encoding="utf-8") as f: f.write(example_checks_content) logger.info(f"创建示例测试文件: {os.path.join(custom_testcases_path, 'example_checks.py')}") # 示例测试用例文件2: specific_feature_tests.py (无适用性限制) specific_tests_content = """ from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity class FeatureXCheck(BaseAPITestCase): id = "TC-FEATUREX-001" name = "特性X的检查" description = "验证特性X的相关功能。" severity = TestSeverity.HIGH tags = ["feature-x"] """ with open(os.path.join(custom_testcases_path, "specific_feature_tests.py"), "w", encoding="utf-8") as f: f.write(specific_tests_content) logger.info(f"创建示例测试文件: {os.path.join(custom_testcases_path, 'specific_feature_tests.py')}") # 测试 TestCaseRegistry registry = TestCaseRegistry(test_cases_dir=custom_testcases_path) logger.info("\n--- 所有已注册的测试用例类 ---") all_cases = registry.get_all_test_case_classes() for tc_class in all_cases: logger.info(f" ID: {tc_class.id}, Name: {tc_class.name}, Methods: {tc_class.applicable_methods}, Path Regex: {tc_class.applicable_paths_regex}") logger.info("\n--- 测试适用性筛选 ---") endpoint1_method = "GET" endpoint1_path = "/api/users/123" logger.info(f"筛选适用于 '{endpoint1_method} {endpoint1_path}':") applicable1 = registry.get_applicable_test_cases(endpoint1_method, endpoint1_path) for tc_class in applicable1: logger.info(f" Applicable: {tc_class.id} ({tc_class.name})") endpoint2_method = "POST" endpoint2_path = "/api/orders" logger.info(f"筛选适用于 '{endpoint2_method} {endpoint2_path}':") applicable2 = registry.get_applicable_test_cases(endpoint2_method, endpoint2_path) for tc_class in applicable2: logger.info(f" Applicable: {tc_class.id} ({tc_class.name})") endpoint3_method = "GET" endpoint3_path = "/api/health" logger.info(f"筛选适用于 '{endpoint3_method} {endpoint3_path}':") applicable3 = registry.get_applicable_test_cases(endpoint3_method, endpoint3_path) for tc_class in applicable3: logger.info(f" Applicable: {tc_class.id} ({tc_class.name})") # 清理临时文件和目录 # os.remove(os.path.join(custom_testcases_path, "example_checks.py")) # os.remove(os.path.join(custom_testcases_path, "specific_feature_tests.py")) # if not os.listdir(custom_testcases_path): # os.rmdir(custom_testcases_path) # logger.info("已清理临时文件和目录。")