228 lines
12 KiB
Python
228 lines
12 KiB
Python
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) |