修改了测试严格程度，高于指定严格程度的用例才影响api是否通过

2025-06-27 18:44:36 +08:00
parent 864a36fe61
commit f003fbbbd1
23 changed files with 52248 additions and 9149 deletions
@@ -5,11 +5,19 @@ from .utils import schema_utils

 class TestSeverity(Enum):
    """测试用例的严重程度"""
-    CRITICAL = "严重"
-    HIGH = "高"
-    MEDIUM = "中"
-    LOW = "低"
-    INFO = "信息"
+    CRITICAL = 5
+    HIGH = 4
+    MEDIUM = 3
+    LOW = 2
+    INFO = 1
+    def __ge__(self, other):
+        return self.value >= other.value
+    def __gt__(self, other):
+        return self.value > other.value
+    def __le__(self, other):
+        return self.value <= other.value
+    def __lt__(self, other):
+        return self.value < other.value

 class ValidationResult:
    """封装单个验证点的结果"""
@@ -80,7 +80,7 @@ class ExecutedTestCaseResult:
        return {
            "test_case_id": self.test_case_id,
            "test_case_name": self.test_case_name,
-            "test_case_severity": self.test_case_severity.value, # 使用枚举值
+            "test_case_severity": self.test_case_severity.name, # 使用枚举名称
            "status": self.status.value,
            "message": message,
            "duration_seconds": self.duration,
@@ -115,41 +115,60 @@ class TestResult: # 原来的 TestResult 被重构为 EndpointExecutionResult
        self.end_time: Optional[datetime.datetime] = None
        self.error_message: Optional[str] = None # 如果整个端点测试出错，记录错误信息
        self.message: Optional[str] = None 
+        self.strictness_level: Optional[TestSeverity] = None

    def add_executed_test_case_result(self, result: ExecutedTestCaseResult):
        self.executed_test_cases.append(result)

-    def finalize_endpoint_test(self):
+    def finalize_endpoint_test(self, strictness_level: Optional[TestSeverity] = None):
        self.end_time = datetime.datetime.now()
-        # 根据所有 executed_test_cases 的状态和严重性来计算 overall_status
-        if not self.executed_test_cases and self.overall_status == TestResult.Status.SKIPPED : # 如果没有执行任何测试用例且状态仍为初始的SKIPPED
-             pass # 保持 SKIPPED
-        elif any(tc.status == ExecutedTestCaseResult.Status.ERROR for tc in self.executed_test_cases):
-            self.overall_status = TestResult.Status.ERROR
-            # 可以考虑将第一个遇到的ERROR的message赋给self.error_message
-            first_error = next((tc.message for tc in self.executed_test_cases if tc.status == ExecutedTestCaseResult.Status.ERROR), None)
-            if first_error:
-                self.error_message = f"测试用例执行错误: {first_error}"
-        else:
-            # 筛选出失败的测试用例
-            failed_tcs = [tc for tc in self.executed_test_cases if tc.status == ExecutedTestCaseResult.Status.FAILED]
-            if not failed_tcs:
-                if not self.executed_test_cases: # 如果没有执行任何测试用例但又不是SKIPPED，可能也算某种形式的错误或特殊通过
-                    self.overall_status = TestResult.Status.PASSED # 或者定义一个"NO_CASES_RUN"状态
-                else:
-                    self.overall_status = TestResult.Status.PASSED
-            else:
-                # 检查失败的测试用例中是否有CRITICAL或HIGH严重级别的
-                if any(tc.test_case_severity in [TestSeverity.CRITICAL, TestSeverity.HIGH] for tc in failed_tcs):
-                    self.overall_status = TestResult.Status.FAILED
-                else: # 所有失败的都是 MEDIUM, LOW, INFO
-                    self.overall_status = TestResult.Status.PARTIAL_SUCCESS
-        
-        if not self.executed_test_cases and self.overall_status not in [TestResult.Status.SKIPPED, TestResult.Status.ERROR]:
-             # 如果没有执行测试用例，并且不是因为错误或明确跳过，这可能是一个配置问题或意外情况
-             self.overall_status = TestResult.Status.ERROR # 或者一个更特定的状态
-             self.error_message = "没有为该端点找到或执行任何适用的测试用例。"
+        self.strictness_level = strictness_level

+        # 检查是否有测试用例执行出错
+        if any(tc.status == ExecutedTestCaseResult.Status.ERROR for tc in self.executed_test_cases):
+            self.overall_status = TestResult.Status.ERROR
+            first_error = next((tc.message for tc in self.executed_test_cases if tc.status == ExecutedTestCaseResult.Status.ERROR), "Unknown test case error")
+            self.error_message = f"测试用例执行错误: {first_error}"
+            return
+        
+        # 如果没有执行任何测试用例
+        if not self.executed_test_cases:
+            if self.overall_status == TestResult.Status.SKIPPED:
+                # 保持 SKIPPED 状态
+                return
+            else:
+                self.overall_status = TestResult.Status.ERROR
+                self.error_message = "没有为该端点找到或执行任何适用的测试用例。"
+                return
+
+        # 根据 strictness_level 决定最终状态
+        failed_tcs = [tc for tc in self.executed_test_cases if tc.status == ExecutedTestCaseResult.Status.FAILED]
+
+        if not failed_tcs:
+            self.overall_status = TestResult.Status.PASSED
+            return
+        logging.info(f"strictness_level: {strictness_level}")
+        # 如果定义了严格等级，只关心高于或等于该等级的失败用例
+        if self.strictness_level:
+            # TestSeverity Enum is ordered, so we can compare them
+            logging.info(f"strictness_level: {strictness_level}")
+            relevant_failed_tcs = [
+                tc for tc in failed_tcs 
+                if tc.test_case_severity >= strictness_level
+            ]
+            if not relevant_failed_tcs:
+                self.overall_status = TestResult.Status.PASSED
+                self.message = f"通过（严格等级: {strictness_level.name}）。注意：存在 {len(failed_tcs)} 个较低严重性的失败用例。"
+            else:
+                self.overall_status = TestResult.Status.FAILED
+                self.message = f"失败（严格等级: {strictness_level.name}）。"
+            logging.info(f"relevant_failed_tcs: {relevant_failed_tcs}")
+        else:
+            # 默认行为：任何失败都可能导致 FAILED 或 PARTIAL_SUCCESS
+            if any(tc.test_case_severity in [TestSeverity.CRITICAL, TestSeverity.HIGH] for tc in failed_tcs):
+                self.overall_status = TestResult.Status.FAILED
+            else: 
+                self.overall_status = TestResult.Status.PARTIAL_SUCCESS

    @property
    def duration(self) -> float:
@@ -407,8 +426,26 @@ class APITestOrchestrator:
                 use_llm_for_path_params: bool = False,
                 use_llm_for_query_params: bool = False,
                 use_llm_for_headers: bool = False,
-                 output_dir: Optional[str] = None 
+                 output_dir: Optional[str] = None,
+                 strictness_level: Optional[str] = None
                 ):
+        """
+        初始化测试编排器。
+
+        Args:
+            base_url (str): API的基础URL。
+            custom_test_cases_dir (Optional[str]): 存放自定义测试用例的目录。
+            stages_dir (Optional[str]): 存放自定义测试阶段的目录。
+            llm_api_key (Optional[str]): LLM服务的API密钥。
+            llm_base_url (Optional[str]): LLM服务的自定义基础URL。
+            llm_model_name (Optional[str]): 要使用的LLM模型名称。
+            use_llm_for_request_body (bool): 是否使用LLM生成请求体。
+            use_llm_for_path_params (bool): 是否使用LLM生成路径参数。
+            use_llm_for_query_params (bool): 是否使用LLM生成查询参数。
+            use_llm_for_headers (bool): 是否使用LLM生成头部参数。
+            output_dir (Optional[str]): 测试报告和工件的输出目录。
+            strictness_level (Optional[str]): 测试的严格等级, 如 'CRITICAL', 'HIGH'。
+        """
        self.logger = logging.getLogger(__name__)
        self.base_url = base_url.rstrip('/') 
        self.api_caller = APICaller() 
@@ -466,6 +503,14 @@ class APITestOrchestrator:
        self.json_resolver_cache: Dict[str, Any] = {} 
        self.json_validator = JSONSchemaValidator()

+        # 将字符串类型的 strictness_level 转换为 TestSeverity 枚举成员
+        self.strictness_level: Optional[TestSeverity] = None
+        if strictness_level and hasattr(TestSeverity, strictness_level):
+            self.strictness_level = TestSeverity[strictness_level]
+            logging.info(f"strictness_level: {self.strictness_level}")
+        elif strictness_level:
+            logging.warning(f"提供了无效的严格等级 '{strictness_level}'。将使用默认行为。有效值: {', '.join([e.name for e in TestSeverity])}")
+
    def get_api_call_details(self) -> List[APICallDetail]:
        """Returns the collected list of API call details."""
        return self.global_api_call_details
@@ -1443,7 +1488,7 @@ class APITestOrchestrator:
            self.logger.warning(f"TestCaseRegistry 未初始化，无法为端点 '{endpoint_id}' 执行自定义测试用例。")
            endpoint_test_result.overall_status = TestResult.Status.SKIPPED
            endpoint_test_result.error_message = "TestCaseRegistry 未初始化。"
-            endpoint_test_result.finalize_endpoint_test()
+            endpoint_test_result.finalize_endpoint_test(strictness_level=self.strictness_level)
            return endpoint_test_result

        applicable_test_case_classes_unordered = self.test_case_registry.get_applicable_test_cases(
@@ -1453,7 +1498,7 @@ class APITestOrchestrator:

        if not applicable_test_case_classes_unordered:
            self.logger.info(f"端点 '{endpoint_id}' 没有找到适用的自定义测试用例。")
-            endpoint_test_result.finalize_endpoint_test() # 确保在返回前调用
+            endpoint_test_result.finalize_endpoint_test(strictness_level=self.strictness_level) # 确保在返回前调用
            return endpoint_test_result
        
        # 根据 execution_order 排序测试用例
@@ -1512,7 +1557,7 @@ class APITestOrchestrator:

            self.logger.debug(f"测试用例 '{tc_class.id}' 执行完毕，状态: {executed_case_result.status.value}")
            
-        endpoint_test_result.finalize_endpoint_test()
+        endpoint_test_result.finalize_endpoint_test(strictness_level=self.strictness_level)
        self.logger.info(f"端点 '{endpoint_id}' 测试完成，最终状态: {endpoint_test_result.overall_status.value}")
        
        return endpoint_test_result