docker
This commit is contained in:
gongwenxin
@@ -3,7 +3,7 @@ API 阶段测试框架模块
|
||||
"""
|
||||
import logging
|
||||
import time
|
||||
from typing import List, Dict, Any, Callable, Optional, Union
|
||||
from typing import List, Dict, Any, Callable, Optional, Union, TYPE_CHECKING
|
||||
from enum import Enum
|
||||
from datetime import datetime
|
||||
from dataclasses import dataclass, field
|
||||
@@ -18,6 +18,9 @@ from .api_caller.caller import APICallDetail
|
||||
from .input_parser.parser import ParsedAPISpec, YAPIEndpoint, SwaggerEndpoint,BaseEndpoint,DMSEndpoint
|
||||
from .utils.context_utils import serialize_context_recursively
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .input_parser.parser import ParsedAPISpec
|
||||
|
||||
# 尝试从 .llm_utils 导入，如果失败则 LLMService 为 None
|
||||
try:
|
||||
from .llm_utils.llm_service import LLMService
|
||||
@@ -262,98 +265,7 @@ class BaseAPIStage:
|
||||
else:
|
||||
self.steps = []
|
||||
|
||||
def get_api_spec_for_operation(self,
|
||||
lookup_key: str,
|
||||
global_api_spec: ParsedAPISpec, # 确保类型正确
|
||||
api_group_name: Optional[str] = None
|
||||
) -> Optional[APIOperationSpec]:
|
||||
"""
|
||||
根据查找键从提供的API规范中获取特定API操作的详细信息。
|
||||
这个方法需要子类根据其查找逻辑来实现，或者提供一个通用的基于operationId或path+method的查找。
|
||||
|
||||
Args:
|
||||
lookup_key: 用于查找API操作的键 (例如 operationId, "METHOD /path", 或自定义键)。
|
||||
global_api_spec: 完整的已解析API规范对象。
|
||||
api_group_name: (可选) 当前API分组的名称，用于更精确的查找或作用域限定。
|
||||
|
||||
Returns:
|
||||
APIOperationSpec 对象如果找到，否则 None。
|
||||
"""
|
||||
self.logger.debug(f"Attempting to find API spec for lookup_key='{lookup_key}', api_group='{api_group_name}'")
|
||||
|
||||
# 尝试基于 operationId 查找 (如果 lookup_key 看起来像 operationId)
|
||||
# 简单的启发式：不包含空格和斜杠的键可能是 operationId
|
||||
if ' ' not in lookup_key and '/' not in lookup_key:
|
||||
for endpoint in global_api_spec.endpoints:
|
||||
endpoint_dict = endpoint.to_dict() # 将YAPIEndpoint或SwaggerEndpoint转换为标准化字典
|
||||
op_id = endpoint_dict.get('operationId')
|
||||
if op_id == lookup_key:
|
||||
# 检查此端点是否属于当前API分组 (如果提供了api_group_name)
|
||||
if api_group_name:
|
||||
tags = endpoint_dict.get('tags', [])
|
||||
if api_group_name not in tags:
|
||||
self.logger.debug(f"Endpoint with operationId '{lookup_key}' found, but not in group '{api_group_name}'. Tags: {tags}")
|
||||
continue # 不在当前组，继续查找
|
||||
|
||||
self.logger.info(f"Found API for operationId '{lookup_key}' (group: {api_group_name}). Path: {endpoint_dict.get('path')}")
|
||||
return APIOperationSpec(
|
||||
method=endpoint_dict.get('method','').upper(),
|
||||
path=endpoint_dict.get('path',''),
|
||||
spec=endpoint_dict, # 传递整个端点字典作为规范
|
||||
operation_id=op_id,
|
||||
summary=endpoint_dict.get('summary'),
|
||||
description=endpoint_dict.get('description'),
|
||||
tags=endpoint_dict.get('tags', [])
|
||||
)
|
||||
|
||||
# 尝试基于 "METHOD /path" 格式的 lookup_key 查找
|
||||
# (这是一个更通用的查找方式，但可能需要更仔细的路径匹配逻辑)
|
||||
parts = lookup_key.split(' ', 1)
|
||||
if len(parts) == 2:
|
||||
method_to_find = parts[0].upper()
|
||||
path_to_find = parts[1]
|
||||
for endpoint in global_api_spec.endpoints:
|
||||
endpoint_dict = endpoint.to_dict()
|
||||
if endpoint_dict.get('method','').upper() == method_to_find and endpoint_dict.get('path','') == path_to_find:
|
||||
if api_group_name:
|
||||
tags = endpoint_dict.get('tags', [])
|
||||
if api_group_name not in tags:
|
||||
self.logger.debug(f"Endpoint '{lookup_key}' found, but not in group '{api_group_name}'. Tags: {tags}")
|
||||
continue
|
||||
self.logger.info(f"Found API for method/path '{lookup_key}' (group: {api_group_name}).")
|
||||
return APIOperationSpec(
|
||||
method=method_to_find,
|
||||
path=path_to_find,
|
||||
spec=endpoint_dict,
|
||||
operation_id=endpoint_dict.get('operationId'),
|
||||
summary=endpoint_dict.get('summary'),
|
||||
description=endpoint_dict.get('description'),
|
||||
tags=endpoint_dict.get('tags', [])
|
||||
)
|
||||
|
||||
self.logger.warning(f"Could not find API operation spec for lookup_key: '{lookup_key}' (api_group: '{api_group_name}') using default search logic (operationId or METHOD /path). Consider overriding get_api_spec_for_operation in your Stage class for custom lookup logic.")
|
||||
return None
|
||||
|
||||
# --- 生命周期钩子 ---
|
||||
def before_stage(self, stage_context: Dict[str, Any], global_api_spec: ParsedAPISpec, api_group_name: Optional[str]):
|
||||
"""在测试阶段所有步骤执行之前调用。"""
|
||||
self.logger.debug(f"Executing before_stage for stage '{self.id}', group '{api_group_name}'. Initial context: {stage_context}")
|
||||
pass
|
||||
|
||||
def after_stage(self, stage_result: ExecutedStageResult, stage_context: Dict[str, Any], global_api_spec: ParsedAPISpec, api_group_name: Optional[str]):
|
||||
"""在测试阶段所有步骤执行完毕之后调用（无论成功、失败或错误）。"""
|
||||
self.logger.debug(f"Executing after_stage for stage '{self.id}', group '{api_group_name}'. Result status: {stage_result.overall_status.value}. Final context: {stage_context}")
|
||||
pass
|
||||
|
||||
def before_step(self, step: StageStepDefinition, stage_context: Dict[str, Any], global_api_spec: ParsedAPISpec, api_group_name: Optional[str]):
|
||||
"""在每个测试步骤执行之前调用。"""
|
||||
self.logger.debug(f"Executing before_step for stage '{self.id}', step '{step.name}', group '{api_group_name}'. Current context: {stage_context}")
|
||||
pass
|
||||
|
||||
def after_step(self, step: StageStepDefinition, step_result: ExecutedStageStepResult, stage_context: Dict[str, Any], global_api_spec: ParsedAPISpec, api_group_name: Optional[str]):
|
||||
"""在每个测试步骤执行之后调用（无论成功、失败或错误）。"""
|
||||
self.logger.debug(f"Executing after_step for stage '{self.id}', step '{step.name}', group '{api_group_name}'. Step status: {step_result.status.value}. Context after step: {stage_context}")
|
||||
pass
|
||||
|
||||
def is_applicable_to_api_group(self, api_group_name: Optional[str], global_api_spec: ParsedAPISpec) -> bool:
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user