""" 测试工具和辅助函数 """ import asyncio import time import json from datetime import datetime from typing import Dict, List, Any from pathlib import Path class TestReporter: """测试报告生成器""" def __init__(self, output_dir: str = "test_reports"): self.output_dir = Path(output_dir) self.output_dir.mkdir(exist_ok=True) self.start_time = datetime.now() def generate_report(self, results: Dict[str, Any], report_name: str = None): """生成测试报告""" if not report_name: report_name = f"test_report_{self.start_time.strftime('%Y%m%d_%H%M%S')}" # 生成 Markdown 报告 md_content = self._generate_markdown_report(results) md_file = self.output_dir / f"{report_name}.md" with open(md_file, 'w', encoding='utf-8') as f: f.write(md_content) # 生成 JSON 报告 json_content = self._generate_json_report(results) json_file = self.output_dir / f"{report_name}.json" with open(json_file, 'w', encoding='utf-8') as f: json.dump(json_content, f, indent=2, ensure_ascii=False) return { "markdown": str(md_file), "json": str(json_file) } def _generate_markdown_report(self, results: Dict[str, Any]) -> str: """生成 Markdown 格式报告""" report = f"""# RAG 系统测试报告 ## 测试概览 - **测试时间**: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')} - **报告生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ## 测试结果汇总 """ # 添加各项测试结果 for test_type, test_results in results.items(): if isinstance(test_results, list): successful = sum(1 for r in test_results if isinstance(r, dict) and r.get('status_code') == 200) total = len(test_results) success_rate = (successful / total * 100) if total > 0 else 0 report += f"### {test_type.replace('_', ' ').title()}\n" report += f"- 总请求数: {total}\n" report += f"- 成功数: {successful}\n" report += f"- 成功率: {success_rate:.1f}%\n" if test_results: avg_time = sum(r.get('response_time', 0) for r in test_results if isinstance(r, dict)) / len(test_results) report += f"- 平均响应时间: {avg_time:.3f}秒\n" report += "\n" return report def _generate_json_report(self, results: Dict[str, Any]) -> Dict[str, Any]: """生成 JSON 格式报告""" return { "test_info": { "start_time": self.start_time.isoformat(), "end_time": datetime.now().isoformat(), "duration": (datetime.now() - self.start_time).total_seconds() }, "results": results, "summary": self._calculate_summary(results) } def _calculate_summary(self, results: Dict[str, Any]) -> Dict[str, Any]: """计算测试摘要""" summary = { "total_requests": 0, "total_successful": 0, "overall_success_rate": 0, "test_types": len(results) } for test_results in results.values(): if isinstance(test_results, list): summary["total_requests"] += len(test_results) summary["total_successful"] += sum( 1 for r in test_results if isinstance(r, dict) and r.get('status_code') == 200 ) if summary["total_requests"] > 0: summary["overall_success_rate"] = ( summary["total_successful"] / summary["total_requests"] * 100 ) return summary class TestDataGenerator: """测试数据生成器""" @staticmethod def generate_test_documents(count: int, base_content: str = None) -> List[Dict[str, str]]: """生成测试文档""" if not base_content: base_content = "这是一个测试文档,包含关于人工智能和机器学习的内容。" documents = [] for i in range(count): content = f"{base_content} 文档编号: {i+1}。" + f"额外内容: {'AI技术' if i % 2 == 0 else 'ML算法'}。" * 10 documents.append({ "content": content, "filename": f"test_doc_{i+1:03d}.txt" }) return documents @staticmethod def generate_test_questions(count: int) -> List[str]: """生成测试问题""" base_questions = [ "什么是人工智能?", "机器学习的应用有哪些?", "深度学习和传统机器学习的区别?", "自然语言处理的主要挑战?", "计算机视觉技术的发展趋势?", ] questions = [] for i in range(count): base_q = base_questions[i % len(base_questions)] questions.append(f"{base_q} (查询 {i+1})") return questions class PerformanceAnalyzer: """性能分析器""" @staticmethod def analyze_response_times(results: List[Dict[str, Any]]) -> Dict[str, float]: """分析响应时间""" times = [r.get('response_time', 0) for r in results if isinstance(r, dict)] if not times: return {} times.sort() n = len(times) return { "min": min(times), "max": max(times), "avg": sum(times) / n, "median": times[n // 2], "p95": times[int(n * 0.95)] if n > 1 else times[0], "p99": times[int(n * 0.99)] if n > 1 else times[0] } @staticmethod def analyze_success_rates(results: List[Dict[str, Any]]) -> Dict[str, Any]: """分析成功率""" total = len(results) successful = sum(1 for r in results if isinstance(r, dict) and r.get('status_code') == 200) return { "total": total, "successful": successful, "failed": total - successful, "success_rate": (successful / total * 100) if total > 0 else 0, "failure_rate": ((total - successful) / total * 100) if total > 0 else 0 } def format_duration(seconds: float) -> str: """格式化持续时间""" if seconds < 1: return f"{seconds * 1000:.1f}ms" elif seconds < 60: return f"{seconds:.2f}s" else: minutes = int(seconds // 60) seconds = seconds % 60 return f"{minutes}m {seconds:.1f}s" def print_test_summary(test_name: str, results: List[Dict[str, Any]]): """打印测试摘要""" if not results: print(f"❌ {test_name}: 没有结果") return analyzer = PerformanceAnalyzer() success_info = analyzer.analyze_success_rates(results) time_info = analyzer.analyze_response_times(results) print(f"✅ {test_name}:") print(f" - 成功率: {success_info['success_rate']:.1f}% ({success_info['successful']}/{success_info['total']})") if time_info: print(f" - 响应时间: 平均 {format_duration(time_info['avg'])}, " f"最大 {format_duration(time_info['max'])}, " f"P95 {format_duration(time_info['p95'])}") async def wait_for_server(base_url: str, timeout: int = 30) -> bool: """等待服务器启动""" import aiohttp print(f"🔍 等待服务器启动 ({base_url})...") async with aiohttp.ClientSession() as session: for i in range(timeout): try: async with session.get(f"{base_url}/health", timeout=1) as response: if response.status == 200: print(f"✅ 服务器已启动 (耗时: {i+1}秒)") return True except: pass await asyncio.sleep(1) if i % 5 == 4: # 每5秒显示一次等待状态 print(f"⏳ 仍在等待服务器启动... ({i+1}/{timeout})") print(f"❌ 服务器启动超时 ({timeout}秒)") return False