8a538bb950
- Renamed `check_environment` to `check_api_key_configured` for clarity, simplifying the API key validation logic. - Removed the blocking behavior of the API key check during application startup, allowing the app to run while providing a prompt for configuration. - Updated `LocalAgentApp` to accept an `api_configured` parameter, enabling conditional messaging for API key setup. - Enhanced the `SandboxRunner` to support backup management and improved execution result handling with detailed metrics. - Integrated data governance strategies into the `HistoryManager`, ensuring compliance and improved data management. - Added privacy settings and metrics tracking across various components to enhance user experience and application safety.
337 lines
10 KiB
Python
337 lines
10 KiB
Python
"""
|
|
测试运行器
|
|
提供统一的测试执行和报告生成
|
|
"""
|
|
|
|
import unittest
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from typing import Dict, List, Any
|
|
|
|
# 添加项目根目录到路径
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
|
|
class TestMetricsCollector(unittest.TestResult):
|
|
"""
|
|
测试指标收集器
|
|
收集测试执行的详细指标
|
|
"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.test_metrics = []
|
|
self.start_time = None
|
|
self.current_test_start = None
|
|
|
|
def startTest(self, test):
|
|
super().startTest(test)
|
|
self.current_test_start = datetime.now()
|
|
|
|
def stopTest(self, test):
|
|
super().stopTest(test)
|
|
duration = (datetime.now() - self.current_test_start).total_seconds()
|
|
|
|
# 确定测试状态
|
|
status = 'passed'
|
|
error_msg = None
|
|
|
|
if test in [t[0] for t in self.failures]:
|
|
status = 'failed'
|
|
error_msg = [e[1] for e in self.failures if e[0] == test][0]
|
|
elif test in [t[0] for t in self.errors]:
|
|
status = 'error'
|
|
error_msg = [e[1] for e in self.errors if e[0] == test][0]
|
|
elif test in self.skipped:
|
|
status = 'skipped'
|
|
|
|
# 记录指标
|
|
self.test_metrics.append({
|
|
'test_name': str(test),
|
|
'test_class': test.__class__.__name__,
|
|
'test_method': test._testMethodName,
|
|
'status': status,
|
|
'duration_seconds': duration,
|
|
'error_message': error_msg
|
|
})
|
|
|
|
def get_summary(self) -> Dict[str, Any]:
|
|
"""获取测试摘要"""
|
|
total = self.testsRun
|
|
passed = len([m for m in self.test_metrics if m['status'] == 'passed'])
|
|
failed = len(self.failures)
|
|
errors = len(self.errors)
|
|
skipped = len(self.skipped)
|
|
|
|
total_duration = sum(m['duration_seconds'] for m in self.test_metrics)
|
|
|
|
return {
|
|
'total_tests': total,
|
|
'passed': passed,
|
|
'failed': failed,
|
|
'errors': errors,
|
|
'skipped': skipped,
|
|
'success_rate': passed / total if total > 0 else 0,
|
|
'total_duration_seconds': total_duration,
|
|
'timestamp': datetime.now().isoformat()
|
|
}
|
|
|
|
|
|
def run_test_suite(test_modules: List[str], output_dir: Path = None) -> Dict[str, Any]:
|
|
"""
|
|
运行测试套件并生成报告
|
|
|
|
Args:
|
|
test_modules: 测试模块名称列表
|
|
output_dir: 报告输出目录
|
|
|
|
Returns:
|
|
测试结果摘要
|
|
"""
|
|
# 创建测试套件
|
|
loader = unittest.TestLoader()
|
|
suite = unittest.TestSuite()
|
|
|
|
for module_name in test_modules:
|
|
try:
|
|
module = __import__(module_name, fromlist=[''])
|
|
suite.addTests(loader.loadTestsFromModule(module))
|
|
except ImportError as e:
|
|
print(f"警告: 无法加载测试模块 {module_name}: {e}")
|
|
|
|
# 运行测试
|
|
print(f"\n{'='*70}")
|
|
print(f"开始运行测试套件 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print(f"{'='*70}\n")
|
|
|
|
result = TestMetricsCollector()
|
|
suite.run(result)
|
|
|
|
# 生成摘要
|
|
summary = result.get_summary()
|
|
|
|
# 打印结果
|
|
print(f"\n{'='*70}")
|
|
print("测试执行摘要")
|
|
print(f"{'='*70}")
|
|
print(f"总测试数: {summary['total_tests']}")
|
|
print(f"通过: {summary['passed']} ✅")
|
|
print(f"失败: {summary['failed']} ❌")
|
|
print(f"错误: {summary['errors']} ⚠️")
|
|
print(f"跳过: {summary['skipped']} ⏭️")
|
|
print(f"成功率: {summary['success_rate']:.1%}")
|
|
print(f"总耗时: {summary['total_duration_seconds']:.2f}秒")
|
|
print(f"{'='*70}\n")
|
|
|
|
# 显示失败的测试
|
|
if result.failures:
|
|
print("失败的测试:")
|
|
for test, traceback in result.failures:
|
|
print(f" ❌ {test}")
|
|
print(f" {traceback.split(chr(10))[0]}")
|
|
|
|
# 显示错误的测试
|
|
if result.errors:
|
|
print("\n错误的测试:")
|
|
for test, traceback in result.errors:
|
|
print(f" ⚠️ {test}")
|
|
print(f" {traceback.split(chr(10))[0]}")
|
|
|
|
# 保存详细报告
|
|
if output_dir:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# JSON报告
|
|
report_data = {
|
|
'summary': summary,
|
|
'test_details': result.test_metrics,
|
|
'failures': [
|
|
{
|
|
'test': str(test),
|
|
'traceback': traceback
|
|
}
|
|
for test, traceback in result.failures
|
|
],
|
|
'errors': [
|
|
{
|
|
'test': str(test),
|
|
'traceback': traceback
|
|
}
|
|
for test, traceback in result.errors
|
|
]
|
|
}
|
|
|
|
report_file = output_dir / f"test_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
with open(report_file, 'w', encoding='utf-8') as f:
|
|
json.dump(report_data, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"\n详细报告已保存到: {report_file}")
|
|
|
|
# Markdown报告
|
|
md_report = generate_markdown_report(summary, result)
|
|
md_file = output_dir / f"test_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
|
|
with open(md_file, 'w', encoding='utf-8') as f:
|
|
f.write(md_report)
|
|
|
|
print(f"Markdown报告已保存到: {md_file}")
|
|
|
|
return summary
|
|
|
|
|
|
def generate_markdown_report(summary: Dict[str, Any], result: TestMetricsCollector) -> str:
|
|
"""生成Markdown格式的测试报告"""
|
|
md = f"""# 测试执行报告
|
|
|
|
**生成时间**: {summary['timestamp']}
|
|
|
|
## 执行摘要
|
|
|
|
| 指标 | 数值 |
|
|
|------|------|
|
|
| 总测试数 | {summary['total_tests']} |
|
|
| 通过 | {summary['passed']} ✅ |
|
|
| 失败 | {summary['failed']} ❌ |
|
|
| 错误 | {summary['errors']} ⚠️ |
|
|
| 跳过 | {summary['skipped']} ⏭️ |
|
|
| 成功率 | {summary['success_rate']:.1%} |
|
|
| 总耗时 | {summary['total_duration_seconds']:.2f}秒 |
|
|
|
|
## 测试覆盖矩阵
|
|
|
|
### 关键路径覆盖
|
|
|
|
"""
|
|
|
|
# 按测试类分组
|
|
test_by_class = {}
|
|
for metric in result.test_metrics:
|
|
class_name = metric['test_class']
|
|
if class_name not in test_by_class:
|
|
test_by_class[class_name] = []
|
|
test_by_class[class_name].append(metric)
|
|
|
|
for class_name, tests in test_by_class.items():
|
|
passed = len([t for t in tests if t['status'] == 'passed'])
|
|
total = len(tests)
|
|
md += f"\n#### {class_name}\n\n"
|
|
md += f"- 覆盖率: {passed}/{total} ({passed/total:.1%})\n"
|
|
md += f"- 测试用例:\n"
|
|
|
|
for test in tests:
|
|
status_icon = {
|
|
'passed': '✅',
|
|
'failed': '❌',
|
|
'error': '⚠️',
|
|
'skipped': '⏭️'
|
|
}.get(test['status'], '❓')
|
|
|
|
md += f" - {status_icon} `{test['test_method']}` ({test['duration_seconds']:.3f}s)\n"
|
|
|
|
# 失败详情
|
|
if result.failures or result.errors:
|
|
md += "\n## 失败详情\n\n"
|
|
|
|
if result.failures:
|
|
md += "### 失败的测试\n\n"
|
|
for test, traceback in result.failures:
|
|
md += f"#### {test}\n\n"
|
|
md += "```\n"
|
|
md += traceback
|
|
md += "\n```\n\n"
|
|
|
|
if result.errors:
|
|
md += "### 错误的测试\n\n"
|
|
for test, traceback in result.errors:
|
|
md += f"#### {test}\n\n"
|
|
md += "```\n"
|
|
md += traceback
|
|
md += "\n```\n\n"
|
|
|
|
# 建议
|
|
md += "\n## 改进建议\n\n"
|
|
|
|
if summary['success_rate'] < 1.0:
|
|
md += "- ⚠️ 存在失败的测试，需要修复\n"
|
|
|
|
if summary['success_rate'] >= 0.95:
|
|
md += "- ✅ 测试覆盖率良好\n"
|
|
elif summary['success_rate'] >= 0.8:
|
|
md += "- ⚠️ 建议提高测试覆盖率\n"
|
|
else:
|
|
md += "- ❌ 测试覆盖率较低，需要补充测试用例\n"
|
|
|
|
return md
|
|
|
|
|
|
def run_critical_path_tests():
|
|
"""运行关键路径测试"""
|
|
test_modules = [
|
|
'test_e2e_integration',
|
|
'test_security_regression',
|
|
]
|
|
|
|
workspace_path = Path(__file__).parent.parent / "workspace"
|
|
output_dir = workspace_path / "test_reports"
|
|
|
|
summary = run_test_suite(test_modules, output_dir)
|
|
|
|
# 返回退出码
|
|
return 0 if summary['failed'] == 0 and summary['errors'] == 0 else 1
|
|
|
|
|
|
def run_all_tests():
|
|
"""运行所有测试"""
|
|
test_modules = [
|
|
'test_intent_classifier',
|
|
'test_rule_checker',
|
|
'test_history_manager',
|
|
'test_task_features',
|
|
'test_data_governance',
|
|
'test_config_refresh',
|
|
'test_retry_fix',
|
|
'test_e2e_integration',
|
|
'test_security_regression',
|
|
]
|
|
|
|
workspace_path = Path(__file__).parent.parent / "workspace"
|
|
output_dir = workspace_path / "test_reports"
|
|
|
|
summary = run_test_suite(test_modules, output_dir)
|
|
|
|
# 返回退出码
|
|
return 0 if summary['failed'] == 0 and summary['errors'] == 0 else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description='LocalAgent 测试运行器')
|
|
parser.add_argument(
|
|
'--mode',
|
|
choices=['all', 'critical', 'unit'],
|
|
default='critical',
|
|
help='测试模式: all(全部), critical(关键路径), unit(单元测试)'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.mode == 'all':
|
|
exit_code = run_all_tests()
|
|
elif args.mode == 'critical':
|
|
exit_code = run_critical_path_tests()
|
|
else: # unit
|
|
test_modules = [
|
|
'test_intent_classifier',
|
|
'test_rule_checker',
|
|
'test_history_manager',
|
|
]
|
|
workspace_path = Path(__file__).parent.parent / "workspace"
|
|
output_dir = workspace_path / "test_reports"
|
|
summary = run_test_suite(test_modules, output_dir)
|
|
exit_code = 0 if summary['failed'] == 0 and summary['errors'] == 0 else 1
|
|
|
|
sys.exit(exit_code)
|
|
|