from typing import List, Dict, Any, Optional, Callable, Union import datetime import logging from enum import Enum from .test_framework_core import ValidationResult, APIRequestContext, APIResponseContext from .api_caller.caller import APICallDetail # 需要APICallDetail来记录每个步骤的调用 class ScenarioStepDefinition: """定义API场景中的一个单独步骤。""" def __init__(self, name: str, endpoint_spec_lookup_key: str, # 用于从全局API规范中查找端点定义的键 request_overrides: Optional[Dict[str, Any]] = None, expected_status_codes: Optional[List[int]] = None, response_assertions: Optional[List[Callable[[APIResponseContext, Dict[str, Any]], List[ValidationResult]]]] = None, outputs_to_context: Optional[Dict[str, str]] = None): """ Args: name: 步骤的可读名称。 endpoint_spec_lookup_key: 用于查找端点定义的键 (例如 "METHOD /path" 或 YAPI的_id)。 request_overrides: 覆盖默认请求参数的字典。值可以是占位符，如 "{{scenario_context.user_id}}"。 支持的键有: "path_params", "query_params", "headers", "body"。 expected_status_codes: 预期的HTTP响应状态码列表。如果为None，则不进行特定状态码检查（除非由response_assertions处理）。 response_assertions: 自定义断言函数列表。每个函数接收 APIResponseContext 和 scenario_context，返回 ValidationResult 列表。 outputs_to_context: 从响应中提取数据到场景上下文的字典。 键是存储到场景上下文中的变量名，值是提取路径 (例如 "response.body.data.id")。 """ self.name = name self.endpoint_spec_lookup_key = endpoint_spec_lookup_key self.request_overrides = request_overrides if request_overrides is not None else {} self.expected_status_codes = expected_status_codes if expected_status_codes is not None else [] self.response_assertions = response_assertions if response_assertions is not None else [] self.outputs_to_context = outputs_to_context if outputs_to_context is not None else {} self.logger = logging.getLogger(f"scenario.step.{name}") class BaseAPIScenario: """ API场景测试用例的基类。 用户应继承此类来创建具体的测试场景。 """ # --- 元数据 (由子类定义) --- id: str = "base_api_scenario" name: str = "基础API场景" description: str = "这是一个基础API场景，应由具体场景继承。" tags: List[str] = [] steps: List[ScenarioStepDefinition] = [] # 子类需要填充此列表 def __init__(self, global_api_spec: Dict[str, Any], # 完整的API规范字典 (YAPI/Swagger解析后的原始字典) parsed_api_endpoints: List[Dict[str, Any]], # 解析后的端点列表，用于通过 lookup_key 查找 llm_service: Optional[Any] = None): """ 初始化API场景。 Args: global_api_spec: 完整的API规范字典。 parsed_api_endpoints: 从YAPI/Swagger解析出来的端点对象列表（通常是YAPIEndpoint或SwaggerEndpoint的to_dict()结果）。 这些对象应包含用于匹配 `endpoint_spec_lookup_key` 的字段。 llm_service: APITestOrchestrator 传入的 LLMService 实例 (可选)。 """ self.global_api_spec = global_api_spec self.parsed_api_endpoints = parsed_api_endpoints # 用于快速查找 self.llm_service = llm_service self.logger = logging.getLogger(f"scenario.{self.id}") self.logger.info(f"API场景 '{self.id}' ({self.name}) 已初始化。") def _get_endpoint_spec_from_global(self, lookup_key: str) -> Optional[Dict[str, Any]]: """ 根据提供的 lookup_key 从 self.parsed_api_endpoints 中查找并返回端点定义。 查找逻辑可能需要根据 lookup_key 的格式 (例如, YAPI _id, METHOD /path) 进行调整。 简单实现：假设 lookup_key 是 METHOD /path 或 title。 """ self.logger.debug(f"尝试为场景步骤查找端点: '{lookup_key}'") for endpoint_data in self.parsed_api_endpoints: # 尝试匹配 "METHOD /path" 格式 (常见于SwaggerEndpoint) method_path_key = f"{str(endpoint_data.get('method', '')).upper()} {endpoint_data.get('path', '')}" if lookup_key == method_path_key: self.logger.debug(f"通过 'METHOD /path' ('{method_path_key}') 找到端点。") return endpoint_data # 尝试匹配 title (常见于YAPIEndpoint) if lookup_key == endpoint_data.get('title'): self.logger.debug(f"通过 'title' ('{endpoint_data.get('title')}') 找到端点。") return endpoint_data # 尝试匹配 YAPI 的 _id (如果可用) if str(lookup_key) == str(endpoint_data.get('_id')): # 转换为字符串以确保比较 self.logger.debug(f"通过 YAPI '_id' ('{endpoint_data.get('_id')}') 找到端点。") return endpoint_data # 尝试匹配 Swagger/OpenAPI 的 operationId if lookup_key == endpoint_data.get('operationId'): self.logger.debug(f"通过 'operationId' ('{endpoint_data.get('operationId')}') 找到端点。") return endpoint_data self.logger.warning(f"未能在 parsed_api_endpoints 中找到 lookup_key 为 '{lookup_key}' 的端点。") return None def before_scenario(self, scenario_context: Dict[str, Any]): """在场景所有步骤执行前调用 (可选，供子类覆盖)""" self.logger.debug(f"Hook: before_scenario for '{self.id}'") pass def after_scenario(self, scenario_context: Dict[str, Any], scenario_result: 'ExecutedScenarioResult'): """在场景所有步骤执行后调用 (可选，供子类覆盖)""" self.logger.debug(f"Hook: after_scenario for '{self.id}'") pass def before_step(self, step_definition: ScenarioStepDefinition, scenario_context: Dict[str, Any]): """在每个步骤执行前调用 (可选，供子类覆盖)""" self.logger.debug(f"Hook: before_step '{step_definition.name}' for '{self.id}'") pass def after_step(self, step_definition: ScenarioStepDefinition, step_result: 'ExecutedScenarioStepResult', scenario_context: Dict[str, Any]): """在每个步骤执行后调用 (可选，供子类覆盖)""" self.logger.debug(f"Hook: after_step '{step_definition.name}' for '{self.id}'") pass class ExecutedScenarioStepResult: """存储单个API场景步骤执行后的结果。""" class Status(str, Enum): PASSED = "通过" FAILED = "失败" ERROR = "执行错误" SKIPPED = "跳过" def __init__(self, step_name: str, status: Status, message: str = "", validation_points: Optional[List[ValidationResult]] = None, duration: float = 0.0, api_call_detail: Optional[APICallDetail] = None, extracted_outputs: Optional[Dict[str, Any]] = None): self.step_name = step_name self.status = status self.message = message self.validation_points = validation_points if validation_points is not None else [] self.duration = duration self.api_call_detail = api_call_detail # 存储此步骤的API调用详情 self.extracted_outputs = extracted_outputs if extracted_outputs is not None else {} # 从此步骤提取并存入上下文的值 self.timestamp = datetime.datetime.now() def to_dict(self) -> Dict[str, Any]: return { "step_name": self.step_name, "status": self.status.value, "message": self.message, "duration_seconds": self.duration, "timestamp": self.timestamp.isoformat(), "validation_points": [vp.to_dict() if hasattr(vp, 'to_dict') else {"passed": vp.passed, "message": vp.message, "details": vp.details} for vp in self.validation_points], "api_call_detail": self.api_call_detail.to_dict() if self.api_call_detail and hasattr(self.api_call_detail, 'to_dict') else None, "extracted_outputs": self.extracted_outputs } class ExecutedScenarioResult: """存储整个API场景执行后的结果。""" class Status(str, Enum): PASSED = "通过" # 所有步骤都通过 FAILED = "失败" # 任何一个步骤失败或出错 SKIPPED = "跳过" # 整个场景被跳过 def __init__(self, scenario_id: str, scenario_name: str, overall_status: Status = Status.SKIPPED, message: str = ""): self.scenario_id = scenario_id self.scenario_name = scenario_name self.overall_status = overall_status self.message = message self.executed_steps: List[ExecutedScenarioStepResult] = [] self.scenario_context_final_state: Dict[str, Any] = {} self.start_time = datetime.datetime.now() self.end_time: Optional[datetime.datetime] = None def add_step_result(self, result: ExecutedScenarioStepResult): self.executed_steps.append(result) def finalize_scenario_result(self, final_context: Dict[str, Any]): self.end_time = datetime.datetime.now() self.scenario_context_final_state = final_context if not self.executed_steps and self.overall_status == ExecutedScenarioResult.Status.SKIPPED: pass # 保持 SKIPPED elif any(step.status == ExecutedScenarioStepResult.Status.ERROR for step in self.executed_steps): self.overall_status = ExecutedScenarioResult.Status.FAILED if not self.message: self.message = "场景中至少一个步骤执行出错。" elif any(step.status == ExecutedScenarioStepResult.Status.FAILED for step in self.executed_steps): self.overall_status = ExecutedScenarioResult.Status.FAILED if not self.message: self.message = "场景中至少一个步骤失败。" elif all(step.status == ExecutedScenarioStepResult.Status.SKIPPED for step in self.executed_steps) and self.executed_steps: self.overall_status = ExecutedScenarioResult.Status.SKIPPED # 如果所有步骤都跳过了 if not self.message: self.message = "场景中的所有步骤都被跳过。" elif not self.executed_steps: # 没有步骤执行，也不是初始的SKIPPED self.overall_status = ExecutedScenarioResult.Status.FAILED # 或 ERROR if not self.message: self.message = "场景中没有步骤被执行。" else: # 所有步骤都通过 self.overall_status = ExecutedScenarioResult.Status.PASSED if not self.message: self.message = "场景所有步骤成功通过。" @property def duration(self) -> float: if self.start_time and self.end_time: return (self.end_time - self.start_time).total_seconds() return 0.0 def to_dict(self) -> Dict[str, Any]: return { "scenario_id": self.scenario_id, "scenario_name": self.scenario_name, "overall_status": self.overall_status.value, "message": self.message, "duration_seconds": f"{self.duration:.2f}", "start_time": self.start_time.isoformat(), "end_time": self.end_time.isoformat() if self.end_time else None, "executed_steps": [step.to_dict() for step in self.executed_steps], "scenario_context_final_state": self.scenario_context_final_state # 可能包含敏感信息，按需处理 } def to_json(self, pretty=True) -> str: import json # 局部导入 indent = 2 if pretty else None # 对于 scenario_context_final_state，可能需要自定义序列化器来处理复杂对象 return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False, default=str)