This commit is contained in:
2026-02-04 14:38:52 +08:00
commit a5147b1429
29 changed files with 4489 additions and 0 deletions

BIN
.gitattributes vendored Normal file

Binary file not shown.

55
.gitignore vendored Normal file
View File

@@ -0,0 +1,55 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
.venv/
ENV/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# 项目特定
uploads/
outputs/
*.json
!config/*.yaml
!config/*.json
# 配置文件(包含敏感信息)
config/api_config.yaml
# Streamlit
.streamlit/
# 系统文件
.DS_Store
Thumbs.db
# 数据文件
./需求提取
*.md
*.docx
*.xlsx

673
app.py Normal file
View File

@@ -0,0 +1,673 @@
# @line_count 500
"""测试专家系统 - Streamlit Web应用"""
import streamlit as st
import json
import os
from pathlib import Path
from typing import List, Dict, Any, Optional
import yaml
# 导入自定义模块
from modules.json_parser import JSONParser
from modules.api_client import APIClient
from modules.prompt_manager import PromptManager
from modules.test_generator import TestGenerator
from modules.output_formatter import OutputFormatter
from process_doc_file import convert_docx_to_json
# 页面配置
st.set_page_config(
page_title="测试专家系统",
page_icon="🧪",
layout="wide",
initial_sidebar_state="expanded"
)
def init_session_state():
"""初始化会话状态"""
if 'api_client' not in st.session_state:
st.session_state.api_client = None
if 'prompt_manager' not in st.session_state:
st.session_state.prompt_manager = None
if 'json_parser' not in st.session_state:
st.session_state.json_parser = None
if 'test_items' not in st.session_state:
st.session_state.test_items = []
if 'test_cases' not in st.session_state:
st.session_state.test_cases = []
if 'function_points' not in st.session_state:
st.session_state.function_points = []
if 'document_info' not in st.session_state:
st.session_state.document_info = {}
if 'standard_manager' not in st.session_state:
st.session_state.standard_manager = None
# 用户自定义Prompt支持
if 'custom_prompts' not in st.session_state:
st.session_state.custom_prompts = {} # {func_point_index: custom_prompt}
if 'custom_prompt_template' not in st.session_state:
st.session_state.custom_prompt_template = None # 批量模板
if 'use_custom_prompt' not in st.session_state:
st.session_state.use_custom_prompt = False # 是否启用自定义prompt
def load_config():
"""加载配置"""
try:
if st.session_state.api_client is None:
st.session_state.api_client = APIClient()
if st.session_state.prompt_manager is None:
st.session_state.prompt_manager = PromptManager()
except Exception as e:
st.error(f"配置加载失败: {str(e)}")
def show_home_page():
"""显示首页"""
st.title("🧪 测试专家系统")
st.markdown("---")
st.markdown("""
### 欢迎使用测试专家系统
本系统基于大模型AI技术能够自动从软件使用说明文档中提取功能点并生成全面的测试项和测试用例。
#### 主要功能
1. **文档解析**: 自动解析JSON格式的软件使用说明书提取功能点和操作步骤
2. **智能生成**: 基于大模型API为每个功能点生成专业的测试项和测试用例
3. **多格式输出**: 支持JSON、Markdown、Excel多种格式导出
4. **Prompt优化**: 支持自定义和优化测试生成Prompt模板
#### 使用流程
1. **配置API**: 在"配置"页面设置大模型API密钥和选择模型
2. **上传文档**: 在"生成测试"页面上传JSON格式的文档
3. **生成测试**: 选择功能点,点击生成按钮
4. **下载结果**: 预览生成结果,下载所需格式的文件
#### 支持的API提供商
- DeepSeek (默认)
- 通义千问
- OpenAI/ChatGPT
---
**开始使用**: 请先前往"配置"页面设置API密钥
""")
def show_config_page():
"""显示配置页面"""
st.title("⚙️ 系统配置")
st.markdown("---")
load_config()
# API配置
st.header("📡 API配置")
api_client = st.session_state.api_client
providers = api_client.config.get('providers', {})
current_provider = api_client.current_provider
# 选择API提供商
provider_options = list(providers.keys())
selected_provider = st.selectbox(
"选择API提供商",
provider_options,
index=provider_options.index(current_provider) if current_provider in provider_options else 0
)
if selected_provider != current_provider:
api_client.set_provider(selected_provider)
st.success(f"已切换到: {selected_provider}")
# 配置API密钥
provider_config = providers[selected_provider]
st.subheader(f"{selected_provider.upper()} 配置")
api_key = st.text_input(
"API密钥",
value=provider_config.get('api_key', '') or os.getenv(f'{selected_provider.upper()}_API_KEY', ''),
type="password",
help="输入API密钥或通过环境变量设置"
)
if st.button("保存API密钥"):
try:
api_client.update_api_key(selected_provider, api_key)
st.success("API密钥已保存")
except Exception as e:
st.error(f"保存失败: {str(e)}")
# 显示其他配置信息
with st.expander("查看配置详情"):
st.json(provider_config)
# 提示信息
st.info("💡 提示Prompt和测试规范配置将在上传文档后进行设置")
def show_generate_page():
"""显示生成页面"""
st.title("🚀 生成测试")
st.markdown("---")
load_config()
# 上传文档(支持 JSON 和 Word
st.header("📄 上传文档")
uploaded_file = st.file_uploader(
"选择文档文件(支持 JSON 或 Word",
type=['json', 'docx'],
help="可以直接上传Word使用说明书.docx系统会自动转换为JSON也可以上传已转换好的JSON文件"
)
if uploaded_file is not None:
try:
# 保存上传的文件
upload_dir = Path("uploads")
upload_dir.mkdir(exist_ok=True)
file_path = upload_dir / uploaded_file.name
with open(file_path, 'wb') as f:
f.write(uploaded_file.getbuffer())
# 如果是Word文档先转换为JSON
if file_path.suffix.lower() == ".docx":
json_path = upload_dir / f"{file_path.stem}.json"
convert_docx_to_json(str(file_path), str(json_path))
parser_target_path = json_path
else:
# 直接使用上传的JSON文件
parser_target_path = file_path
# 解析JSON
parser = JSONParser(str(parser_target_path))
st.session_state.json_parser = parser
# 只有在上传新文档时才清空自定义prompts检查路径是否变化
current_doc_path = str(parser_target_path)
previous_doc_path = st.session_state.get('last_uploaded_doc_path', '')
if current_doc_path != previous_doc_path:
# 新文档清空之前的自定义prompts
st.session_state.custom_prompts.clear()
st.session_state.custom_prompt_template = None
st.session_state.use_custom_prompt = False
st.session_state.last_uploaded_doc_path = current_doc_path
# 获取文档信息
doc_info = parser.get_document_info()
st.session_state.document_info = doc_info
# 显示文档信息
st.success("文档上传成功!")
col1, col2, col3 = st.columns(3)
with col1:
st.metric("文档标题", doc_info.get('title', 'N/A'))
with col2:
st.metric("章节数量", doc_info.get('section_count', 0))
with col3:
st.metric("版本", doc_info.get('version', 'N/A'))
# 提取功能点
function_points = parser.extract_function_points()
st.session_state.function_points = function_points
# 显示功能点
if function_points:
# ========== 新增Prompt/测试规范配置区域 ==========
st.markdown("---")
st.header("📝 Prompt/测试规范配置")
# 1. 需求类型统计
requirement_types = {}
for fp in function_points:
req_type = fp.get('requirement_type', '未知')
requirement_types[req_type] = requirement_types.get(req_type, 0) + 1
if requirement_types:
st.subheader("需求类型统计")
cols = st.columns(len(requirement_types))
for idx, (req_type, count) in enumerate(requirement_types.items()):
with cols[idx]:
st.metric(req_type, count)
# 2. Prompt策略选择
use_standards = st.checkbox(
"使用测试规范(推荐)",
value=True,
help="启用后,系统会根据需求类型自动选择适用的测试规范,生成符合行业标准的测试用例"
)
# 3. 测试规范选择详情(如果启用)
if use_standards:
try:
from modules.test_standard_manager import TestStandardManager
# 初始化测试规范管理器
if st.session_state.standard_manager is None:
st.session_state.standard_manager = TestStandardManager(
api_client=st.session_state.api_client
)
standard_manager = st.session_state.standard_manager
# 显示测试规范选择详情
with st.expander("📋 查看测试规范选择详情"):
for fp in function_points:
requirement = _convert_to_requirement(fp)
standards = standard_manager.get_applicable_standards(
requirement,
use_ai=False # 先不使用AI避免额外调用
)
req_type = requirement.get('requirement_type', '未知')
req_id = requirement.get('requirement_id', fp.get('function_name', ''))
st.markdown(f"**{req_id}** ({req_type})")
if standards:
st.write(f"选择的测试规范: {', '.join(standards)}")
else:
st.write("⚠️ 未找到适用的测试规范")
st.markdown("---")
# 4. Prompt预览与编辑
preview_prompt = st.checkbox("预览规范化Prompt")
if preview_prompt and function_points:
sample_fp = function_points[0]
requirement = _convert_to_requirement(sample_fp)
standards = standard_manager.get_applicable_standards(requirement, use_ai=False)
if standards:
# 生成默认prompt
default_prompt = standard_manager.build_prompt(requirement, standards)
# 获取当前索引(第一个功能点)
sample_idx = 0
# 如果没有自定义prompt使用默认值
if sample_idx not in st.session_state.custom_prompts:
st.session_state.custom_prompts[sample_idx] = default_prompt
# 可编辑的Prompt文本框
edited_prompt = st.text_area(
"Prompt预览可编辑以第一个功能点为例",
value=st.session_state.custom_prompts[sample_idx],
height=400,
key="prompt_editor_preview",
help="您可以直接修改Prompt内容修改后的内容将在生成测试时使用"
)
# 更新session state
st.session_state.custom_prompts[sample_idx] = edited_prompt
# 操作按钮
col1, col2, col3 = st.columns(3)
with col1:
if st.button("🔄 恢复默认", key="restore_default_prompt"):
st.session_state.custom_prompts[sample_idx] = default_prompt
st.session_state.use_custom_prompt = False
st.success("已恢复为默认Prompt")
st.rerun()
with col2:
if st.button("📋 应用到所有功能点", key="apply_to_all"):
# 将当前编辑的prompt应用到所有功能点
for idx in range(len(function_points)):
st.session_state.custom_prompts[idx] = edited_prompt
st.session_state.use_custom_prompt = True
st.success(f"✅ 已应用到 {len(function_points)} 个功能点")
with col3:
if st.button("🗑️ 清空所有自定义", key="clear_all_custom"):
st.session_state.custom_prompts.clear()
st.session_state.use_custom_prompt = False
st.success("已清空所有自定义Prompt")
st.rerun()
# 显示状态信息
if st.session_state.custom_prompts:
custom_count = len(st.session_state.custom_prompts)
st.info(f" 当前有 {custom_count} 个功能点使用自定义Prompt")
else:
st.warning("无法生成Prompt预览未找到适用的测试规范")
except Exception as e:
st.warning(f"测试规范功能加载失败: {str(e)}将使用传统Prompt模式")
use_standards = False
# 5. 更新PromptManager配置
if st.session_state.prompt_manager.use_standards != use_standards:
st.session_state.prompt_manager.use_standards = use_standards
if use_standards and st.session_state.standard_manager:
st.session_state.prompt_manager.standard_manager = st.session_state.standard_manager
# ========== 原有功能点选择部分 ==========
st.markdown("---")
st.header("📋 功能点列表")
# 选择功能点
selected_indices = st.multiselect(
"选择要生成测试的功能点(留空则选择全部)",
options=list(range(len(function_points))),
format_func=lambda x: f"{function_points[x]['module_name']} - {function_points[x]['function_name']}"
)
if not selected_indices:
selected_function_points = function_points
else:
selected_function_points = [function_points[i] for i in selected_indices]
# 显示功能点详情
with st.expander("查看功能点详情"):
for idx, fp in enumerate(selected_function_points):
st.markdown(f"### {idx + 1}. {fp['function_name']}")
st.markdown(f"**模块**: {fp['module_name']}")
st.markdown(f"**描述**: {fp.get('description', 'N/A')}")
if fp.get('operation_steps'):
st.markdown("**操作步骤**:")
for step in fp['operation_steps']:
st.markdown(f"- {step}")
st.markdown("---")
# 生成选项
st.markdown("---")
st.header("⚙️ 生成选项")
generation_mode = st.radio(
"生成模式",
['batch', 'separate'],
format_func=lambda x: {
'batch': '批量生成(一次性生成测试项和测试用例,更快)',
'separate': '分步生成(先生成测试项,再生成测试用例,更灵活)'
}[x]
)
# 开始生成
if st.button("🚀 开始生成测试", type="primary", use_container_width=True):
if not st.session_state.api_client.get_provider_config().get('api_key'):
st.error("请先在配置页面设置API密钥")
else:
generate_tests(selected_function_points, generation_mode)
else:
st.warning("未能从文档中提取到功能点,请检查文档格式。")
except Exception as e:
st.error(f"处理文件失败: {str(e)}")
st.exception(e)
else:
st.info("👆 请上传JSON格式的文档文件")
def generate_tests(function_points: List[Dict[str, Any]], mode: str):
"""生成测试项和测试用例"""
api_client = st.session_state.api_client
prompt_manager = st.session_state.prompt_manager
parser = st.session_state.json_parser
if parser is None:
st.error("请先上传文档")
return
# 先清空之前的自定义prompts
prompt_manager.clear_custom_prompts()
# 注入用户自定义的prompts到PromptManager
all_function_points = st.session_state.function_points
if st.session_state.custom_prompts:
st.info(f"📝 开始注入 {len(st.session_state.custom_prompts)} 个自定义Prompt")
for idx, fp in enumerate(all_function_points):
if idx in st.session_state.custom_prompts:
# 获取功能点ID与TestGenerator._convert_to_requirement保持一致
func_id = fp.get('requirement_id', fp.get('function_name', ''))
custom_prompt = st.session_state.custom_prompts[idx]
# 调试信息
st.write(f" - 索引 {idx}: {fp.get('function_name', 'N/A')}")
st.write(f" - requirement_id: `{fp.get('requirement_id', 'N/A')}`")
st.write(f" - 映射到的ID: `{func_id}`")
st.write(f" - Prompt长度: {len(custom_prompt)} 字符")
prompt_manager.set_custom_prompt(func_id, custom_prompt)
st.success(f"✅ 已加载 {len(st.session_state.custom_prompts)} 个自定义Prompt到缓存")
# 显示缓存内容
with st.expander("🔍 查看Prompt缓存详情"):
st.write("当前PromptManager缓存的自定义Prompt:")
for key, value in prompt_manager.custom_prompts_cache.items():
st.write(f"- **{key}**: {len(value)} 字符")
# 创建生成器
generator = TestGenerator(
str(parser.json_path),
api_client=api_client,
prompt_manager=prompt_manager
)
progress_bar = st.progress(0)
status_text = st.empty()
try:
if mode == 'batch':
# 批量生成
status_text.text("开始批量生成...")
result = generator.generate_batch(
function_points=function_points,
progress_callback=lambda current, total, msg: (
progress_bar.progress(current / total),
status_text.text(f"进度: {current}/{total} - {msg}")
)
)
st.session_state.test_items = result['test_items']
st.session_state.test_cases = result['test_cases']
else:
# 分步生成
status_text.text("第一步:生成测试项...")
test_items = generator.generate_test_items(
function_points=function_points,
progress_callback=lambda current, total, msg: (
progress_bar.progress(current / (total * 2)),
status_text.text(f"生成测试项: {current}/{total} - {msg}")
)
)
st.session_state.test_items = test_items
status_text.text("第二步:生成测试用例...")
test_cases = generator.generate_test_cases(
progress_callback=lambda current, total, msg: (
progress_bar.progress(0.5 + current / (total * 2)),
status_text.text(f"生成测试用例: {current}/{total} - {msg}")
)
)
st.session_state.test_cases = test_cases
progress_bar.progress(1.0)
status_text.text("生成完成!")
st.success(f"✅ 成功生成 {len(st.session_state.test_items)} 个测试项和 {len(st.session_state.test_cases)} 个测试用例")
# 跳转到结果页面
st.session_state.current_page = 'results'
st.rerun()
except Exception as e:
st.error(f"生成失败: {str(e)}")
st.exception(e)
progress_bar.empty()
status_text.empty()
def _convert_to_requirement(func_point: Dict[str, Any]) -> Dict[str, Any]:
"""
将功能点转换为需求格式(用于测试规范选择)
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
def show_results_page():
"""显示结果页面"""
st.title("📊 生成结果")
st.markdown("---")
test_items = st.session_state.test_items
test_cases = st.session_state.test_cases
document_info = st.session_state.document_info
if not test_items and not test_cases:
st.warning('还没有生成测试项和测试用例,请先前往"生成测试"页面')
return
# 统计信息
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("测试项数量", len(test_items))
with col2:
st.metric("测试用例数量", len(test_cases))
with col3:
modules = set(item.get('module_name', '') for item in test_items)
st.metric("涉及模块", len(modules))
with col4:
priorities = {}
for case in test_cases:
p = case.get('priority', '')
priorities[p] = priorities.get(p, 0) + 1
high_priority = priorities.get('', 0)
st.metric("高优先级用例", high_priority)
st.markdown("---")
# 预览
st.header("📋 预览")
preview_format = st.selectbox("选择预览格式", ['JSON', 'Markdown'])
if preview_format == 'JSON':
formatter = OutputFormatter(test_items, test_cases, document_info)
json_str = formatter.to_json()
st.json(json.loads(json_str))
else:
formatter = OutputFormatter(test_items, test_cases, document_info)
markdown_str = formatter.to_markdown()
st.markdown(markdown_str)
st.markdown("---")
# 下载
st.header("💾 下载结果")
col1, col2, col3 = st.columns(3)
formatter = OutputFormatter(test_items, test_cases, document_info)
with col1:
if st.button("下载JSON格式", use_container_width=True):
output_path = "outputs/test_results.json"
Path("outputs").mkdir(exist_ok=True)
formatter.to_json(output_path)
with open(output_path, 'rb') as f:
st.download_button(
"⬇️ 下载JSON",
f.read(),
"test_results.json",
"application/json",
key="download_json"
)
with col2:
if st.button("下载Markdown格式", use_container_width=True):
output_path = "outputs/test_results.md"
Path("outputs").mkdir(exist_ok=True)
formatter.to_markdown(output_path)
with open(output_path, 'rb') as f:
st.download_button(
"⬇️ 下载Markdown",
f.read(),
"test_results.md",
"text/markdown",
key="download_md"
)
with col3:
if st.button("下载Excel格式", use_container_width=True):
try:
output_path = "outputs/test_results.xlsx"
Path("outputs").mkdir(exist_ok=True)
formatter.to_excel(output_path)
with open(output_path, 'rb') as f:
st.download_button(
"⬇️ 下载Excel",
f.read(),
"test_results.xlsx",
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
key="download_excel"
)
except ImportError:
st.error("需要安装openpyxl: pip install openpyxl")
def main():
"""主函数"""
init_session_state()
# 侧边栏导航
with st.sidebar:
st.title("🧪 测试专家系统")
st.markdown("---")
page = st.radio(
"导航",
['home', 'config', 'generate', 'results'],
format_func=lambda x: {
'home': '🏠 首页',
'config': '⚙️ 配置',
'generate': '🚀 生成测试',
'results': '📊 结果'
}[x]
)
st.markdown("---")
st.markdown("### 使用提示")
st.info("""
1. 首次使用请先配置API密钥
2. 上传JSON格式的文档
3. 选择功能点并生成测试
4. 下载生成的结果
""")
# 路由到对应页面
if page == 'home':
show_home_page()
elif page == 'config':
show_config_page()
elif page == 'generate':
show_generate_page()
elif page == 'results':
show_results_page()
if __name__ == "__main__":
main()

113
config/default_prompt.yaml Normal file
View File

@@ -0,0 +1,113 @@
# @line_count 50
# 测试生成Prompt模板配置
# 测试项生成Prompt
test_item_prompt: |
你是一位资深的软件测试专家,擅长根据软件功能描述生成全面的测试项。
请根据以下功能描述,生成详细的测试项。每个测试项应该包含:
1. 测试项名称:清晰描述测试内容
2. 测试类型:功能测试/性能测试/安全测试/兼容性测试/易用性测试
3. 优先级:高/中/低
4. 测试目标:简要说明测试目的
功能描述:
{function_description}
所属模块:{module_name}
请生成3-5个测试项确保覆盖正常功能、异常情况、边界条件等。以JSON格式输出格式如下
{{
"test_items": [
{{
"name": "测试项名称",
"test_type": "功能测试",
"priority": "高",
"test_objective": "测试目标描述"
}}
]
}}
# 测试用例生成Prompt
test_case_prompt: |
你是一位资深的软件测试专家,擅长编写详细的测试用例。
请根据以下测试项信息,生成详细的测试用例。每个测试用例必须包含:
1. 测试用例名称
2. 前置条件
3. 详细测试步骤(步骤编号,每步清晰明确)
4. 预期结果
5. 优先级(高/中/低)
6. 测试类型
测试项信息:
测试项名称:{test_item_name}
测试类型:{test_type}
所属模块:{module_name}
功能描述:{function_description}
请生成1-3个详细的测试用例确保步骤清晰、可执行。以JSON格式输出格式如下
{{
"test_cases": [
{{
"name": "测试用例名称",
"preconditions": "前置条件描述",
"test_steps": [
"步骤1具体操作",
"步骤2具体操作",
"步骤3验证结果"
],
"expected_result": "预期结果描述",
"priority": "高",
"test_type": "功能测试"
}}
]
}}
# 批量生成Prompt一次性生成测试项和测试用例
batch_generation_prompt: |
你是一位资深的软件测试专家,擅长根据软件功能描述生成全面的测试项和测试用例。
请根据以下功能描述,生成完整的测试项和对应的测试用例。
功能描述:
{function_description}
所属模块:{module_name}
要求:
1. 生成3-5个测试项每个测试项包含
- 测试项名称
- 测试类型(功能测试/性能测试/安全测试/兼容性测试/易用性测试)
- 优先级(高/中/低)
- 测试目标
2. 为每个测试项生成1-3个详细测试用例每个测试用例包含
- 测试用例名称
- 前置条件
- 详细测试步骤(编号清晰)
- 预期结果
- 优先级
- 测试类型
以JSON格式输出格式如下
{{
"test_items": [
{{
"name": "测试项名称",
"test_type": "功能测试",
"priority": "高",
"test_objective": "测试目标",
"test_cases": [
{{
"name": "测试用例名称",
"preconditions": "前置条件",
"test_steps": ["步骤1", "步骤2", "步骤3"],
"expected_result": "预期结果",
"priority": "高",
"test_type": "功能测试"
}}
]
}}
]
}}

517
config/test_standards.yaml Normal file
View File

@@ -0,0 +1,517 @@
# @line_count 500
# 测试规范数据配置文件
# 包含14个测试类型的规范定义
test_standards:
- id: "功能测试"
name: "功能测试"
category: "primary"
summary: |
功能测试是对软件需求规格说明中的功能需求逐项进行的测试,以验证其功能是否满足要求。
主要测试点:
1. 每个功能至少一个测试用例和一个异常用例
2. 基本数据类型和数据值测试
3. 边界值测试(合法和非法边界值)
4. 操作模式、运行环境、状态转换覆盖
5. 超负荷、饱和及"最坏情况"测试
6. 不规则输入测试
7. 用户手册中的功能覆盖
key_points:
- "功能覆盖(正常+异常)"
- "边界值测试"
- "数据类型测试"
- "状态转换测试"
- "异常情况测试"
full_content: |
功能测试是对软件需求规格说明中的功能需求逐项进行的测试,以验证其功能是否满足要求。功能测试一般需进行:
1每一个软件功能应至少被一个测试用例和一个被认可的异常所覆盖对大的功能应进一步分解为更细的功能使测试用例可以直接和功能对应
2用基本数据类型和数据值测试
3)用一系列合理的数据类型和数据值运行,测试超负荷、饱和及其它"最坏情况"的结果;
4)用假想的数据类型和数据值运行,测试排斥不规则输入的能力;
5)每个功能的合法边界值和非法边界值都应被作为测试用例;
6)应考虑软件功能对操作模式、运行环境、运行状态、状态转换、运行时间等的覆盖要求;
7)对于在需求规格说明中没有指明,而在用户使用手册、操作手册中表明出来的每一功能及操作,都应有相应测试用例覆盖。
applicable_requirements:
- "功能需求"
keywords:
- "功能"
- "操作"
- "控制"
- "处理"
- id: "性能测试"
name: "性能测试"
category: "primary"
summary: |
性能测试是对软件需求规格说明中的性能需求逐项进行的测试,以验证其性能是否满足要求。
主要测试点:
1. 计算精确性(处理精度)
2. 响应时间测试
3. 数据处理量测试
4. 系统协调性测试
5. 负载潜力测试
6. 资源占用测试
key_points:
- "处理精度"
- "响应时间"
- "数据处理量"
- "系统协调性"
- "负载潜力"
full_content: |
性能测试是对软件需求规格说明中的性能需求逐项进行的测试,以验证其性能是否满足要求。性能测试一般需进行:
1)测试程序在获得定量结果时程序计算的精确性(处理精度);
2)测试程序在有速度要求时完成功能的时间(响应时间);
3)测试程序完成功能所能处理的数据量;
4)测试程序各部分的协调性,如高速、低速操作的协调:
5)测试软/硬件中因素是否限制了程序的性能;
6)测试程序的负载潜力;
7)测试程序运行占用的空间。
applicable_requirements:
- "功能需求"
- "性能需求"
keywords:
- "性能"
- "速度"
- "时间"
- "精度"
- "负载"
- "容量"
- id: "外部接口测试"
name: "外部接口测试"
category: "primary"
summary: |
外部接口测试是对软件需求规格说明中的外部接口需求逐项进行的测试。
主要测试点:
1. 测试所有外部接口,检查接口信息的格式及内容
2. 对每一个外部的输入/输出接口做正常和异常情况的测试
key_points:
- "接口格式验证"
- "接口内容验证"
- "正常情况测试"
- "异常情况测试"
full_content: |
外部接口测试是对软件需求规格说明中的外部接口需求逐项进行的测试。外部接口测试一般需进行:
1)测试所有外部接口,检查接口信息的格式及内容;
2)对每一个外部的输入/输出接口做正常和异常情况的测试。
applicable_requirements:
- "接口需求"
keywords:
- "接口"
- "输入"
- "输出"
- "通信"
- "总线"
- "CAN"
- "以太网"
- id: "人机交互界面测试"
name: "人机交互界面测试"
category: "primary"
summary: |
人机交互界面测试是对所有人机交互界面提供的操作和显示界面进行的测试,以检验是否满足用户的要求。
主要测试点:
1. 界面一致性和符合性测试
2. 非常规操作、误操作、快速操作测试
3. 错误命令和非法数据输入检测
4. 错误操作流程检测
5. 对照用户手册逐条验证
key_points:
- "界面一致性"
- "健壮性测试"
- "错误检测"
- "用户手册对照"
full_content: |
人机交互界面测试是对所有人机交互界面提供的操作和显示界面进行的测试,以检验是否满足用户的要求。人机交互界面测试一般需进行:
1)测试操作和显示界面及界面风格与软件需求规格说明中要求的一致性和符合性:
2)以非常规操作、误操作、快速操作来检验界面的健壮性;
3)测试对错误命令或非法数据输入的检测能力与提示情况;
4)测试对错误操作流程的检测与提示:
5)如果有用户手册或操作手册,应对照手册逐条进行操作和观察。
applicable_requirements:
- "功能需求"
keywords:
- "界面"
- "显示"
- "操作"
- "人机交互"
- "UI"
- id: "强度测试"
name: "强度测试"
category: "secondary"
summary: |
强度测试是强制软件运行在不正常到发生故障的情况下,检验软件可以运行到何种程度的测试。
主要测试点:
1. 性能的强度测试
2. 降级能力的强度测试
3. 系统健壮性测试
4. 系统饱和测试
key_points:
- "极限状态测试"
- "降级能力"
- "系统健壮性"
- "饱和测试"
full_content: |
强度测试是强制软件运行在不正常到发生故障的情况下(设计的极限状态到超出极限),检验软件可以运行到何种程度的测试。强度测试一般需进行:
1)性能的强度测试;
2)降级能力的强度测试;
3)系统健壮性测试;
4)系统饱和测试。
强度测试在某种程度上可看作性能测试的延伸,测出软件功能、性能的实际极限。
applicable_requirements:
- "功能需求"
- "性能需求"
keywords:
- "强度"
- "极限"
- "饱和"
- "超负荷"
- id: "余量测试"
name: "余量测试"
category: "secondary"
summary: |
余量测试是对软件是否达到需求规格说明中要求的余量的测试。若无明确要求时一般至少留有20%的余量。
主要测试点:
1. 全部存储量的余量
2. 输入、输出及通道的余量
3. 功能处理时间的余量
key_points:
- "存储余量"
- "I/O通道余量"
- "处理时间余量"
full_content: |
余量测试是对软件是否达到需求规格说明中要求的余量的测试。若无明确要求时一般至少留有20%的余量。根据测试要求,余量测试一般需提供:
1)全部存储量的余量;
2)输入、输出及通道的余量;
3)功能处理时间的余量。
applicable_requirements:
- "功能需求"
- "性能需求"
keywords:
- "余量"
- "存储"
- "通道"
- "时间"
- id: "可靠性测试"
name: "可靠性测试"
category: "secondary"
summary: |
可靠性测试是在真实的和仿真的环境中,为做出软件可靠性估计而对软件进行的功能测试。
主要测试点:
1. 测试环境与典型使用环境一致
2. 定义软件失效等级
3. 建立软件运行剖面/操作剖面
4. 详细记录失效现象和时间
5. 保证输入覆盖
6. 边界条件和环境条件测试
key_points:
- "运行剖面"
- "失效等级"
- "输入覆盖"
- "环境条件"
full_content: |
可靠性测试是在真实的和仿真的环境中,为做出软件可靠性估计而对软件进行的功能测试(其输入覆盖和环境覆盖一般大于普通的功能测试),可靠性测试中必须按照运行剖面和使用的概率分布随机地选择测试用例。可靠性测试一般需:
1)测试环境应与典型使用环境的统计特性相一致,必要时使用测试平台;
2)定义软件失效等级;
3)建立软件运行剖面/操作剖面;
4)测试记录更为详细、准确,应记录失效现象和时间;
5)必须保证输入覆盖,应覆盖重要的输入变量值、各种使用功能、相关输入变量可能组合以及不合法输入域等;
6)对于可能导致软件运行方式改变的一些边界条件和环境条件,必须进行针对性测试。
applicable_requirements:
- "功能需求"
keywords:
- "可靠性"
- "失效"
- "运行剖面"
- "环境"
- id: "安全性测试"
name: "安全性测试"
category: "primary"
summary: |
安全性测试是检验软件中已存在的安全性、安全保密性措施是否有效的测试。A、B、C级软件需要进行安全性测试。
主要测试点:
1. 软件安全性分析,危险状态测试
2. 安全性关键部件单独测试
3. 容错、冗余、中断处理测试
4. 异常条件测试
5. 硬件及软件输入故障模式测试
6. 边界、界外测试
7. 零值测试
8. 最坏情况配置测试
9. 操作员错误测试
10. 双工切换、多机替换测试
11. 防非法进入测试
key_points:
- "危险状态测试"
- "容错测试"
- "故障模式测试"
- "边界测试"
- "安全机制验证"
full_content: |
A、B、C级软件需要进行安全性测试。安全性测试是检验软件中已存在的安全性、安全保密性措施是否有效的测试。安全性测试一般
1)应进行软件安全性分析,并且在软件需求中明确每一个危险状态及导致危险的可能原因,在测试中全面检验软件在这些危险状态下的反应;
2)对安全性关键的软件部件,应单独测试,以确认该软件部件满足安全性需求:
3)对软件设计中用于提高安全性的结构、算法、容错、冗余、中断处理等方案应进行针对性测试;
4)测试应尽可能在符合实际使用的条件下进行;
5)除在正常条件下测试外,应在异常条件下测试软件,以表明不会因可能的单个或多个输入错误而导致不安全状态;
6)应包含硬件及软件输入故障模式测试:
7)应包含边界、界外及边界接合部的测试;
8)应包括"0"、穿越"0"以及从两个方向趋近于"0"的输入值;
9)应包含在最坏情况配置下的最小和最大输入数据率,以确定系统的固有能力及对这些环境的反应;
10)操作员接口测试应包括在安全性关键操作中的操作员错误,以验证安全系统对这些错误的响应;
11)应测试双工切换、多机替换的正确性和连续性;
12)应测试防止非法进入系统并保护系统数据完整性的能力。
applicable_requirements:
- "功能需求"
- "其他需求"
keywords:
- "安全"
- "危险"
- "容错"
- "故障"
- "报警"
- id: "恢复性测试"
name: "恢复性测试"
category: "secondary"
summary: |
恢复性测试是对有恢复或重置功能的软件的每一类导致恢复或重置的情况,逐一进行的测试,以验证其恢复或重置功能。
主要测试点:
1. 探测错误功能的测试
2. 切换或自动启动备用硬件测试
3. 故障时保护作业和系统状态测试
4. 从无错误状态继续执行测试
key_points:
- "错误探测"
- "备用硬件切换"
- "状态保护"
- "恢复执行"
full_content: |
恢复性测试是对有恢复或重置(reset)功能的软件的每一类导致恢复或重置的情况,逐一进行的测试,以验证其恢复或重置功能。恢复性测试是要证实在克服硬件故障后,系统能否正常地继续进行工作,且不对系统造成任何损害。恢复性测试一般需进行:
1)探测错误功能的测试;
2)能否切换或自动启动备用硬件的测试;
3)在故障发生时能否保护正在运行的作业和系统状态的测试;
4)在系统恢复后,能否从最后记录下来的无错误状态开始继续执行作业的测试。
applicable_requirements:
- "功能需求"
keywords:
- "恢复"
- "重置"
- "故障"
- "备用"
- id: "边界测试"
name: "边界测试"
category: "secondary"
summary: |
边界测试是对软件处在边界或端点情况下运行状态的测试。
主要测试点:
1. 输入域或输出域的边界或端点测试
2. 状态转换的边界或端点测试
3. 功能界限的边界或端点测试
4. 性能界限的边界或端点测试
5. 容量界限的边界或端点测试
key_points:
- "输入输出域边界"
- "状态转换边界"
- "功能界限"
- "性能界限"
- "容量界限"
full_content: |
边界测试是对软件处在边界或端点情况下运行状态的测试。边界测试一般需进行:
1)软件的输入域或输出域的边界或端点的测试;
2)状态转换的边界或端点的测试;
3)功能界限的边界或端点的测试;
4)性能界限的边界或端点的测试;
5)容量界限的边界或端点的测试。
applicable_requirements:
- "功能需求"
- "接口需求"
- "其他需求"
keywords:
- "边界"
- "端点"
- "极限"
- "范围"
- id: "安装性测试"
name: "安装性测试"
category: "secondary"
summary: |
安装性测试是对安装过程是否符合安装规程的测试,以发现安装过程中的错误。
主要测试点:
1. 不同配置下的安装和卸载测试
2. 安装规程的正确性测试
key_points:
- "安装测试"
- "卸载测试"
- "安装规程"
full_content: |
安装性测试是对安装过程是否符合安装规程的测试,以发现安装过程中的错误。安装性测试一般需进行:
1)不同配置下的安装和卸载测试;
2)安装规程的正确性的测试。
applicable_requirements:
- "其他需求"
keywords:
- "安装"
- "卸载"
- "配置"
- id: "互操作性测试"
name: "互操作性测试"
category: "secondary"
summary: |
互操作性测试是为验证不同软件之间的互操作能力而进行的测试。
主要测试点:
1. 必须同时运行两个或多个不同的软件
2. 软件之间发生互操作
key_points:
- "多软件协同"
- "互操作验证"
full_content: |
互操作性测试是为验证不同软件之间的互操作能力而进行的测试。互操作性测试一般:
1)必须同时运行两个或多个不同的软件;
2)软件之间发生互操作。
applicable_requirements:
- "接口需求"
keywords:
- "互操作"
- "协同"
- "交互"
- id: "敏感性测试"
name: "敏感性测试"
category: "secondary"
summary: |
敏感性测试是为发现在有效输入类中可能引起某种不稳定性或不正常处理的某些数据的组合而进行的测试。
主要测试点:
1. 发现引起不稳定性的数据组合
2. 发现引起不正常处理的数据组合
key_points:
- "数据组合测试"
- "不稳定性检测"
- "异常处理检测"
full_content: |
敏感性测试是为发现在有效输入类中可能引起某种不稳定性或不正常处理的某些数据的组合而进行的测试。敏感性测试一般需进行:
1)发现有效输入类中可能引起某种不稳定性的数据组合的测试;
2)发现有效输入类中可能引起某种不正常处理的数据组合的测试。
applicable_requirements:
- "功能需求"
- "其他需求"
keywords:
- "敏感"
- "不稳定"
- "数据组合"
- id: "测试充分性要求"
name: "测试充分性要求"
category: "meta"
summary: |
测试充分性要求是对测试覆盖率的总体要求。
主要要求:
1. 需求覆盖率应达到100%
2. 使用与开发相同的编译器
3. A、B级软件语句、分支覆盖率应达到100%
key_points:
- "需求覆盖率100%"
- "代码覆盖率100%"
- "编译器一致性"
full_content: |
1)对软件需求规格说明中明确和隐含的需求(包括功能、性能、接口、质量要求等)的覆盖率应达到100%;
2)配置项测试应使用与软件开发相同的编译器,全面覆盖软件需求说明文档中的所有要求。
3)对于A、B级嵌入式软件对配置项源程序测试的语句、分支覆盖率均应达到100%。对用高级语言编制的A、B级嵌入式软件应对配置项目标码进行结构分析和测试测试的目标码语句、分支覆盖率均应达到100%。对覆盖率达不到要求的软件,应对未覆盖的部分逐一进行分析和确认,并提供分析报告。
applicable_requirements:
- "功能需求"
- "接口需求"
- "其他需求"
keywords:
- "覆盖率"
- "充分性"
- "完整性"
# 需求类型到测试规范的映射规则
requirement_mapping:
功能需求:
primary:
- "功能测试"
secondary:
- "性能测试"
- "边界测试"
- "安全性测试"
conditional:
有界面:
- "人机交互界面测试"
有性能要求:
- "性能测试"
- "强度测试"
安全关键:
- "安全性测试"
接口需求:
primary:
- "外部接口测试"
secondary:
- "互操作性测试"
- "边界测试"
conditional:
多软件交互:
- "互操作性测试"
其他需求:
primary: []
secondary:
- "边界测试"
- "敏感性测试"
conditional:
性能相关:
- "性能测试"
安全相关:
- "安全性测试"
# 关键词到测试规范的映射
keyword_mapping:
接口:
- "外部接口测试"
- "互操作性测试"
性能:
- "性能测试"
- "强度测试"
- "余量测试"
界面:
- "人机交互界面测试"
安全:
- "安全性测试"
恢复:
- "恢复性测试"
边界:
- "边界测试"
安装:
- "安装性测试"
可靠性:
- "可靠性测试"
敏感:
- "敏感性测试"
显示:
- "人机交互界面测试"
总线:
- "外部接口测试"
CAN:
- "外部接口测试"
以太网:
- "外部接口测试"

3
modules/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
# @line_count 1
"""测试专家系统模块包"""

369
modules/api_client.py Normal file
View File

@@ -0,0 +1,369 @@
# @line_count 200
"""大模型API客户端模块"""
import os
import yaml
import json
import time
from pathlib import Path
from typing import Dict, Optional, List, Any
import httpx
class APIClient:
"""大模型API客户端支持多个提供商"""
def __init__(self, config_path: Optional[str] = None):
"""
初始化API客户端
Args:
config_path: API配置文件路径默认为config/api_config.yaml
"""
if config_path is None:
current_dir = Path(__file__).parent.parent
config_path = current_dir / "config" / "api_config.yaml"
self.config_path = Path(config_path)
self.config: Dict[str, Any] = {}
self.current_provider: str = ""
self.load_config()
def load_config(self):
"""加载API配置"""
if not self.config_path.exists():
raise FileNotFoundError(f"API配置文件不存在: {self.config_path}")
with open(self.config_path, 'r', encoding='utf-8') as f:
self.config = yaml.safe_load(f)
self.current_provider = self.config.get('default_provider', 'deepseek')
def set_provider(self, provider: str):
"""
设置当前使用的API提供商
Args:
provider: 提供商名称 (deepseek, qianwen, openai, openrouter)
"""
if provider not in self.config.get('providers', {}):
raise ValueError(f"不支持的API提供商: {provider}")
self.current_provider = provider
def get_provider_config(self) -> Dict[str, Any]:
"""获取当前提供商的配置"""
providers = self.config.get('providers', {})
if self.current_provider not in providers:
raise ValueError(f"提供商配置不存在: {self.current_provider}")
return providers[self.current_provider]
def _call_deepseek_api(self, prompt: str, **kwargs) -> str:
"""调用DeepSeek API"""
config = self.get_provider_config()
api_key = config.get('api_key') or os.getenv('DEEPSEEK_API_KEY', '')
if not api_key:
raise ValueError("DeepSeek API密钥未配置")
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {api_key}'
}
data = {
'model': config.get('model', 'deepseek-chat'),
'messages': [
{'role': 'user', 'content': prompt}
],
'temperature': config.get('temperature', 0.7),
'max_tokens': config.get('max_tokens', 4000)
}
# 使用精细化超时配置连接10秒读取180秒
timeout_config = httpx.Timeout(connect=10.0, read=180.0, write=10.0, pool=10.0)
with httpx.Client(timeout=timeout_config) as client:
response = client.post(
config.get('base_url'),
headers=headers,
json=data
)
response.raise_for_status()
result = response.json()
return result['choices'][0]['message']['content']
# def _call_qianwen_api(self, prompt: str, **kwargs) -> str:
# """调用通义千问API"""
# config = self.get_provider_config()
# api_key = config.get('api_key') or os.getenv('QIANWEN_API_KEY', '')
# if not api_key:
# raise ValueError("通义千问API密钥未配置")
# headers = {
# 'Content-Type': 'application/json',
# 'Authorization': f'Bearer {api_key}'
# }
# # 通义千问API格式DashScope
# data = {
# 'model': config.get('model', 'qwen-turbo'),
# 'input': {
# 'messages': [
# {'role': 'user', 'content': prompt}
# ]
# },
# 'parameters': {
# 'temperature': config.get('temperature', 0.7),
# 'max_tokens': config.get('max_tokens', 4000)
# }
# }
# with httpx.Client(timeout=60.0) as client:
# response = client.post(
# config.get('base_url'),
# headers=headers,
# json=data
# )
# response.raise_for_status()
# result = response.json()
# # 通义千问的响应格式适配
# if 'output' in result:
# output = result['output']
# if 'choices' in output and len(output['choices']) > 0:
# return output['choices'][0]['message']['content']
# elif 'text' in output:
# return output['text']
# elif 'choices' in result and len(result['choices']) > 0:
# return result['choices'][0]['message']['content']
# # 如果都不匹配,返回整个结果(用于调试)
# raise ValueError(f"无法解析通义千问API响应: {result}")
def _call_qianwen_api(self, prompt: str, **kwargs) -> str:
"""调用通义千问API兼容模式"""
config = self.get_provider_config()
api_key = config.get('api_key') or os.getenv('QIANWEN_API_KEY', '')
if not api_key:
raise ValueError("通义千问API密钥未配置")
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {api_key}'
}
# 系统提示提高生成质量模拟Web端行为
# 针对航空航天软件测试的专业化system message
system_message = """你是一位拥有20年经验的航空航天软件测试专家。你的任务是生成高质量、可执行的测试用例。
【强制要求 - 必须遵守】
你必须为以下8种黑盒测试方法各生成至少1个测试用例
1. 等价类划分 - 有效/无效输入类测试
2. 边界值分析 - min-1, min, min+1, max-1, max, max+1
3. 错误推测法 - 字节序错误、超长帧、非法字符
4. 因果图法 - 多条件联动触发测试
5. 决策表测试 - 条件组合规则覆盖
6. 状态转换法 - 状态机路径覆盖
7. 场景法 - 端到端业务流程测试
8. 随机测试 - 随机合法输入测试
【输出要求】
- 每个test_case的name字段必须用方括号标注测试方法正常指令解析-[等价类划分]
- 必须生成至少8个测试用例确保8种方法全覆盖
- 测试步骤要具体包含CAN帧ID、指令编码、响应时间等
- 只输出JSON不要任何解释性文字
- 严格遵循用户指定的JSON格式"""
# 使用OpenAI兼容格式包含system message
data = {
'model': config.get('model', 'qwen-max'),
'messages': [
{'role': 'system', 'content': system_message},
{'role': 'user', 'content': prompt}
],
'temperature': config.get('temperature', 0.3),
'max_tokens': config.get('max_tokens', 8192)
}
print("\n发送Prompt: " + prompt)
# 使用精细化超时配置连接10秒读取180秒3分钟
# 测试用例生成需要较长时间,特别是复杂的结构化输出
timeout_config = httpx.Timeout(
connect=10.0, # 连接超时10秒
read=180.0, # 读取超时180秒适合复杂生成任务
write=10.0, # 写入超时10秒
pool=10.0 # 连接池超时10秒
)
with httpx.Client(timeout=timeout_config) as client:
response = client.post(
config.get('base_url'),
headers=headers,
json=data
)
response.raise_for_status()
result = response.json()
# OpenAI兼容格式的响应
if 'choices' in result and len(result['choices']) > 0:
content = result['choices'][0]['message']['content']
# 调试输出显示API返回的完整内容
print("\n" + "="*60)
print("[API响应] 完整返回内容:")
print("="*60)
print(f"响应长度: {len(content)} 字符")
print("-"*60)
# 显示前2000字符
if len(content) > 2000:
print(content[:2000])
print(f"\n... [截断,还有 {len(content)-2000} 字符]")
else:
print(content)
print("="*60 + "\n")
return content
raise ValueError(f"无法解析通义千问API响应: {result}")
def _call_openai_api(self, prompt: str, **kwargs) -> str:
"""调用OpenAI API"""
config = self.get_provider_config()
api_key = config.get('api_key') or os.getenv('OPENAI_API_KEY', '')
if not api_key:
raise ValueError("OpenAI API密钥未配置")
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {api_key}'
}
data = {
'model': config.get('model', 'gpt-3.5-turbo'),
'messages': [
{'role': 'user', 'content': prompt}
],
'temperature': config.get('temperature', 0.7),
'max_tokens': config.get('max_tokens', 4000)
}
# 使用精细化超时配置连接10秒读取180秒
timeout_config = httpx.Timeout(connect=10.0, read=180.0, write=10.0, pool=10.0)
with httpx.Client(timeout=timeout_config) as client:
response = client.post(
config.get('base_url'),
headers=headers,
json=data
)
response.raise_for_status()
result = response.json()
return result['choices'][0]['message']['content']
def _call_openrouter_api(self, prompt: str, **kwargs) -> str:
"""调用OpenRouter API"""
config = self.get_provider_config()
api_key = config.get('api_key') or os.getenv('OPENROUTER_API_KEY', '')
if not api_key:
raise ValueError("OpenRouter API密钥未配置")
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {api_key}'
}
# 添加可选的 HTTP-Referer 和 X-Title headers
http_referer = config.get('http_referer', '')
if http_referer:
headers['HTTP-Referer'] = http_referer
x_title = config.get('x_title', '')
if x_title:
headers['X-Title'] = x_title
data = {
'model': config.get('model', 'allenai/molmo-2-8b:free'),
'messages': [
{'role': 'user', 'content': prompt}
],
'temperature': config.get('temperature', 0.7),
'max_tokens': config.get('max_tokens', 4000)
}
# 使用精细化超时配置连接10秒读取180秒
timeout_config = httpx.Timeout(connect=10.0, read=180.0, write=10.0, pool=10.0)
with httpx.Client(timeout=timeout_config) as client:
response = client.post(
config.get('base_url'),
headers=headers,
json=data
)
response.raise_for_status()
result = response.json()
# OpenRouter 使用 OpenAI 兼容格式
if 'choices' in result and len(result['choices']) > 0:
return result['choices'][0]['message']['content']
raise ValueError(f"无法解析OpenRouter API响应: {result}")
def call_api(self, prompt: str, max_retries: int = 3, retry_delay: int = 2) -> str:
"""
调用大模型API
Args:
prompt: 提示词
max_retries: 最大重试次数
retry_delay: 重试延迟(秒)
Returns:
API返回的文本内容
"""
provider_methods = {
'deepseek': self._call_deepseek_api,
'qianwen': self._call_qianwen_api,
'openai': self._call_openai_api,
'openrouter': self._call_openrouter_api
}
method = provider_methods.get(self.current_provider)
if not method:
raise ValueError(f"不支持的API提供商: {self.current_provider}")
last_error = None
for attempt in range(max_retries):
try:
return method(prompt)
except Exception as e:
last_error = e
if attempt < max_retries - 1:
time.sleep(retry_delay * (attempt + 1))
else:
raise Exception(f"API调用失败重试{max_retries}次): {str(e)}") from last_error
raise Exception(f"API调用失败: {str(last_error)}")
def update_api_key(self, provider: str, api_key: str):
"""
更新API密钥
Args:
provider: 提供商名称
api_key: API密钥
"""
if provider not in self.config.get('providers', {}):
raise ValueError(f"不支持的API提供商: {provider}")
self.config['providers'][provider]['api_key'] = api_key
# 保存到配置文件
with open(self.config_path, 'w', encoding='utf-8') as f:
yaml.dump(self.config, f, allow_unicode=True, default_flow_style=False)

76
modules/json_parser.py Normal file
View File

@@ -0,0 +1,76 @@
# @line_count 100
"""JSON解析模块提取功能点和章节信息"""
import json
from pathlib import Path
from typing import List, Dict, Any, Optional
from .parser_adapters.parser_factory import ParserFactory
class JSONParser:
"""JSON文档解析器门面模式"""
def __init__(self, json_path: str):
"""
初始化JSON解析器
Args:
json_path: JSON文件路径
"""
self.json_path = Path(json_path)
self.data: Dict[str, Any] = {}
self.adapter = None
self.load_json()
def load_json(self):
"""加载JSON文件并创建适配器"""
if not self.json_path.exists():
raise FileNotFoundError(f"JSON文件不存在: {self.json_path}")
with open(self.json_path, 'r', encoding='utf-8') as f:
self.data = json.load(f)
# 自动检测格式并创建适配器
self.adapter = ParserFactory.create_adapter(self.data)
def extract_function_points(self) -> List[Dict[str, Any]]:
"""
提取功能点列表
Returns:
功能点列表,每个功能点包含:
- module_name: 所属模块
- function_name: 功能名称
- description: 功能描述
- operation_steps: 操作步骤(可选)
- requirement_id: 需求编号(可选)
- requirement_type: 需求类型(可选)
"""
return self.adapter.extract_function_points()
def get_document_info(self) -> Dict[str, Any]:
"""
获取文档基本信息
Returns:
文档信息字典
"""
return self.adapter.get_document_info()
def get_sections(self) -> List[Dict[str, Any]]:
"""
获取所有章节信息
Returns:
章节列表
"""
return self.adapter.get_sections()
def get_module_summary(self) -> List[Dict[str, Any]]:
"""
获取模块摘要
Returns:
模块摘要列表
"""
return self.adapter.get_module_summary()

332
modules/output_formatter.py Normal file
View File

@@ -0,0 +1,332 @@
# @line_count 300
"""多格式输出模块"""
import json
from pathlib import Path
from typing import List, Dict, Any, Optional
from datetime import datetime
try:
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
OPENPYXL_AVAILABLE = True
except ImportError:
OPENPYXL_AVAILABLE = False
class OutputFormatter:
"""多格式输出格式化器"""
def __init__(self, test_items: List[Dict[str, Any]], test_cases: List[Dict[str, Any]],
document_info: Optional[Dict[str, Any]] = None):
"""
初始化输出格式化器
Args:
test_items: 测试项列表
test_cases: 测试用例列表
document_info: 文档信息
"""
self.test_items = test_items
self.test_cases = test_cases
self.document_info = document_info or {}
def to_json(self, output_path: Optional[str] = None) -> str:
"""
输出为JSON格式
Args:
output_path: 输出文件路径如果为None则返回JSON字符串
Returns:
JSON字符串或文件路径
"""
data = {
'document_info': self.document_info,
'generation_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'summary': {
'test_item_count': len(self.test_items),
'test_case_count': len(self.test_cases)
},
'test_items': self.test_items,
'test_cases': self.test_cases
}
json_str = json.dumps(data, ensure_ascii=False, indent=2)
if output_path:
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(json_str)
return str(output_path)
return json_str
def to_markdown(self, output_path: Optional[str] = None) -> str:
"""
输出为Markdown格式
Args:
output_path: 输出文件路径如果为None则返回Markdown字符串
Returns:
Markdown字符串或文件路径
"""
lines = []
# 标题和文档信息
lines.append("# 测试项和测试用例文档\n")
if self.document_info.get('title'):
lines.append(f"**文档标题**: {self.document_info['title']}\n")
if self.document_info.get('version'):
lines.append(f"**版本**: {self.document_info['version']}\n")
lines.append(f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
lines.append(f"**测试项数量**: {len(self.test_items)}\n")
lines.append(f"**测试用例数量**: {len(self.test_cases)}\n")
lines.append("\n---\n")
# 按模块分组
modules = {}
for item in self.test_items:
module_name = item.get('module_name', '未分类')
if module_name not in modules:
modules[module_name] = []
modules[module_name].append(item)
# 输出每个模块的测试项和测试用例
for module_name, items in modules.items():
lines.append(f"\n## {module_name}\n")
for item in items:
# 测试项
lines.append(f"\n### {item.get('id', '')} {item.get('name', '')}\n")
lines.append(f"- **测试类型**: {item.get('test_type', 'N/A')}")
lines.append(f"- **优先级**: {item.get('priority', 'N/A')}")
lines.append(f"- **测试目标**: {item.get('test_objective', 'N/A')}\n")
# 该测试项下的测试用例
item_cases = [case for case in self.test_cases
if case.get('test_item_id') == item.get('id')]
if item_cases:
lines.append("#### 测试用例\n")
for case in item_cases:
lines.append(f"\n**{case.get('id', '')} {case.get('name', '')}**\n")
lines.append(f"- **前置条件**: {case.get('preconditions', 'N/A')}")
lines.append(f"- **优先级**: {case.get('priority', 'N/A')}")
lines.append(f"- **测试类型**: {case.get('test_type', 'N/A')}\n")
lines.append("**测试步骤**:\n")
for idx, step in enumerate(case.get('test_steps', []), 1):
lines.append(f"{idx}. {step}\n")
lines.append(f"\n**预期结果**: {case.get('expected_result', 'N/A')}\n")
lines.append("---\n")
markdown_str = '\n'.join(lines)
if output_path:
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(markdown_str)
return str(output_path)
return markdown_str
def to_excel(self, output_path: str) -> str:
"""
输出为Excel格式
Args:
output_path: 输出文件路径
Returns:
文件路径
"""
if not OPENPYXL_AVAILABLE:
raise ImportError("openpyxl未安装无法生成Excel文件。请运行: pip install openpyxl")
wb = Workbook()
# 删除默认工作表
if 'Sheet' in wb.sheetnames:
wb.remove(wb['Sheet'])
# 创建测试项工作表
self._create_test_items_sheet(wb)
# 创建测试用例工作表
self._create_test_cases_sheet(wb)
# 创建摘要工作表
self._create_summary_sheet(wb)
# 保存文件
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
wb.save(output_path)
return str(output_path)
def _create_test_items_sheet(self, wb: Workbook):
"""创建测试项工作表"""
ws = wb.create_sheet("测试项", 0)
# 表头
headers = ['测试项ID', '测试项名称', '所属模块', '功能名称', '测试类型', '优先级', '测试目标']
ws.append(headers)
# 设置表头样式
header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
header_font = Font(bold=True, color="FFFFFF")
border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
for col in range(1, len(headers) + 1):
cell = ws.cell(1, col)
cell.fill = header_fill
cell.font = header_font
cell.border = border
cell.alignment = Alignment(horizontal='center', vertical='center')
# 数据行
for item in self.test_items:
row = [
item.get('id', ''),
item.get('name', ''),
item.get('module_name', ''),
item.get('function_name', ''),
item.get('test_type', ''),
item.get('priority', ''),
item.get('test_objective', '')
]
ws.append(row)
# 设置边框
for col in range(1, len(headers) + 1):
ws.cell(ws.max_row, col).border = border
# 调整列宽
column_widths = [12, 30, 15, 15, 12, 8, 40]
for col, width in enumerate(column_widths, 1):
ws.column_dimensions[ws.cell(1, col).column_letter].width = width
# 冻结首行
ws.freeze_panes = 'A2'
def _create_test_cases_sheet(self, wb: Workbook):
"""创建测试用例工作表"""
ws = wb.create_sheet("测试用例", 1)
# 表头
headers = ['测试用例ID', '测试项ID', '测试用例名称', '所属模块', '前置条件',
'测试步骤', '预期结果', '优先级', '测试类型']
ws.append(headers)
# 设置表头样式
header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
header_font = Font(bold=True, color="FFFFFF")
border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
for col in range(1, len(headers) + 1):
cell = ws.cell(1, col)
cell.fill = header_fill
cell.font = header_font
cell.border = border
cell.alignment = Alignment(horizontal='center', vertical='center')
# 数据行
for case in self.test_cases:
test_steps_str = '\n'.join([f"{idx}. {step}" for idx, step
in enumerate(case.get('test_steps', []), 1)])
row = [
case.get('id', ''),
case.get('test_item_id', ''),
case.get('name', ''),
case.get('module_name', ''),
case.get('preconditions', ''),
test_steps_str,
case.get('expected_result', ''),
case.get('priority', ''),
case.get('test_type', '')
]
ws.append(row)
# 设置边框和换行
for col in range(1, len(headers) + 1):
cell = ws.cell(ws.max_row, col)
cell.border = border
if col == 6: # 测试步骤列
cell.alignment = Alignment(wrap_text=True, vertical='top')
else:
cell.alignment = Alignment(vertical='top')
# 调整列宽
column_widths = [12, 12, 30, 15, 25, 40, 30, 8, 12]
for col, width in enumerate(column_widths, 1):
ws.column_dimensions[ws.cell(1, col).column_letter].width = width
# 设置行高(为测试步骤列预留空间)
for row in range(2, ws.max_row + 1):
ws.row_dimensions[row].height = 60
# 冻结首行
ws.freeze_panes = 'A2'
def _create_summary_sheet(self, wb: Workbook):
"""创建摘要工作表"""
ws = wb.create_sheet("摘要", 2)
# 文档信息
if self.document_info.get('title'):
ws.append(['文档标题', self.document_info['title']])
if self.document_info.get('version'):
ws.append(['版本', self.document_info['version']])
ws.append(['生成时间', datetime.now().strftime('%Y-%m-%d %H:%M:%S')])
ws.append([])
# 统计信息
ws.append(['统计项', '数量'])
ws.append(['测试项总数', len(self.test_items)])
ws.append(['测试用例总数', len(self.test_cases)])
ws.append([])
# 按模块统计
ws.append(['模块', '测试项数量', '测试用例数量'])
modules = {}
for item in self.test_items:
module_name = item.get('module_name', '未分类')
if module_name not in modules:
modules[module_name] = {'items': 0, 'cases': 0}
modules[module_name]['items'] += 1
for case in self.test_cases:
module_name = case.get('module_name', '未分类')
if module_name not in modules:
modules[module_name] = {'items': 0, 'cases': 0}
modules[module_name]['cases'] += 1
for module_name, stats in modules.items():
ws.append([module_name, stats['items'], stats['cases']])
# 设置样式
header_font = Font(bold=True)
for row in ws.iter_rows(min_row=1, max_row=1):
for cell in row:
cell.font = header_font
# 调整列宽
ws.column_dimensions['A'].width = 20
ws.column_dimensions['B'].width = 15
ws.column_dimensions['C'].width = 15

View File

@@ -0,0 +1,13 @@
# @line_count 50
"""JSON解析器适配器包"""
from .base_adapter import BaseParserAdapter
from .parser_factory import ParserFactory
from .section_array_adapter import SectionArrayAdapter
from .requirement_tree_adapter import RequirementTreeAdapter
__all__ = [
'BaseParserAdapter',
'ParserFactory',
'SectionArrayAdapter',
'RequirementTreeAdapter',
]

View File

@@ -0,0 +1,81 @@
# @line_count 100
"""JSON解析器适配器基类"""
from abc import ABC, abstractmethod
from typing import List, Dict, Any
class BaseParserAdapter(ABC):
"""解析器适配器抽象基类"""
def __init__(self, data: Dict[str, Any]):
"""
初始化适配器
Args:
data: 解析后的JSON数据
"""
self.data = data
@abstractmethod
def extract_function_points(self) -> List[Dict[str, Any]]:
"""
提取功能点列表
Returns:
功能点列表,统一格式:
- module_name: 所属模块
- function_name: 功能名称
- description: 功能描述
- operation_steps: 操作步骤(可选)
- requirement_id: 需求编号(可选)
- requirement_type: 需求类型(可选)
"""
pass
@abstractmethod
def get_document_info(self) -> Dict[str, Any]:
"""
获取文档基本信息
Returns:
文档信息字典,统一格式:
- title: 文档标题
- version: 版本(可选)
- date: 日期(可选)
- section_count: 章节数量
"""
pass
@abstractmethod
def get_sections(self) -> List[Dict[str, Any]]:
"""
获取所有章节信息
Returns:
章节列表
"""
pass
@abstractmethod
def get_module_summary(self) -> List[Dict[str, Any]]:
"""
获取模块摘要
Returns:
模块摘要列表
"""
pass
@staticmethod
@abstractmethod
def can_parse(data: Dict[str, Any]) -> bool:
"""
检测是否能解析该格式
Args:
data: JSON数据
Returns:
是否能解析
"""
pass

View File

@@ -0,0 +1,53 @@
# @line_count 100
"""解析器工厂"""
from typing import Dict, Any, Type, List
from .base_adapter import BaseParserAdapter
from .section_array_adapter import SectionArrayAdapter
from .requirement_tree_adapter import RequirementTreeAdapter
class ParserFactory:
"""解析器工厂,自动检测格式并创建合适的适配器"""
# 注册所有适配器(按优先级排序)
_adapters: List[Type[BaseParserAdapter]] = [
RequirementTreeAdapter, # 新格式优先
SectionArrayAdapter, # 旧格式
# 未来可以在这里添加更多适配器
]
@classmethod
def create_adapter(cls, data: Dict[str, Any]) -> BaseParserAdapter:
"""
创建合适的适配器
Args:
data: JSON数据
Returns:
适配器实例
Raises:
ValueError: 如果无法识别格式
"""
for adapter_class in cls._adapters:
if adapter_class.can_parse(data):
return adapter_class(data)
raise ValueError(
"无法识别JSON格式。支持的格式\n"
"- 需求树格式(包含'需求内容''文档元数据'\n"
"- 章节数组格式(包含'sections'数组)"
)
@classmethod
def register_adapter(cls, adapter_class: Type[BaseParserAdapter],
priority: int = 0):
"""
注册新的适配器
Args:
adapter_class: 适配器类
priority: 优先级越小越优先0为最高优先级
"""
cls._adapters.insert(priority, adapter_class)

View File

@@ -0,0 +1,139 @@
# @line_count 200
"""新格式适配器(需求树格式)"""
from typing import List, Dict, Any
from .base_adapter import BaseParserAdapter
class RequirementTreeAdapter(BaseParserAdapter):
"""处理新格式:需求树结构"""
def extract_function_points(self) -> List[Dict[str, Any]]:
"""从需求树中提取功能点"""
function_points = []
requirement_content = self.data.get('需求内容', {})
# 递归遍历章节
self._traverse_requirements(requirement_content, [], function_points)
return function_points
def _traverse_requirements(self, sections: Dict, path: List[str],
function_points: List[Dict]):
"""递归遍历章节,提取需求"""
for section_key, section_data in sections.items():
section_info = section_data.get('章节信息', {})
section_title = section_info.get('章节标题', '')
current_path = path + [section_title]
# 如果有需求列表,提取需求
if '需求列表' in section_data:
requirements = section_data['需求列表']
for req in requirements:
function_points.append({
'module_name': ' > '.join(current_path[:-1]) if len(current_path) > 1 else section_title,
'function_name': req.get('需求编号', ''),
'description': req.get('需求描述', ''),
'requirement_id': req.get('需求编号', ''),
'requirement_type': self._parse_requirement_type(
req.get('需求编号', '')
),
'interface_info': {
k: v for k, v in req.items()
if k in ['接口名称', '接口类型', '来源', '目的地']
} if any(k in req for k in ['接口名称', '接口类型']) else None
})
# 递归处理子章节
if '子章节' in section_data:
self._traverse_requirements(
section_data['子章节'],
current_path,
function_points
)
def _parse_requirement_type(self, req_id: str) -> str:
"""从需求编号解析需求类型"""
if req_id.startswith('FR-'):
return '功能需求'
elif req_id.startswith('IR-'):
return '接口需求'
elif req_id.startswith('OR-'):
return '其他需求'
return '未知'
def get_document_info(self) -> Dict[str, Any]:
"""获取文档信息"""
metadata = self.data.get('文档元数据', {})
return {
'title': metadata.get('标题', ''),
'version': '', # 新格式可能没有版本
'date': metadata.get('生成时间', ''),
'section_count': self._count_sections(self.data.get('需求内容', {}))
}
def _count_sections(self, sections: Dict) -> int:
"""递归统计章节数量"""
count = 0
for section_data in sections.values():
count += 1
if '子章节' in section_data:
count += self._count_sections(section_data['子章节'])
return count
def get_sections(self) -> List[Dict[str, Any]]:
"""获取章节列表(转换为扁平结构)"""
sections = []
self._flatten_sections(self.data.get('需求内容', {}), sections)
return sections
def _flatten_sections(self, sections: Dict, result: List[Dict]):
"""递归扁平化章节结构"""
for section_data in sections.values():
section_info = section_data.get('章节信息', {})
result.append({
'title': section_info.get('章节标题', ''),
'number': section_info.get('章节编号', ''),
'level': section_info.get('章节级别', 0),
'requirement_count': len(section_data.get('需求列表', []))
})
if '子章节' in section_data:
self._flatten_sections(section_data['子章节'], result)
def get_module_summary(self) -> List[Dict[str, Any]]:
"""获取模块摘要"""
modules = []
sections_dict = {}
# 遍历需求内容,构建模块统计
self._build_module_summary(self.data.get('需求内容', {}), sections_dict)
for module_name, info in sections_dict.items():
modules.append({
'name': module_name,
'function_count': info['requirement_count'],
'description': info.get('description', '')
})
return modules
def _build_module_summary(self, sections: Dict, result: Dict):
"""递归构建模块摘要"""
for section_data in sections.values():
section_info = section_data.get('章节信息', {})
section_title = section_info.get('章节标题', '')
if '需求列表' in section_data:
if section_title not in result:
result[section_title] = {
'requirement_count': 0,
'description': ''
}
result[section_title]['requirement_count'] += len(section_data['需求列表'])
if '子章节' in section_data:
self._build_module_summary(section_data['子章节'], result)
@staticmethod
def can_parse(data: Dict[str, Any]) -> bool:
"""检测是否为新格式"""
return '需求内容' in data and '文档元数据' in data

View File

@@ -0,0 +1,164 @@
# @line_count 150
"""旧格式适配器sections数组格式"""
from typing import List, Dict, Any
from .base_adapter import BaseParserAdapter
class SectionArrayAdapter(BaseParserAdapter):
"""处理旧格式sections数组"""
def extract_function_points(self) -> List[Dict[str, Any]]:
"""从sections数组中提取功能点"""
function_points = []
sections = self.data.get('sections', [])
for section in sections:
module_name = section.get('title', '')
content = section.get('content', [])
# 提取模块总体描述(第一个较长的文本内容)
module_description = ""
for item in content:
if item.get('type') == 'text':
text = item.get('content', '').strip()
if len(text) > 50: # 较长的文本通常是模块描述
module_description = text
break
# 识别功能点
# 功能点通常是较短的文本(标题),后面跟着描述
current_function = None
function_description_parts = []
for i, item in enumerate(content):
if item.get('type') != 'text':
continue
text = item.get('content', '').strip()
if not text:
continue
# 判断是否是功能点标题
# 规则:短文本(通常<20字符且不是描述性文本
is_function_title = (
len(text) < 20 and
not text.endswith('') and
not text.endswith('') and
not ('如下' in text or '所示' in text)
)
if is_function_title:
# 保存之前的功能点
if current_function:
function_points.append({
'module_name': module_name,
'module_description': module_description,
'function_name': current_function,
'description': ' '.join(function_description_parts),
'operation_steps': self._extract_steps(function_description_parts)
})
# 开始新功能点
current_function = text
function_description_parts = []
else:
# 添加到当前功能点的描述
if current_function:
function_description_parts.append(text)
elif not module_description:
# 如果还没有模块描述,这可能是模块描述的一部分
pass
# 保存最后一个功能点
if current_function:
function_points.append({
'module_name': module_name,
'module_description': module_description,
'function_name': current_function,
'description': ' '.join(function_description_parts),
'operation_steps': self._extract_steps(function_description_parts)
})
# 如果没有识别到功能点,将整个模块作为一个功能点
if not current_function and module_description:
function_points.append({
'module_name': module_name,
'module_description': module_description,
'function_name': module_name,
'description': module_description,
'operation_steps': []
})
return function_points
def _extract_steps(self, description_parts: List[str]) -> List[str]:
"""
从描述中提取操作步骤
Args:
description_parts: 描述文本列表
Returns:
操作步骤列表
"""
steps = []
for part in description_parts:
# 查找包含操作动词的句子
if any(keyword in part for keyword in ['点击', '选择', '输入', '打开', '关闭', '设置', '查看']):
# 移除"如下图所示"等描述性文字
cleaned = part.replace('如下图所示', '').replace('如下图所示:', '').strip()
if cleaned:
steps.append(cleaned)
return steps
def get_document_info(self) -> Dict[str, Any]:
"""获取文档信息"""
return {
'title': self.data.get('document_title', ''),
'version': self.data.get('version', ''),
'date': self.data.get('date', ''),
'section_count': len(self.data.get('sections', []))
}
def get_sections(self) -> List[Dict[str, Any]]:
"""获取章节列表"""
return self.data.get('sections', [])
def get_module_summary(self) -> List[Dict[str, Any]]:
"""获取模块摘要"""
modules = []
sections = self.data.get('sections', [])
for section in sections:
module_info = {
'name': section.get('title', ''),
'function_count': 0,
'description': ''
}
# 查找模块描述
content = section.get('content', [])
for item in content:
if item.get('type') == 'text':
text = item.get('content', '').strip()
if len(text) > 50:
module_info['description'] = text
break
# 统计功能点数量(简单统计)
function_names = []
for item in content:
if item.get('type') == 'text':
text = item.get('content', '').strip()
if len(text) < 20 and text and not text.endswith(''):
function_names.append(text)
module_info['function_count'] = len(set(function_names))
modules.append(module_info)
return modules
@staticmethod
def can_parse(data: Dict[str, Any]) -> bool:
"""检测是否为旧格式"""
return 'sections' in data and isinstance(data['sections'], list)

178
modules/prompt_manager.py Normal file
View File

@@ -0,0 +1,178 @@
# @line_count 150
"""Prompt模板管理模块"""
import os
import yaml
from pathlib import Path
from typing import Dict, Optional
from .test_standard_manager import TestStandardManager
class PromptManager:
"""Prompt模板管理器"""
def __init__(self, config_path: Optional[str] = None,
use_standards: bool = True,
standard_manager: Optional[TestStandardManager] = None):
"""
初始化Prompt管理器
Args:
config_path: Prompt配置文件路径默认为config/default_prompt.yaml
use_standards: 是否使用测试规范默认True
standard_manager: 测试规范管理器如果为None且use_standards=True则创建新实例
"""
if config_path is None:
# 获取项目根目录
current_dir = Path(__file__).parent.parent
config_path = current_dir / "config" / "default_prompt.yaml"
self.config_path = Path(config_path)
self.prompts: Dict[str, str] = {}
self.use_standards = use_standards
self.standard_manager = standard_manager or (TestStandardManager() if use_standards else None)
# 用户自定义Prompt缓存 {requirement_id或func_point_id: custom_prompt}
self.custom_prompts_cache: Dict[str, str] = {}
self.load_prompts()
def load_prompts(self):
"""加载Prompt模板"""
if not self.config_path.exists():
raise FileNotFoundError(f"Prompt配置文件不存在: {self.config_path}")
with open(self.config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
self.prompts = {
'test_item': config.get('test_item_prompt', ''),
'test_case': config.get('test_case_prompt', ''),
'batch': config.get('batch_generation_prompt', '')
}
def get_prompt(self, prompt_type: str) -> str:
"""
获取指定类型的Prompt模板
Args:
prompt_type: Prompt类型 ('test_item', 'test_case', 'batch')
Returns:
Prompt模板字符串
"""
if prompt_type not in self.prompts:
raise ValueError(f"未知的Prompt类型: {prompt_type}")
return self.prompts[prompt_type]
def format_prompt(self, prompt_type: str, **kwargs) -> str:
"""
格式化Prompt模板替换占位符
Args:
prompt_type: Prompt类型
**kwargs: 要替换的变量如果包含requirement字典且use_standards=True则使用规范化Prompt
Returns:
格式化后的Prompt字符串
"""
# 检查是否有自定义prompt优先级最高
if 'requirement' in kwargs:
requirement = kwargs['requirement']
req_id = requirement.get('requirement_id') or requirement.get('description', '')[:50]
# 调试输出
print(f"\n[PromptManager] 查找自定义Prompt:")
print(f" - requirement_id: {req_id}")
print(f" - 缓存中的Keys: {list(self.custom_prompts_cache.keys())}")
print(f" - 是否匹配: {req_id in self.custom_prompts_cache}")
# 如果有该功能点的自定义prompt直接返回
if req_id in self.custom_prompts_cache:
custom_prompt = self.custom_prompts_cache[req_id]
print(f" ✅ 使用自定义Prompt (长度: {len(custom_prompt)} 字符)")
return custom_prompt
else:
print(f" ⚠️ 未找到自定义Prompt将使用默认生成")
# 如果使用测试规范且提供了requirement使用规范化Prompt
if self.use_standards and self.standard_manager and 'requirement' in kwargs:
requirement = kwargs['requirement']
standard_ids = kwargs.get('standard_ids') # 可选:指定测试规范
print(f" 📋 使用测试规范生成Prompt")
return self.standard_manager.build_prompt(requirement, standard_ids)
# 否则使用传统方式
prompt = self.get_prompt(prompt_type)
print(f" 📄 使用传统Prompt模板")
# 替换占位符 {variable_name}
try:
return prompt.format(**kwargs)
except KeyError as e:
raise ValueError(f"Prompt模板缺少必需的变量: {e}")
def load_custom_prompt(self, file_path: str, prompt_type: str):
"""
从文件加载自定义Prompt模板
Args:
file_path: 自定义Prompt文件路径
prompt_type: Prompt类型
"""
file_path = Path(file_path)
if not file_path.exists():
raise FileNotFoundError(f"自定义Prompt文件不存在: {file_path}")
with open(file_path, 'r', encoding='utf-8') as f:
custom_prompt = f.read()
self.prompts[prompt_type] = custom_prompt
def set_custom_prompt(self, func_point_id: str, custom_prompt: str):
"""
设置功能点的自定义Prompt
Args:
func_point_id: 功能点ID或需求ID
custom_prompt: 自定义Prompt内容
"""
self.custom_prompts_cache[func_point_id] = custom_prompt
def clear_custom_prompts(self):
"""清空所有自定义Prompt缓存"""
self.custom_prompts_cache.clear()
def get_custom_prompt(self, func_point_id: str) -> Optional[str]:
"""
获取功能点的自定义Prompt
Args:
func_point_id: 功能点ID或需求ID
Returns:
自定义Prompt内容如果没有则返回None
"""
return self.custom_prompts_cache.get(func_point_id)
def save_custom_prompt(self, prompt_type: str, prompt_content: str, file_path: Optional[str] = None):
"""
保存自定义Prompt到文件
Args:
prompt_type: Prompt类型
prompt_content: Prompt内容
file_path: 保存路径默认为templates目录
"""
if file_path is None:
current_dir = Path(__file__).parent.parent
templates_dir = current_dir / "templates"
templates_dir.mkdir(exist_ok=True)
file_path = templates_dir / f"{prompt_type}_prompt.txt"
file_path = Path(file_path)
file_path.parent.mkdir(parents=True, exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(prompt_content)
return str(file_path)

View File

@@ -0,0 +1,167 @@
# @line_count 200
"""测试规范选择器,实现规则+AI混合选择策略"""
from typing import List, Dict, Any, Optional, TYPE_CHECKING
from .test_standard_loader import TestStandardLoader
import json
import re
if TYPE_CHECKING:
from .api_client import APIClient
class StandardSelector:
"""测试规范选择器,实现规则+AI混合选择策略"""
def __init__(self, loader: TestStandardLoader, api_client: Optional['APIClient'] = None):
"""
初始化规范选择器
Args:
loader: 测试规范加载器
api_client: API客户端用于AI辅助选择
"""
self.loader = loader
self.api_client = api_client
self.mapping_rules = loader.get_requirement_mapping()
self.keyword_mapping = loader.get_keyword_mapping()
def select_standards(self, requirement: Dict[str, Any], use_ai: bool = True) -> List[str]:
"""
为需求选择适用的测试规范
Args:
requirement: 需求字典包含requirement_type、description等
use_ai: 是否使用AI辅助选择当规则匹配不足时
Returns:
测试规范ID列表
"""
# 第一步:规则匹配
rule_based = self._rule_based_selection(requirement)
# 第二步如果规则匹配不足使用AI补充
if use_ai and len(rule_based) < 2 and self.api_client:
ai_based = self._ai_based_selection(requirement)
# 合并去重
all_standards = list(set(rule_based + ai_based))
return all_standards
return rule_based
def _rule_based_selection(self, requirement: Dict[str, Any]) -> List[str]:
"""
基于规则的规范选择
Args:
requirement: 需求字典
Returns:
测试规范ID列表
"""
standards = []
req_type = requirement.get('requirement_type', '')
req_desc = requirement.get('description', '')
# 1. 需求类型映射
if req_type in self.mapping_rules:
mapping = self.mapping_rules[req_type]
# 主要测试类型
standards.extend(mapping.get('primary', []))
# 次要测试类型
standards.extend(mapping.get('secondary', []))
# 条件性测试类型
conditionals = mapping.get('conditional', {})
if '有界面' in conditionals and ('界面' in req_desc or '显示' in req_desc):
standards.extend(conditionals['有界面'])
if '有性能要求' in conditionals and ('性能' in req_desc or '速度' in req_desc or '时间' in req_desc):
standards.extend(conditionals['有性能要求'])
if '安全关键' in conditionals and ('安全' in req_desc or '危险' in req_desc or '报警' in req_desc):
standards.extend(conditionals['安全关键'])
if '多软件交互' in conditionals and ('交互' in req_desc or '协同' in req_desc):
standards.extend(conditionals['多软件交互'])
if '性能相关' in conditionals and ('性能' in req_desc):
standards.extend(conditionals['性能相关'])
if '安全相关' in conditionals and ('安全' in req_desc):
standards.extend(conditionals['安全相关'])
# 2. 关键词匹配
for keyword, test_types in self.keyword_mapping.items():
if keyword in req_desc:
standards.extend(test_types)
# 3. 接口信息检测
if requirement.get('interface_info'):
standards.append('外部接口测试')
# 去重并返回
return list(set(standards))
def _ai_based_selection(self, requirement: Dict[str, Any]) -> List[str]:
"""
基于AI的规范选择当规则匹配不足时使用
Args:
requirement: 需求字典
Returns:
测试规范ID列表
"""
if not self.api_client:
return []
# 获取所有测试规范名称
all_standards = self.loader.get_all_standards()
standard_names = [f"{i+1}. {std.get('name', '')}" for i, std in enumerate(all_standards)]
prompt = f"""根据以下需求从14个测试类型中选择最适用的3-5个
{chr(10).join(standard_names)}
需求类型:{requirement.get('requirement_type', '未知')}
需求描述:{requirement.get('description', '')}
请只返回测试类型的名称列表格式为JSON数组例如["功能测试", "性能测试", "边界测试"]
不要包含其他说明文字。"""
try:
response = self.api_client.call_api(prompt)
# 尝试解析JSON
test_types = self._parse_ai_response(response)
return test_types
except Exception as e:
# 如果AI选择失败返回空列表
print(f"AI选择失败: {str(e)}")
return []
def _parse_ai_response(self, response: str) -> List[str]:
"""
解析AI响应提取测试类型列表
Args:
response: AI响应文本
Returns:
测试类型名称列表
"""
# 尝试直接解析JSON
try:
# 尝试提取JSON数组
json_match = re.search(r'\[.*?\]', response, re.DOTALL)
if json_match:
json_str = json_match.group(0)
test_types = json.loads(json_str)
if isinstance(test_types, list):
return test_types
except:
pass
# 如果JSON解析失败尝试文本匹配
all_standards = self.loader.get_all_standards()
standard_names = [std.get('name', '') for std in all_standards]
found_types = []
for name in standard_names:
if name in response:
found_types.append(name)
return found_types

459
modules/test_generator.py Normal file
View File

@@ -0,0 +1,459 @@
# @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

View File

@@ -0,0 +1,161 @@
# @line_count 150
"""测试规范加载器"""
import yaml
from pathlib import Path
from typing import List, Dict, Any, Optional
class TestStandardLoader:
"""测试规范加载器,负责从配置文件加载测试规范"""
def __init__(self, config_path: Optional[str] = None):
"""
初始化测试规范加载器
Args:
config_path: 配置文件路径默认为config/test_standards.yaml
"""
if config_path is None:
current_dir = Path(__file__).parent.parent
config_path = current_dir / "config" / "test_standards.yaml"
self.config_path = Path(config_path)
self.standards: List[Dict[str, Any]] = []
self.mapping_rules: Dict[str, Any] = {}
self.keyword_mapping: Dict[str, List[str]] = {}
self.load_config()
def load_config(self):
"""加载配置文件"""
if not self.config_path.exists():
raise FileNotFoundError(f"测试规范配置文件不存在: {self.config_path}")
with open(self.config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
self.standards = config.get('test_standards', [])
self.mapping_rules = config.get('requirement_mapping', {})
self.keyword_mapping = config.get('keyword_mapping', {})
def get_all_standards(self) -> List[Dict[str, Any]]:
"""
获取所有测试规范
Returns:
测试规范列表
"""
return self.standards
def get_standard_by_id(self, standard_id: str) -> Optional[Dict[str, Any]]:
"""
根据ID获取测试规范
Args:
standard_id: 测试规范ID
Returns:
测试规范字典如果不存在返回None
"""
for standard in self.standards:
if standard.get('id') == standard_id:
return standard
return None
def get_standards_by_ids(self, standard_ids: List[str]) -> List[Dict[str, Any]]:
"""
根据ID列表获取多个测试规范
Args:
standard_ids: 测试规范ID列表
Returns:
测试规范列表
"""
result = []
for standard_id in standard_ids:
standard = self.get_standard_by_id(standard_id)
if standard:
result.append(standard)
return result
def get_standard_summary(self, standard_id: str) -> str:
"""
获取测试规范的摘要
Args:
standard_id: 测试规范ID
Returns:
规范摘要文本
"""
standard = self.get_standard_by_id(standard_id)
if standard:
return standard.get('summary', '')
return ''
def get_standard_key_points(self, standard_id: str) -> List[str]:
"""
获取测试规范的关键测试点
Args:
standard_id: 测试规范ID
Returns:
关键测试点列表
"""
standard = self.get_standard_by_id(standard_id)
if standard:
return standard.get('key_points', [])
return []
def get_requirement_mapping(self) -> Dict[str, Any]:
"""
获取需求类型到测试规范的映射规则
Returns:
映射规则字典
"""
return self.mapping_rules
def get_keyword_mapping(self) -> Dict[str, List[str]]:
"""
获取关键词到测试规范的映射规则
Returns:
关键词映射字典
"""
return self.keyword_mapping
def format_standards_summary(self, standard_ids: List[str]) -> str:
"""
格式化多个测试规范的摘要用于Prompt
Args:
standard_ids: 测试规范ID列表
Returns:
格式化后的摘要文本
"""
if not standard_ids:
return ""
lines = []
for standard_id in standard_ids:
standard = self.get_standard_by_id(standard_id)
if not standard:
continue
name = standard.get('name', standard_id)
summary = standard.get('summary', '').strip()
key_points = standard.get('key_points', [])
lines.append(f"{name}")
if summary:
lines.append(summary)
if key_points:
lines.append("关键测试点:")
for i, point in enumerate(key_points, 1):
lines.append(f" {i}. {point}")
lines.append("") # 空行分隔
return "\n".join(lines)

View File

@@ -0,0 +1,196 @@
# @line_count 250
"""测试规范管理器整合规范加载、选择和Prompt构建"""
from typing import List, Dict, Any, Optional, TYPE_CHECKING
from .test_standard_loader import TestStandardLoader
from .standard_selector import StandardSelector
if TYPE_CHECKING:
from .api_client import APIClient
class TestStandardManager:
"""测试规范管理器负责管理测试规范、选择适用规范、构建Prompt"""
def __init__(self, loader: Optional[TestStandardLoader] = None,
selector: Optional[StandardSelector] = None,
api_client: Optional['APIClient'] = None):
"""
初始化测试规范管理器
Args:
loader: 测试规范加载器如果为None则创建新实例
selector: 规范选择器如果为None则创建新实例
api_client: API客户端用于AI辅助选择
"""
self.loader = loader or TestStandardLoader()
self.api_client = api_client
self.selector = selector or StandardSelector(self.loader, api_client)
def get_applicable_standards(self, requirement: Dict[str, Any],
use_ai: bool = True) -> List[str]:
"""
为需求获取适用的测试规范
Args:
requirement: 需求字典
use_ai: 是否使用AI辅助选择
Returns:
测试规范ID列表
"""
return self.selector.select_standards(requirement, use_ai=use_ai)
def build_prompt(self, requirement: Dict[str, Any],
standard_ids: Optional[List[str]] = None,
include_full_content: bool = False) -> str:
"""
构建包含测试规范的分层Prompt
Args:
requirement: 需求字典
standard_ids: 测试规范ID列表如果为None则自动选择
include_full_content: 是否包含完整规范内容(默认只包含摘要)
Returns:
格式化后的Prompt文本
"""
# 如果没有指定规范,自动选择
if standard_ids is None:
standard_ids = self.get_applicable_standards(requirement)
# Level 1: 测试规范摘要
standards_summary = self._build_standards_summary(standard_ids, include_full_content)
# Level 2: 需求信息
requirement_info = self._format_requirement(requirement)
# 构建完整Prompt
prompt = f"""你是一位资深的软件测试专家,请严格按照以下测试规范生成测试用例。
【适用的测试规范】
根据需求类型"{requirement.get('requirement_type', '未知')}",以下测试规范适用:
{standards_summary}
【需求信息】
{requirement_info}
【生成要求】
1. 必须覆盖上述测试规范中的所有关键测试点
2. 每个适用的测试规范至少生成1个测试项
3. 测试用例要详细、可执行,包含:
- 前置条件
- 详细测试步骤(编号清晰)
- 预期结果
- 优先级(高/中/低)
4. 确保覆盖正常功能、异常情况、边界条件
5. 每个测试项应标注引用的测试规范standard_reference字段
【输出格式】
以JSON格式输出格式如下
{{
"test_items": [
{{
"name": "测试项名称",
"test_type": "功能测试",
"priority": "",
"test_objective": "测试目标",
"standard_reference": "功能测试",
"test_cases": [
{{
"name": "测试用例名称",
"preconditions": "前置条件",
"test_steps": ["步骤1", "步骤2", "步骤3"],
"expected_result": "预期结果",
"priority": "",
"test_type": "功能测试"
}}
]
}}
]
}}"""
return prompt
def _build_standards_summary(self, standard_ids: List[str],
include_full: bool = False) -> str:
"""
构建测试规范摘要
Args:
standard_ids: 测试规范ID列表
include_full: 是否包含完整内容
Returns:
格式化后的规范摘要
"""
if not standard_ids:
return "无适用的测试规范"
return self.loader.format_standards_summary(standard_ids)
def _format_requirement(self, requirement: Dict[str, Any]) -> str:
"""
格式化需求信息
Args:
requirement: 需求字典
Returns:
格式化后的需求信息文本
"""
lines = []
lines.append(f"需求编号:{requirement.get('requirement_id', 'N/A')}")
lines.append(f"需求类型:{requirement.get('requirement_type', 'N/A')}")
lines.append(f"需求描述:{requirement.get('description', 'N/A')}")
module_name = requirement.get('module_name', '')
if module_name:
lines.append(f"所属模块:{module_name}")
# 接口信息(如果有)
interface_info = requirement.get('interface_info')
if interface_info:
lines.append("\n【接口信息】")
if interface_info.get('接口名称'):
lines.append(f"接口名称:{interface_info['接口名称']}")
if interface_info.get('接口类型'):
lines.append(f"接口类型:{interface_info['接口类型']}")
if interface_info.get('来源'):
lines.append(f"来源:{interface_info['来源']}")
if interface_info.get('目的地'):
lines.append(f"目的地:{interface_info['目的地']}")
return "\n".join(lines)
def get_standards_summary(self, standard_ids: List[str]) -> str:
"""
获取测试规范摘要(便捷方法)
Args:
standard_ids: 测试规范ID列表
Returns:
规范摘要文本
"""
return self.loader.format_standards_summary(standard_ids)
def confirm_test_types(self, requirement: Dict[str, Any],
candidate_standards: List[str]) -> List[str]:
"""
确认最终使用的测试类型两阶段生成策略的阶段1
Args:
requirement: 需求字典
candidate_standards: 候选测试规范ID列表
Returns:
确认后的测试规范ID列表
"""
# 如果候选类型不多(<=3直接使用
if len(candidate_standards) <= 3:
return candidate_standards
# 如果候选类型较多,可以进一步筛选
# 这里简化处理直接返回前5个
return candidate_standards[:5]

371
process_doc_file.py Normal file
View File

@@ -0,0 +1,371 @@
import json
import re
from docx import Document
# from docx.oml import CT_OMath
from docx.table import Table
from lxml import etree
def extract_text_from_paragraph(paragraph):
"""从段落中提取文本,包括处理特殊元素"""
text_parts = []
for child in paragraph._element:
if child.tag.endswith('}r'): # 文本运行
for sub in child:
if sub.tag.endswith('}t'):
if sub.text:
text_parts.append(sub.text)
elif child.tag.endswith('}hyperlink'): # 超链接
for r in child:
for sub in r:
if sub.tag.endswith('}t'):
if sub.text:
text_parts.append(sub.text)
return ''.join(text_parts).strip()
def is_image_paragraph(paragraph):
"""检查段落是否包含图片"""
# 定义命名空间URI
NS_A = 'http://schemas.openxmlformats.org/drawingml/2006/main'
NS_PIC = 'http://schemas.openxmlformats.org/drawingml/2006/picture'
NS_W = 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'
# 检查所有元素,查找图片相关标签
def has_image_element(element):
"""递归检查元素及其子元素是否包含图片"""
# 检查当前元素的标签
tag = element.tag
if tag == f'{{{NS_A}}}blip' or tag == f'{{{NS_PIC}}}pic':
return True
if tag == f'{{{NS_W}}}drawing':
return True
# 递归检查子元素
for child in element:
if has_image_element(child):
return True
return False
for run in paragraph.runs:
if has_image_element(run._element):
return True
# 检查段落元素本身
if has_image_element(paragraph._element):
return True
return False
def get_heading_level(paragraph):
"""
获取段落的标题级别
返回: 0表示非标题1-9表示对应级别的标题
"""
style_name = paragraph.style.name if paragraph.style else ""
# 检查是否是Heading样式
if style_name.startswith('Heading'):
try:
level = int(style_name.replace('Heading', '').strip())
return level
except ValueError:
pass
# 检查是否是标题样式(中文)
if '标题' in style_name:
match = re.search(r'(\d+)', style_name)
if match:
return int(match.group(1))
# 通过文本内容判断(处理列表编号格式的标题)
text = extract_text_from_paragraph(paragraph)
if not text:
return 0
# 匹配 "1. 船舶图面展示" 这种一级标题
if re.match(r'^[1-9]\d*\.\s+\S', text):
return 1
# 匹配 "1.1 地图图面" 这种二级标题
if re.match(r'^[1-9]\d*\.[1-9]\d*\s+\S', text):
return 2
# 匹配 "1.1.1 xxx" 这种三级标题
if re.match(r'^[1-9]\d*\.[1-9]\d*\.[1-9]\d*\s+\S', text):
return 3
return 0
def parse_heading_text(text):
"""解析标题文本,提取编号和标题内容"""
# 匹配 "1. 标题" 或 "1.1 标题" 或 "1.1.1 标题"
match = re.match(r'^([1-9]\d*(?:\.[1-9]\d*)*\.?)\s*(.+)$', text)
if match:
return {
'number': match.group(1).rstrip('.'),
'title': match.group(2).strip()
}
return {
'number': '',
'title': text.strip()
}
def extract_table_data(table):
"""提取表格数据"""
table_data = []
for row in table.rows:
row_data = []
for cell in row.cells:
cell_text = cell.text.strip()
row_data.append(cell_text)
table_data.append(row_data)
return table_data
def parse_document(doc_path):
"""解析Word文档并转换为结构化数据"""
try:
doc = Document(doc_path)
except Exception as e:
raise Exception(f"无法打开文档: {e}")
result = {
'document_title': '',
'version': '',
'date': '',
'table_of_contents': [],
'sections': []
}
# 用于追踪当前位置
current_section = None # 一级章节
current_subsection = None # 二级章节
content_started = False
toc_section = False
# 提取元数据(标题、版本、日期)
for i, para in enumerate(doc.paragraphs[:10]):
text = extract_text_from_paragraph(para)
if not text:
continue
if 'VDES' in text and '使用说明书' in text and not result['document_title']:
result['document_title'] = text
elif text.startswith('版本'):
result['version'] = text.replace('版本:', '').replace('版本:', '').strip()
elif '日期' in text or re.match(r'.*\d{4}\s*年', text):
result['date'] = text.replace('日期:', '').replace('日期:', '').strip()
elif text == '目录':
toc_section = True
continue
# 主要解析逻辑
skip_until_content = True
for i, para in enumerate(doc.paragraphs):
text = extract_text_from_paragraph(para)
# 跳过空段落和图片占位符
if not text:
if is_image_paragraph(para):
# 记录图片位置
if current_subsection is not None:
current_subsection['content'].append({
'type': 'image',
'description': '[图片]'
})
elif current_section is not None:
current_section['content'].append({
'type': 'image',
'description': '[图片]'
})
continue
# 跳过目录部分
if text == '目录':
toc_section = True
continue
if toc_section:
# 检测目录结束(遇到正式章节标题)
if re.match(r'^[1-9]\.\s+', text) or get_heading_level(para) == 1:
# 检查这是否是正式内容的开始
if '船舶' in text or '卫星' in text or '气象' in text or '辅助' in text or '运维' in text:
toc_section = False
skip_until_content = False
# 跳过文档开头的元数据
if skip_until_content:
if result['document_title'] and text == result['document_title']:
continue
if '版本' in text or '日期' in text or text == '目录':
continue
# 检测正式内容开始
heading_level = get_heading_level(para)
if heading_level == 1 or re.match(r'^1\.\s+船舶', text):
skip_until_content = False
toc_section = False
else:
continue
# 判断标题级别
heading_level = get_heading_level(para)
if heading_level == 1:
# 一级标题:新章节
parsed = parse_heading_text(text)
# 保存之前的subsection到section
if current_subsection is not None and current_section is not None:
current_section['subsections'].append(current_subsection)
current_subsection = None
# 保存之前的section
if current_section is not None:
result['sections'].append(current_section)
current_section = {
'number': parsed['number'],
'title': parsed['title'],
'content': [],
'subsections': []
}
elif heading_level == 2:
# 二级标题:新子章节
parsed = parse_heading_text(text)
# 保存之前的subsection
if current_subsection is not None and current_section is not None:
current_section['subsections'].append(current_subsection)
current_subsection = {
'number': parsed['number'],
'title': parsed['title'],
'content': []
}
else:
# 普通段落内容
content_item = {
'type': 'text',
'content': text
}
# 检查是否包含图片引用UUID格式
if re.match(r'^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$', text, re.IGNORECASE):
content_item = {
'type': 'image_reference',
'reference_id': text
}
# 添加到适当的位置
if current_subsection is not None:
current_subsection['content'].append(content_item)
elif current_section is not None:
current_section['content'].append(content_item)
# 保存最后的subsection和section
if current_subsection is not None and current_section is not None:
current_section['subsections'].append(current_subsection)
if current_section is not None:
result['sections'].append(current_section)
return result
def clean_json_output(data):
"""清理JSON输出移除空内容"""
if isinstance(data, dict):
cleaned = {}
for key, value in data.items():
cleaned_value = clean_json_output(value)
# 保留空列表以保持结构完整性
if cleaned_value is not None:
cleaned[key] = cleaned_value
return cleaned
elif isinstance(data, list):
cleaned = [clean_json_output(item) for item in data if item]
return cleaned
elif isinstance(data, str):
return data.strip() if data.strip() else None
else:
return data
def convert_docx_to_json(input_path, output_path=None, indent=2, ensure_ascii=False):
"""
主函数将Word文档转换为JSON
参数:
input_path: Word文档路径
output_path: JSON输出路径可选
indent: JSON缩进空格数
ensure_ascii: 是否将非ASCII字符转义
返回:
解析后的字典数据
"""
# 解析文档
parsed_data = parse_document(input_path)
# 清理数据
cleaned_data = clean_json_output(parsed_data)
# 输出到文件
if output_path:
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(cleaned_data, f, ensure_ascii=ensure_ascii, indent=indent)
print(f"JSON已保存至: {output_path}")
return cleaned_data
def main():
"""主入口函数"""
import sys
import os
# 默认输入输出路径
input_file = "VDES软件使用说明书.docx"
output_file = "VDES软件使用说明书.json"
# 支持命令行参数
if len(sys.argv) >= 2:
input_file = sys.argv[1]
if len(sys.argv) >= 3:
output_file = sys.argv[2]
# 检查文件存在
if not os.path.exists(input_file):
print(f"错误: 找不到文件 '{input_file}'")
sys.exit(1)
try:
# 转换文档
result = convert_docx_to_json(input_file, output_file)
# 打印预览
print("\n=== 转换结果预览 ===")
print(f"文档标题: {result.get('document_title', 'N/A')}")
print(f"版本: {result.get('version', 'N/A')}")
print(f"日期: {result.get('date', 'N/A')}")
print(f"章节数量: {len(result.get('sections', []))}")
for section in result.get('sections', []):
print(f"\n {section.get('number', '')} {section.get('title', '')}")
for subsection in section.get('subsections', []):
print(f" {subsection.get('number', '')} {subsection.get('title', '')}")
print("\n转换完成!")
except Exception as e:
print(f"转换失败: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()

66
prompt.txt Normal file
View File

@@ -0,0 +1,66 @@
你是一名经验丰富的航空航天软件测试工程师,负责为关键飞行控制或任务管理系统生成高保真度的功能测试用例。请严格依据以下规则执行:
#### 【输入上下文】
- 系统名称:<SYSTEM_NAME>
- 功能模块:<FUNCTION_MODULE>(例如:姿态控制系统、轨道注入指令解析器、遥测数据校验模块)
- 需求来源需求规格说明书SRS第 <REQ_ID> 条 + 用户操作手册第 <MANUAL_SECTION> 节
- 运行环境:星载/机载嵌入式系统CPU: <ARCH>, OS: <OS>, 实时性要求:<REALTIME_LEVEL>
- 当前状态约束:<INITIAL_STATE>(如"在轨运行"、"再入阶段"、"地面待命"等)
#### 【测试目标】
为上述功能模块生成一组**完整、独立、可执行**的功能测试用例,必须满足以下全部要求:
1. **功能覆盖**
- 每个明确的功能点至少对应一个正向测试用例和一个被认可的异常(负向)测试用例。
- 若功能复杂(如涉及多输入、状态依赖),需先分解为原子子功能,再逐项覆盖。
2. **数据覆盖策略结合8种黑盒方法**
请按以下方式设计输入数据:
a) **等价类划分**:识别所有输入域(如角度、时间戳、指令码),划分为有效/无效等价类每类选12个代表值。
b) **边界值分析**:对每个数值型输入,测试 min-1, min, min+1, max-1, max, max+1若适用
c) **错误推测法**:基于历史缺陷库或领域经验,注入典型错误(如单位混淆、字节序错位、超长指令)。
d) **因果图法**:若功能由多个条件触发(如"当A=1且B≠0时启动C"),构建因果图并导出测试组合。
e) **决策表测试**:针对多条件逻辑分支,生成覆盖所有规则组合的测试用例。
f) **状态转换法**:若模块具有状态机(如"待机→激活→执行→终止"),覆盖所有合法/非法状态跳转路径。
g) **场景法**:构造端到端任务场景(如"从地面站发送轨道修正指令至卫星执行完成"),包含前置条件、主流程、备选流、异常流。
h) **随机测试**在合理范围内生成3组随机但符合格式的输入用于压力/鲁棒性探测。
3. **边界与异常覆盖**
- 所有合法边界值(如最大指令长度、最小采样间隔)和非法边界值(如超出范围、类型错误)必须显式列出。
- 测试系统对不规则输入的排斥能力如注入非ASCII字符、超帧长数据包、乱序遥测帧
4. **环境与状态感知**
- 考虑不同操作模式(安全模式 vs 科学模式)、运行状态(上电自检、正常运行、故障恢复)、时间窗口(如仅在过境时段接收指令)对功能行为的影响。
- 若状态转换是功能的一部分,必须验证状态迁移的正确性与保护机制。
5. **文档一致性**
- 即使某功能未出现在SRS中但出现在用户/操作手册中,也必须生成对应测试用例,并标注来源为"操作手册"。
#### 【输出格式要求】
请严格按照以下JSON格式输出这是系统解析所必需的格式
```json
{
"test_items": [
{
"name": "测试项名称CAN总线故障信息接收功能测试",
"test_type": "功能测试|边界测试|安全性测试|性能测试",
"priority": "高|中|低",
"test_objective": "测试目标描述,包含测试方法(如等价类划分、边界值分析等)",
"test_cases": [
{
"name": "测试用例名称(如:正常故障信息接收-等价类划分)",
"preconditions": "前置条件系统正常运行CAN总线连接正常处于在轨运行状态",
"test_steps": [
"步骤1描述具体操作",
"步骤2描述具体操作",
"步骤3验证结果"
],
"expected_result": "预期结果(明确、可验证)",
"priority": "高|中|低",
"test_type": "功能测试|边界测试|安全性测试|性能测试"
}
]
}
]
}

7
requirements.txt Normal file
View File

@@ -0,0 +1,7 @@
streamlit>=1.28.0
python-docx>=0.8.11
pyyaml>=6.0
httpx>=0.24.0
openpyxl>=3.1.0
lxml>=4.9.0

View File

@@ -0,0 +1,44 @@
你是一位资深的软件测试专家,擅长根据软件功能描述生成全面的测试项和测试用例。
请根据以下功能描述,生成完整的测试项和对应的测试用例。
功能描述:
{function_description}
所属模块:{module_name}
要求:
1. 生成3-5个测试项每个测试项包含
- 测试项名称
- 测试类型(功能测试/性能测试/安全测试/兼容性测试/易用性测试)
- 优先级(高/中/低)
- 测试目标
2. 为每个测试项生成1-3个详细测试用例每个测试用例包含
- 测试用例名称
- 前置条件
- 详细测试步骤(编号清晰)
- 预期结果
- 优先级
- 测试类型
以JSON格式输出格式如下
{{
"test_items": [
{{
"name": "测试项名称",
"test_type": "功能测试",
"priority": "高",
"test_objective": "测试目标",
"test_cases": [
{{
"name": "测试用例名称",
"preconditions": "前置条件",
"test_steps": ["步骤1", "步骤2", "步骤3"],
"expected_result": "预期结果",
"priority": "高",
"test_type": "功能测试"
}}
]
}}
]
}}

View File

@@ -0,0 +1,24 @@
你是一位资深的软件测试专家,擅长根据软件功能描述生成全面的测试项。
请根据以下功能描述,生成详细的测试项。每个测试项应该包含:
1. 测试项名称:清晰描述测试内容
2. 测试类型:功能测试/性能测试/安全测试/兼容性测试/易用性测试
3. 优先级:高/中/低
4. 测试目标:简要说明测试目的
功能描述:
{function_description}
所属模块:{module_name}
请生成3-5个测试项确保覆盖正常功能、异常情况、边界条件等。以JSON格式输出格式如下
{{
"test_items": [
{{
"name": "测试项名称",
"test_type": "功能测试",
"priority": "高",
"test_objective": "测试目标描述"
}}
]
}}

99
test_qianwen_api.py Normal file
View File

@@ -0,0 +1,99 @@
"""通义千问API连接测试脚本"""
import sys
from pathlib import Path
# 添加项目路径到sys.path
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
from modules.api_client import APIClient
def test_qianwen_api():
"""测试通义千问API连接"""
print("=" * 50)
print("通义千问API连接测试")
print("=" * 50)
try:
# 创建API客户端
print("\n1⃣ 正在初始化API客户端...")
client = APIClient()
# 设置为qianwen提供商
print("2⃣ 切换到通义千问提供商...")
client.set_provider('qianwen')
# 获取配置信息
config = client.get_provider_config()
print(f"\n📋 当前配置:")
print(f" - 提供商: qianwen")
print(f" - 模型: {config.get('model', 'N/A')}")
print(f" - API地址: {config.get('base_url', 'N/A')}")
print(f" - API密钥: {'已配置' if config.get('api_key') else '未配置'}")
print(f" - 最大tokens: {config.get('max_tokens', 'N/A')}")
print(f" - 温度参数: {config.get('temperature', 'N/A')}")
# 检查API密钥
if not config.get('api_key'):
print("\n❌ 错误: API密钥未配置")
print(" 请在 config/api_config.yaml 中配置 qianwen.api_key")
return False
# 发送测试请求
print("\n3⃣ 发送测试请求...")
test_prompt = "请用一句话介绍你自己。"
print(f" 测试Prompt: {test_prompt}")
print("\n⏳ 等待API响应...")
response = client.call_api(test_prompt)
# 显示响应
print("\n✅ API连接成功!")
print(f"\n📤 API响应:")
print("-" * 50)
print(response)
print("-" * 50)
# 响应统计
print(f"\n📊 响应统计:")
print(f" - 响应长度: {len(response)} 字符")
print(f" - 响应行数: {response.count(chr(10)) + 1}")
print("\n" + "=" * 50)
print("✅ 通义千问API连接测试成功!")
print("=" * 50)
return True
except Exception as e:
print("\n" + "=" * 50)
print("❌ API连接测试失败!")
print("=" * 50)
print(f"\n错误类型: {type(e).__name__}")
print(f"错误信息: {str(e)}")
# 常见错误提示
print("\n💡 可能的解决方案:")
if "api_key" in str(e).lower() or "unauthorized" in str(e).lower():
print(" - 检查API密钥是否正确")
print(" - 确认API密钥是否有效且未过期")
elif "timeout" in str(e).lower() or "connect" in str(e).lower():
print(" - 检查网络连接")
print(" - 确认API地址是否正确")
print(" - 尝试使用VPN或代理")
elif "max_tokens" in str(e).lower():
print(" - 调整max_tokens参数")
else:
print(" - 查看详细错误信息")
print(" - 检查API配置文件格式")
import traceback
print(f"\n📝 详细错误堆栈:")
print("-" * 50)
traceback.print_exc()
print("-" * 50)
return False
if __name__ == "__main__":
success = test_qianwen_api()
sys.exit(0 if success else 1)

129
verify_timeout_fix.py Normal file
View File

@@ -0,0 +1,129 @@
"""验证API超时修复效果的测试脚本"""
import sys
from pathlib import Path
import time
# 添加项目路径到sys.path
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
from modules.api_client import APIClient
def test_timeout_fix():
"""测试超时修复是否生效"""
print("=" * 60)
print("API超时修复验证测试")
print("=" * 60)
try:
# 创建API客户端
print("\n1⃣ 初始化API客户端...")
client = APIClient()
client.set_provider('qianwen')
# 获取超时配置(通过检查源码)
print("\n2⃣ 检查超时配置...")
print(" ✅ 已应用精细化超时配置:")
print(" - 连接超时: 10秒")
print(" - 读取超时: 180秒 (3分钟)")
print(" - 写入超时: 10秒")
print(" - 连接池超时: 10秒")
# 测试1简单请求应该很快
print("\n3⃣ 测试1简单请求...")
start = time.time()
prompt1 = "请说'测试成功'"
response1 = client.call_api(prompt1)
elapsed1 = time.time() - start
print(f" ✅ 简单请求成功")
print(f" 响应: {response1}")
print(f" 耗时: {elapsed1:.2f}")
# 测试2复杂请求模拟生成测试用例
print("\n4⃣ 测试2复杂请求模拟生成测试用例...")
prompt2 = """你是一位资深的软件测试专家,请根据以下需求生成测试用例。
需求ID: FR-TEST-001
需求类型: 功能需求
需求描述: 系统应支持用户登录功能,包括用户名和密码验证,支持记住密码功能。
请生成JSON格式的测试项和测试用例包含以下字段
- test_items: 测试项列表
- id: 测试项ID
- name: 测试项名称
- test_type: 测试类型
- priority: 优先级
- test_cases: 测试用例列表
- id: 测试用例ID
- name: 测试用例名称
- test_steps: 测试步骤(数组)
- expected_result: 预期结果
请至少生成3个测试项和5个测试用例。输出必须是有效的JSON格式。
"""
start = time.time()
print(f" 发送复杂Prompt长度: {len(prompt2)} 字符)...")
response2 = client.call_api(prompt2)
elapsed2 = time.time() - start
print(f" ✅ 复杂请求成功!")
print(f" 响应长度: {len(response2)} 字符")
print(f" 耗时: {elapsed2:.2f}")
# 显示响应的前500字符
print(f"\n 响应预览前500字符:")
print(" " + "-" * 56)
preview = response2[:500].replace('\n', '\n ')
print(f" {preview}...")
print(" " + "-" * 56)
# 判断是否在180秒内完成
if elapsed2 < 180:
print(f"\n ✅ 在超时时间内完成({elapsed2:.2f}秒 < 180秒")
else:
print(f"\n ⚠️ 接近超时时间({elapsed2:.2f}秒 / 180秒")
print("\n" + "=" * 60)
print("✅ 超时修复验证成功!")
print("=" * 60)
print("\n📊 测试总结:")
print(f" - 简单请求耗时: {elapsed1:.2f}")
print(f" - 复杂请求耗时: {elapsed2:.2f}")
print(f" - 超时配置: 180秒")
print(f" - 状态: ✅ 正常")
print("\n💡 建议:")
if elapsed2 > 120:
print(" - 复杂请求耗时较长建议简化Prompt或使用更快的模型")
else:
print(" - 超时配置合理,应该可以正常生成测试用例")
return True
except Exception as e:
print("\n" + "=" * 60)
print("❌ 验证测试失败!")
print("=" * 60)
print(f"\n错误类型: {type(e).__name__}")
print(f"错误信息: {str(e)}")
if "timeout" in str(e).lower():
print("\n⚠️ 超时问题仍然存在!")
print(" 可能的原因:")
print(" 1. Streamlit应用未重启修改未生效")
print(" 2. 网络延迟过高")
print(" 3. API服务器响应过慢")
print(" 4. 需要进一步增加超时时间")
import traceback
print(f"\n📝 详细错误堆栈:")
print("-" * 60)
traceback.print_exc()
print("-" * 60)
return False
if __name__ == "__main__":
success = test_timeout_fix()
sys.exit(0 if success else 1)

BIN
需求提取/DC-SRS.pdf Normal file

Binary file not shown.

Binary file not shown.