fix:pdf\md\json

ddms_compliance_suite/stage_framework.py

@@ -14,7 +14,7 @@ from pydantic import BaseModel
 from .test_framework_core import ValidationResult, APIResponseContext
 from .api_caller.caller import APICallDetail
 # Import ParsedAPISpec and endpoint types for type hinting and usage
-from .input_parser.parser import ParsedAPISpec, YAPIEndpoint, SwaggerEndpoint
+from .input_parser.parser import ParsedAPISpec, BaseEndpoint, YAPIEndpoint, SwaggerEndpoint, DMSEndpoint
 
 # 尝试从 .llm_utils 导入，如果失败则 LLMService 为 None
 try:
@@ -199,7 +199,7 @@ class BaseAPIStage:
 
     def __init__(self,
                  api_group_metadata: Dict[str, Any],
-                 apis_in_group: List[Union[YAPIEndpoint, SwaggerEndpoint]], # MODIFIED TYPE HINT
+                 apis_in_group: List[BaseEndpoint], # MODIFIED TYPE HINT to use BaseEndpoint
                  llm_service: Optional[LLMService] = None,
                  global_api_spec: Optional[ParsedAPISpec] = None, # <--- 修改类型注解
                  operation_keywords: Optional[Dict[str, List[str]]] = None):
@@ -541,6 +541,7 @@ class ExecutedStageStepResult:
                  validation_points: Optional[List[ValidationResult]] = None,
                  duration: float = 0.0,
                  api_call_detail: Optional[APICallDetail] = None,
+                 api_operation_spec: Optional[BaseEndpoint] = None,  # <--- 添加此行
                  extracted_outputs: Optional[Dict[str, Any]] = None,
                  description: Optional[str] = None,
                  lookup_key: Optional[Union[str, Dict[str, str]]] = None,
@@ -555,6 +556,7 @@ class ExecutedStageStepResult:
         self.duration = duration
         self.timestamp = time.time()
         self.api_call_detail = api_call_detail
+        self.api_operation_spec = api_operation_spec # <--- 添加此行
         self.extracted_outputs = extracted_outputs or {}
         self.description = description
         self.lookup_key = lookup_key
@@ -608,6 +610,10 @@ class ExecutedStageStepResult:
             if failed_vp_messages:
                 current_message = "; ".join(failed_vp_messages)
 
+        api_op_dict = self.api_operation_spec # Changed from self.api_operation
+        if isinstance(api_op_dict, BaseEndpoint):
+            api_op_dict = api_op_dict.model_dump()
+
         return {
             "step_name": self.step_name,
             "description": self.description,
@@ -619,6 +625,7 @@ class ExecutedStageStepResult:
             "timestamp": time.strftime('%Y-%m-%dT%H:%M:%S%z', time.localtime(self.timestamp)),
             "validation_points": vps_details_for_output,
             "api_call_curl": getattr(getattr(self, 'api_call_detail', None), 'curl_command', 'N/A'),
+            "api_operation_spec": api_op_dict, # <--- 添加此行
             "request_details": self.request_details,
             "extracted_outputs": {k: str(v)[:200] + '...' if isinstance(v, (str, bytes)) and len(v) > 200 else v
                                   for k, v in self.extracted_outputs.items()},
@@ -672,17 +679,17 @@ class ExecutedStageResult:
         if not self.executed_steps and self.overall_status == ExecutedStageResult.Status.SKIPPED:
             # 如果没有执行任何步骤且状态是初始的 SKIPPED，则保持
             if not self.message: self.message = "此阶段没有执行任何步骤，被跳过。"
-        elif any(step.status == ExecutedStageStepResult.Status.ERROR for step in self.executed_steps):
+        elif any(step.status == ExecutedStageResult.Status.ERROR for step in self.executed_steps):
             self.overall_status = ExecutedStageResult.Status.FAILED # 步骤执行错误导致阶段失败
             if not self.message: self.message = "一个或多个步骤执行时发生内部错误。"
-        elif any(step.status == ExecutedStageStepResult.Status.FAILED for step in self.executed_steps):
+        elif any(step.status == ExecutedStageResult.Status.FAILED for step in self.executed_steps):
             self.overall_status = ExecutedStageResult.Status.FAILED
             if not self.message: self.message = "一个或多个步骤验证失败。"
-        elif all(step.status == ExecutedStageStepResult.Status.SKIPPED for step in self.executed_steps) and self.executed_steps:
+        elif all(step.status == ExecutedStageResult.Status.SKIPPED for step in self.executed_steps) and self.executed_steps:
              self.overall_status = ExecutedStageResult.Status.SKIPPED # 所有步骤都跳过了
              if not self.message: self.message = "所有步骤均被跳过。"
-        elif all(step.status == ExecutedStageStepResult.Status.PASSED or step.status == ExecutedStageStepResult.Status.SKIPPED for step in self.executed_steps) and \
-             any(step.status == ExecutedStageStepResult.Status.PASSED for step in self.executed_steps) :
+        elif all(step.status == ExecutedStageResult.Status.PASSED or step.status == ExecutedStageResult.Status.SKIPPED for step in self.executed_steps) and \
+             any(step.status == ExecutedStageResult.Status.PASSED for step in self.executed_steps) :
             self.overall_status = ExecutedStageResult.Status.PASSED # 至少一个通过，其他是跳过或通过
             if not self.message: self.message = "阶段执行成功。"
         else: # 其他情况，例如没有步骤但状态不是 SKIPPED (不应发生)，或者混合状态未被明确处理

ddms_compliance_suite/test_orchestrator.py

@@ -24,7 +24,7 @@ from pydantic import BaseModel, Field, create_model, HttpUrl # Added HttpUrl for
 from pydantic.networks import EmailStr
 from pydantic.types import Literal # Explicitly import Literal
 
-from .input_parser.parser import InputParser, YAPIEndpoint, SwaggerEndpoint, ParsedYAPISpec, ParsedSwaggerSpec, ParsedAPISpec, DMSEndpoint, ParsedDMSSpec
+from .input_parser.parser import InputParser, BaseEndpoint, YAPIEndpoint, SwaggerEndpoint, ParsedYAPISpec, ParsedSwaggerSpec, ParsedAPISpec, DMSEndpoint, ParsedDMSSpec
 from .api_caller.caller import APICaller, APIRequest, APIResponse, APICallDetail # Ensure APICallDetail is imported
 from .json_schema_validator.validator import JSONSchemaValidator
 from .test_framework_core import ValidationResult, TestSeverity, APIRequestContext, APIResponseContext, BaseAPITestCase
@@ -342,8 +342,14 @@ class TestSummary:
         return data
         
     def to_json(self, pretty=True) -> str:
-        indent = 2 if pretty else None
-        return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False)
+        def custom_serializer(obj):
+            """JSON serializer for objects not serializable by default json code"""
+            if isinstance(obj, BaseEndpoint):
+                return obj.to_dict()
+            raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
+
+        indent = 4 if pretty else None
+        return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False, default=custom_serializer)
 
     def print_summary_to_console(self): # Renamed from print_summary
         # (Implementation can be more detailed based on the new stats)