step1

ddms_compliance_suite/test_case_registry.py

@@ -0,0 +1,210 @@
+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)
+        
+        self.logger.info(f"测试用例发现完成。总共注册了 {len(self._registry)} 个独特的测试用例 (基于ID)。发现 {found_count} 个测试用例类。")
+
+    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("已清理临时文件和目录。") 

ddms_compliance_suite/test_framework_core.py

@@ -0,0 +1,143 @@
+from enum import Enum
+from typing import Any, Dict, Optional, List, Tuple, Type
+import logging
+
+class TestSeverity(Enum):
+    """测试用例的严重程度"""
+    CRITICAL = "严重"
+    HIGH = "高"
+    MEDIUM = "中"
+    LOW = "低"
+    INFO = "信息"
+
+class ValidationResult:
+    """封装单个验证点的结果"""
+    def __init__(self, passed: bool, message: str, details: Optional[Dict[str, Any]] = None):
+        self.passed = passed      # True 表示通过, False 表示失败
+        self.message = message    # 验证结果的描述信息
+        self.details = details or {} # 其他详细信息，如实际值、期望值等
+
+    def __repr__(self):
+        return f"ValidationResult(passed={self.passed}, message='{self.message}')"
+
+class APIRequestContext:
+    """封装当前API请求的上下文信息"""
+    def __init__(self, 
+                 method: str, 
+                 url: str, 
+                 path_params: Dict[str, Any],
+                 query_params: Dict[str, Any], 
+                 headers: Dict[str, str], 
+                 body: Optional[Any],
+                 endpoint_spec: Dict[str, Any] # 添加 endpoint_spec 到请求上下文
+                 ):
+        self.method = method
+        self.url = url
+        self.path_params = path_params
+        self.query_params = query_params
+        self.headers = headers
+        self.body = body
+        self.endpoint_spec = endpoint_spec # 当前测试端点的API规范部分
+
+    def __repr__(self):
+        return (f"APIRequestContext(method='{self.method}', url='{self.url}', "
+                f"query_params={self.query_params}, headers_keys={list(self.headers.keys())}, "
+                f"has_body={self.body is not None})")
+
+class APIResponseContext:
+    """封装当前API响应的上下文信息"""
+    def __init__(self, 
+                 status_code: int, 
+                 headers: Dict[str, str],
+                 json_content: Optional[Any], 
+                 text_content: Optional[str],
+                 elapsed_time: float, 
+                 original_response: Any, # 例如 requests.Response 对象
+                 request_context: APIRequestContext # 对应的请求上下文
+                 ):
+        self.status_code = status_code
+        self.headers = headers
+        self.json_content = json_content
+        self.text_content = text_content
+        self.elapsed_time = elapsed_time 
+        self.original_response = original_response
+        self.request_context = request_context # 包含触发此响应的请求信息
+
+    def __repr__(self):
+        return (f"APIResponseContext(status_code={self.status_code}, "
+                f"headers_keys={list(self.headers.keys())}, has_json={self.json_content is not None}, "
+                f"elapsed_time={self.elapsed_time:.4f}s)")
+
+
+class BaseAPITestCase:
+    """
+    自定义API测试用例的基类。
+    用户应继承此类来创建具体的测试用例。
+    """
+    # --- 元数据 (由子类定义) ---
+    id: str = "base_test_case"                 
+    name: str = "基础API测试用例"             
+    description: str = "这是一个基础测试用例，应由具体测试用例继承。" 
+    severity: TestSeverity = TestSeverity.MEDIUM 
+    tags: List[str] = []                       
+    
+    applicable_methods: Optional[List[str]] = None  
+    applicable_paths_regex: Optional[str] = None    
+
+    def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any]):
+        """
+        初始化测试用例。
+        Args:
+            endpoint_spec: 当前被测API端点的详细定义 (来自YAPI/Swagger解析结果)。
+            global_api_spec: 完整的API规范文档 (来自YAPI/Swagger解析结果)。
+        """
+        self.endpoint_spec = endpoint_spec
+        self.global_api_spec = global_api_spec
+        self.logger = logging.getLogger(f"testcase.{self.id}")
+        self.logger.debug(f"Test case '{self.id}' initialized for endpoint: {self.endpoint_spec.get('method', '')} {self.endpoint_spec.get('path', '')}")
+
+    # --- 1. 请求生成与修改阶段 ---
+    def generate_query_params(self, current_query_params: Dict[str, Any]) -> Dict[str, Any]:
+        self.logger.debug(f"Hook: generate_query_params, current: {current_query_params}")
+        return current_query_params
+
+    def generate_headers(self, current_headers: Dict[str, str]) -> Dict[str, str]:
+        self.logger.debug(f"Hook: generate_headers, current keys: {list(current_headers.keys())}")
+        return current_headers
+
+    def generate_request_body(self, current_body: Optional[Any]) -> Optional[Any]:
+        self.logger.debug(f"Hook: generate_request_body, current body type: {type(current_body)}")
+        return current_body
+
+    # --- 2. 请求预校验阶段 ---
+    def validate_request_url(self, url: str, request_context: APIRequestContext) -> List[ValidationResult]:
+        self.logger.debug(f"Hook: validate_request_url, url: {url}")
+        return []
+
+    def validate_request_headers(self, headers: Dict[str, str], request_context: APIRequestContext) -> List[ValidationResult]:
+        self.logger.debug(f"Hook: validate_request_headers, header keys: {list(headers.keys())}")
+        return []
+
+    def validate_request_body(self, body: Optional[Any], request_context: APIRequestContext) -> List[ValidationResult]:
+        self.logger.debug(f"Hook: validate_request_body, body type: {type(body)}")
+        return []
+
+    # --- 3. 响应验证阶段 ---
+    def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
+        self.logger.debug(f"Hook: validate_response, status: {response_context.status_code}")
+        return []
+
+    # --- 4. 性能与附加检查阶段 (可选) ---
+    def check_performance(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
+        self.logger.debug(f"Hook: check_performance, elapsed: {response_context.elapsed_time}")
+        return []
+
+    # --- Helper to easily create a passed ValidationResult ---
+    @staticmethod
+    def passed(message: str, details: Optional[Dict[str, Any]] = None) -> ValidationResult:
+        return ValidationResult(passed=True, message=message, details=details)
+
+    # --- Helper to easily create a failed ValidationResult ---
+    @staticmethod
+    def failed(message: str, details: Optional[Dict[str, Any]] = None) -> ValidationResult:
+        return ValidationResult(passed=False, message=message, details=details)