460 lines
19 KiB
Python
460 lines
19 KiB
Python
# @line_count 250
|
||
"""测试生成引擎模块"""
|
||
import json
|
||
import re
|
||
from typing import List, Dict, Any, Optional, Callable
|
||
from .json_parser import JSONParser
|
||
from .api_client import APIClient
|
||
from .prompt_manager import PromptManager
|
||
|
||
|
||
class TestGenerator:
|
||
"""测试项和测试用例生成器"""
|
||
|
||
def __init__(self, json_path: str, api_client: Optional[APIClient] = None,
|
||
prompt_manager: Optional[PromptManager] = None):
|
||
"""
|
||
初始化测试生成器
|
||
|
||
Args:
|
||
json_path: JSON文档路径
|
||
api_client: API客户端实例
|
||
prompt_manager: Prompt管理器实例
|
||
"""
|
||
self.parser = JSONParser(json_path)
|
||
self.api_client = api_client or APIClient()
|
||
self.prompt_manager = prompt_manager or PromptManager()
|
||
self.test_items: List[Dict[str, Any]] = []
|
||
self.test_cases: List[Dict[str, Any]] = []
|
||
self.test_case_counter = 0
|
||
|
||
def generate_test_items(self, function_points: Optional[List[Dict[str, Any]]] = None,
|
||
progress_callback: Optional[Callable] = None) -> List[Dict[str, Any]]:
|
||
"""
|
||
生成测试项
|
||
|
||
Args:
|
||
function_points: 功能点列表，如果为None则自动提取
|
||
progress_callback: 进度回调函数 (current, total, message)
|
||
|
||
Returns:
|
||
测试项列表
|
||
"""
|
||
if function_points is None:
|
||
function_points = self.parser.extract_function_points()
|
||
|
||
test_items = []
|
||
total = len(function_points)
|
||
|
||
for idx, func_point in enumerate(function_points):
|
||
if progress_callback:
|
||
progress_callback(idx + 1, total, f"正在生成测试项: {func_point['function_name']}")
|
||
|
||
try:
|
||
# 构建功能描述
|
||
function_description = self._build_function_description(func_point)
|
||
|
||
# 格式化prompt
|
||
prompt = self.prompt_manager.format_prompt(
|
||
'test_item',
|
||
function_description=function_description,
|
||
module_name=func_point['module_name']
|
||
)
|
||
|
||
# 调用API
|
||
response = self.api_client.call_api(prompt)
|
||
|
||
# 解析响应
|
||
test_items_for_func = self._parse_test_items_response(response, func_point)
|
||
test_items.extend(test_items_for_func)
|
||
|
||
except Exception as e:
|
||
# 如果生成失败，创建一个默认测试项
|
||
test_items.append({
|
||
'id': f"TI-{len(test_items) + 1:03d}",
|
||
'name': f"{func_point['function_name']}功能测试",
|
||
'module_name': func_point['module_name'],
|
||
'function_name': func_point['function_name'],
|
||
'test_type': '功能测试',
|
||
'priority': '中',
|
||
'test_objective': f"测试{func_point['function_name']}功能是否正常工作",
|
||
'error': str(e)
|
||
})
|
||
|
||
self.test_items = test_items
|
||
return test_items
|
||
|
||
def generate_test_cases(self, test_items: Optional[List[Dict[str, Any]]] = None,
|
||
progress_callback: Optional[Callable] = None) -> List[Dict[str, Any]]:
|
||
"""
|
||
为测试项生成测试用例
|
||
|
||
Args:
|
||
test_items: 测试项列表，如果为None则使用已生成的测试项
|
||
progress_callback: 进度回调函数
|
||
|
||
Returns:
|
||
测试用例列表
|
||
"""
|
||
if test_items is None:
|
||
test_items = self.test_items
|
||
|
||
if not test_items:
|
||
raise ValueError("没有可用的测试项，请先生成测试项")
|
||
|
||
test_cases = []
|
||
total = len(test_items)
|
||
|
||
for idx, test_item in enumerate(test_items):
|
||
if progress_callback:
|
||
progress_callback(idx + 1, total, f"正在生成测试用例: {test_item['name']}")
|
||
|
||
try:
|
||
# 获取功能点信息
|
||
function_points = self.parser.extract_function_points()
|
||
func_point = next(
|
||
(fp for fp in function_points if fp['function_name'] == test_item.get('function_name', '')),
|
||
None
|
||
)
|
||
|
||
if not func_point:
|
||
continue
|
||
|
||
function_description = self._build_function_description(func_point)
|
||
|
||
# 格式化prompt
|
||
prompt = self.prompt_manager.format_prompt(
|
||
'test_case',
|
||
test_item_name=test_item['name'],
|
||
test_type=test_item.get('test_type', '功能测试'),
|
||
module_name=test_item['module_name'],
|
||
function_description=function_description
|
||
)
|
||
|
||
# 调用API
|
||
response = self.api_client.call_api(prompt)
|
||
|
||
# 解析响应
|
||
cases_for_item = self._parse_test_cases_response(response, test_item)
|
||
test_cases.extend(cases_for_item)
|
||
|
||
except Exception as e:
|
||
# 如果生成失败，创建一个默认测试用例
|
||
self.test_case_counter += 1
|
||
test_cases.append({
|
||
'id': f"TC-{self.test_case_counter:03d}",
|
||
'test_item_id': test_item.get('id', ''),
|
||
'name': f"{test_item['name']} - 基础功能验证",
|
||
'module_name': test_item['module_name'],
|
||
'preconditions': '系统正常运行',
|
||
'test_steps': ['步骤1：打开功能模块', '步骤2：执行操作', '步骤3：验证结果'],
|
||
'expected_result': '功能正常工作',
|
||
'priority': test_item.get('priority', '中'),
|
||
'test_type': test_item.get('test_type', '功能测试'),
|
||
'error': str(e)
|
||
})
|
||
|
||
self.test_cases = test_cases
|
||
return test_cases
|
||
|
||
def generate_batch(self, function_points: Optional[List[Dict[str, Any]]] = None,
|
||
progress_callback: Optional[Callable] = None) -> Dict[str, List[Dict[str, Any]]]:
|
||
"""
|
||
批量生成测试项和测试用例（一次性生成）
|
||
|
||
Args:
|
||
function_points: 功能点列表
|
||
progress_callback: 进度回调函数
|
||
|
||
Returns:
|
||
包含test_items和test_cases的字典
|
||
"""
|
||
if function_points is None:
|
||
function_points = self.parser.extract_function_points()
|
||
|
||
all_test_items = []
|
||
all_test_cases = []
|
||
total = len(function_points)
|
||
self.test_case_counter = 0
|
||
|
||
for idx, func_point in enumerate(function_points):
|
||
if progress_callback:
|
||
progress_callback(idx + 1, total, f"正在生成: {func_point['function_name']}")
|
||
|
||
try:
|
||
# 转换为requirement格式（如果使用规范化Prompt）
|
||
requirement = self._convert_to_requirement(func_point)
|
||
|
||
# 尝试使用规范化Prompt（如果支持）
|
||
if self.prompt_manager.use_standards:
|
||
prompt = self.prompt_manager.format_prompt(
|
||
'batch',
|
||
requirement=requirement
|
||
)
|
||
else:
|
||
# 使用传统方式
|
||
function_description = self._build_function_description(func_point)
|
||
prompt = self.prompt_manager.format_prompt(
|
||
'batch',
|
||
function_description=function_description,
|
||
module_name=func_point['module_name']
|
||
)
|
||
|
||
# 调用API
|
||
response = self.api_client.call_api(prompt)
|
||
|
||
# 解析批量响应
|
||
result = self._parse_batch_response(response, func_point)
|
||
all_test_items.extend(result['test_items'])
|
||
all_test_cases.extend(result['test_cases'])
|
||
|
||
except Exception as e:
|
||
# 创建默认测试项和测试用例
|
||
item_id = f"TI-{len(all_test_items) + 1:03d}"
|
||
all_test_items.append({
|
||
'id': item_id,
|
||
'name': f"{func_point['function_name']}功能测试",
|
||
'module_name': func_point['module_name'],
|
||
'function_name': func_point['function_name'],
|
||
'test_type': '功能测试',
|
||
'priority': '中',
|
||
'test_objective': f"测试{func_point['function_name']}功能",
|
||
'error': str(e)
|
||
})
|
||
|
||
self.test_case_counter += 1
|
||
all_test_cases.append({
|
||
'id': f"TC-{self.test_case_counter:03d}",
|
||
'test_item_id': item_id,
|
||
'name': f"{func_point['function_name']} - 基础验证",
|
||
'module_name': func_point['module_name'],
|
||
'preconditions': '系统正常运行',
|
||
'test_steps': ['执行功能操作', '验证结果'],
|
||
'expected_result': '功能正常',
|
||
'priority': '中',
|
||
'test_type': '功能测试',
|
||
'error': str(e)
|
||
})
|
||
|
||
self.test_items = all_test_items
|
||
self.test_cases = all_test_cases
|
||
|
||
return {
|
||
'test_items': all_test_items,
|
||
'test_cases': all_test_cases
|
||
}
|
||
|
||
def _build_function_description(self, func_point: Dict[str, Any]) -> str:
|
||
"""构建功能描述文本"""
|
||
parts = []
|
||
|
||
if func_point.get('module_description'):
|
||
parts.append(f"模块描述：{func_point['module_description']}")
|
||
|
||
if func_point.get('description'):
|
||
parts.append(f"功能描述：{func_point['description']}")
|
||
|
||
if func_point.get('operation_steps'):
|
||
parts.append(f"操作步骤：{'；'.join(func_point['operation_steps'])}")
|
||
|
||
return '\n'.join(parts) if parts else func_point.get('function_name', '')
|
||
|
||
def _parse_test_items_response(self, response: str, func_point: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||
"""解析测试项API响应"""
|
||
try:
|
||
# 尝试提取JSON
|
||
json_str = self._extract_json_from_response(response)
|
||
data = json.loads(json_str)
|
||
|
||
test_items = []
|
||
for idx, item in enumerate(data.get('test_items', [])):
|
||
test_items.append({
|
||
'id': f"TI-{len(self.test_items) + len(test_items) + 1:03d}",
|
||
'name': item.get('name', ''),
|
||
'module_name': func_point['module_name'],
|
||
'function_name': func_point['function_name'],
|
||
'test_type': item.get('test_type', '功能测试'),
|
||
'priority': item.get('priority', '中'),
|
||
'test_objective': item.get('test_objective', '')
|
||
})
|
||
|
||
return test_items if test_items else [self._create_default_test_item(func_point)]
|
||
|
||
except Exception:
|
||
return [self._create_default_test_item(func_point)]
|
||
|
||
def _parse_test_cases_response(self, response: str, test_item: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||
"""解析测试用例API响应"""
|
||
try:
|
||
json_str = self._extract_json_from_response(response)
|
||
data = json.loads(json_str)
|
||
|
||
test_cases = []
|
||
for case in data.get('test_cases', []):
|
||
self.test_case_counter += 1
|
||
test_cases.append({
|
||
'id': f"TC-{self.test_case_counter:03d}",
|
||
'test_item_id': test_item.get('id', ''),
|
||
'name': case.get('name', ''),
|
||
'module_name': test_item['module_name'],
|
||
'preconditions': case.get('preconditions', ''),
|
||
'test_steps': case.get('test_steps', []),
|
||
'expected_result': case.get('expected_result', ''),
|
||
'priority': case.get('priority', test_item.get('priority', '中')),
|
||
'test_type': case.get('test_type', test_item.get('test_type', '功能测试'))
|
||
})
|
||
|
||
return test_cases if test_cases else [self._create_default_test_case(test_item)]
|
||
|
||
except Exception:
|
||
return [self._create_default_test_case(test_item)]
|
||
|
||
def _parse_batch_response(self, response: str, func_point: Dict[str, Any]) -> Dict[str, List[Dict[str, Any]]]:
|
||
"""解析批量生成API响应"""
|
||
try:
|
||
print("\n" + "-"*60)
|
||
print("[解析响应] 开始解析API响应...")
|
||
print(f"原始响应长度: {len(response)} 字符")
|
||
|
||
json_str = self._extract_json_from_response(response)
|
||
print(f"提取的JSON长度: {len(json_str)} 字符")
|
||
|
||
data = json.loads(json_str)
|
||
print(f"JSON解析成功！")
|
||
print(f" - test_items数量: {len(data.get('test_items', []))}")
|
||
|
||
test_items = []
|
||
test_cases = []
|
||
|
||
for item_data in data.get('test_items', []):
|
||
item_id = f"TI-{len(self.test_items) + len(test_items) + 1:03d}"
|
||
test_item = {
|
||
'id': item_id,
|
||
'name': item_data.get('name', ''),
|
||
'module_name': func_point['module_name'],
|
||
'function_name': func_point['function_name'],
|
||
'test_type': item_data.get('test_type', '功能测试'),
|
||
'priority': item_data.get('priority', '中'),
|
||
'test_objective': item_data.get('test_objective', '')
|
||
}
|
||
test_items.append(test_item)
|
||
print(f" ✓ 测试项: {test_item['name']}")
|
||
|
||
# 处理该测试项下的测试用例
|
||
cases_in_item = item_data.get('test_cases', [])
|
||
print(f" - 测试用例数量: {len(cases_in_item)}")
|
||
|
||
for case_data in cases_in_item:
|
||
self.test_case_counter += 1
|
||
test_cases.append({
|
||
'id': f"TC-{self.test_case_counter:03d}",
|
||
'test_item_id': item_id,
|
||
'name': case_data.get('name', ''),
|
||
'module_name': func_point['module_name'],
|
||
'preconditions': case_data.get('preconditions', ''),
|
||
'test_steps': case_data.get('test_steps', []),
|
||
'expected_result': case_data.get('expected_result', ''),
|
||
'priority': case_data.get('priority', test_item['priority']),
|
||
'test_type': case_data.get('test_type', test_item['test_type'])
|
||
})
|
||
|
||
if not test_items:
|
||
print(" ⚠️ 没有解析到test_items，使用默认值")
|
||
test_item = self._create_default_test_item(func_point)
|
||
test_items.append(test_item)
|
||
test_cases.append(self._create_default_test_case(test_item))
|
||
|
||
print(f"[解析响应] 完成！共 {len(test_items)} 个测试项，{len(test_cases)} 个测试用例")
|
||
print("-"*60 + "\n")
|
||
|
||
return {
|
||
'test_items': test_items,
|
||
'test_cases': test_cases
|
||
}
|
||
|
||
except Exception as e:
|
||
print("\n" + "-"*60)
|
||
print(f"[解析响应] ❌ 解析失败！错误: {str(e)}")
|
||
print(f"原始响应前500字符: {response[:500]}")
|
||
print("-"*60 + "\n")
|
||
|
||
test_item = self._create_default_test_item(func_point)
|
||
return {
|
||
'test_items': [test_item],
|
||
'test_cases': [self._create_default_test_case(test_item)]
|
||
}
|
||
|
||
|
||
def _extract_json_from_response(self, response: str) -> str:
|
||
"""从API响应中提取JSON字符串"""
|
||
# 尝试直接解析
|
||
try:
|
||
json.loads(response)
|
||
return response
|
||
except:
|
||
pass
|
||
|
||
# 尝试提取代码块中的JSON
|
||
json_pattern = r'```(?:json)?\s*(\{.*?\})\s*```'
|
||
match = re.search(json_pattern, response, re.DOTALL)
|
||
if match:
|
||
return match.group(1)
|
||
|
||
# 尝试提取大括号中的内容
|
||
brace_match = re.search(r'\{.*\}', response, re.DOTALL)
|
||
if brace_match:
|
||
return brace_match.group(0)
|
||
|
||
raise ValueError("无法从响应中提取JSON")
|
||
|
||
def _create_default_test_item(self, func_point: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""创建默认测试项"""
|
||
return {
|
||
'id': f"TI-{len(self.test_items) + 1:03d}",
|
||
'name': f"{func_point['function_name']}功能测试",
|
||
'module_name': func_point['module_name'],
|
||
'function_name': func_point['function_name'],
|
||
'test_type': '功能测试',
|
||
'priority': '中',
|
||
'test_objective': f"测试{func_point['function_name']}功能是否正常工作"
|
||
}
|
||
|
||
def _create_default_test_case(self, test_item: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""创建默认测试用例"""
|
||
self.test_case_counter += 1
|
||
return {
|
||
'id': f"TC-{self.test_case_counter:03d}",
|
||
'test_item_id': test_item.get('id', ''),
|
||
'name': f"{test_item['name']} - 基础功能验证",
|
||
'module_name': test_item['module_name'],
|
||
'preconditions': '系统正常运行',
|
||
'test_steps': ['步骤1：打开功能模块', '步骤2：执行操作', '步骤3：验证结果'],
|
||
'expected_result': '功能正常工作',
|
||
'priority': test_item.get('priority', '中'),
|
||
'test_type': test_item.get('test_type', '功能测试')
|
||
}
|
||
|
||
def _convert_to_requirement(self, func_point: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
将功能点转换为需求格式（用于规范化Prompt）
|
||
|
||
Args:
|
||
func_point: 功能点字典
|
||
|
||
Returns:
|
||
需求字典
|
||
"""
|
||
requirement = {
|
||
'requirement_id': func_point.get('requirement_id', func_point.get('function_name', '')),
|
||
'requirement_type': func_point.get('requirement_type', '功能需求'),
|
||
'description': func_point.get('description', func_point.get('function_name', '')),
|
||
'module_name': func_point.get('module_name', '')
|
||
}
|
||
|
||
# 如果有接口信息，添加进去
|
||
if func_point.get('interface_info'):
|
||
requirement['interface_info'] = func_point['interface_info']
|
||
|
||
return requirement
|
||
|