#!/usr/bin/env python3 """ 前后端API交互自动化测试工具（重构版） 模块化架构，支持配置文件化、数据生成、多种验证规则和多种报告格式 """ import sys import time from pathlib import Path from typing import List, Dict, Any # 添加项目根目录到Python路径 project_root = Path(__file__).parent sys.path.insert(0, str(project_root)) from config.settings import config from core.api_tester import APITester from utils.logger import TestLogger, LoggerFactory from utils.reporter import ReportGenerator, TestResult as ReportTestResult, TestSummary from test_cases.api_tests import APITestCases class TestRunner: """测试运行器""" def __init__(self): """初始化测试运行器""" self.logger = LoggerFactory.get_logger( "test_runner", config.logging_file, config.logging_level, config.logging_console ) self.tester = APITester(self.logger) self.test_cases = APITestCases(self.tester) self.report_generator = ReportGenerator(config.report_output_dir) self.test_results: List[ReportTestResult] = [] self.start_time: float = 0 def run_test(self, test_name: str, test_func) -> bool: """ 运行单个测试 Args: test_name: 测试名称 test_func: 测试函数 Returns: 是否通过 """ self.logger.log_test_start(test_name) try: start_time = time.time() passed = test_func() duration = (time.time() - start_time) * 1000 self.logger.log_test_end(test_name, passed, duration) # 记录测试结果 test_result = ReportTestResult( test_name=test_name, test_type="API", url=config.api_base_url, method="N/A", status_code=200 if passed else 0, response_time=duration, success=passed, error_message="" if passed else f"测试失败", request_data={}, response_data={}, timestamp=time.strftime("%Y-%m-%d %H:%M:%S") ) self.test_results.append(test_result) return passed except Exception as e: duration = (time.time() - start_time) * 1000 self.logger.log_error(e) test_result = ReportTestResult( test_name=test_name, test_type="API", url=config.api_base_url, method="N/A", status_code=0, response_time=duration, success=False, error_message=str(e), request_data={}, response_data={}, timestamp=time.strftime("%Y-%m-%d %H:%M:%S") ) self.test_results.append(test_result) return False def run_all_tests(self) -> Dict[str, Any]: """ 运行所有测试 Returns: 测试结果摘要 """ self.start_time = time.time() self.logger.info("="*60) self.logger.info("开始运行所有测试用例") self.logger.info("="*60) # 1. 健康检查 self.run_test("健康检查", self.test_cases.test_health_check) # 2. 用户登录 self.run_test("用户登录", self.test_cases.test_login) # 3. 获取列表 self.run_test("获取用户列表", self.test_cases.test_get_user_list) self.run_test("获取角色列表", self.test_cases.test_get_role_list) self.run_test("获取菜单列表", self.test_cases.test_get_menu_list) # 4. 创建数据 self.run_test("创建用户", self.test_cases.test_create_user) self.run_test("创建角色", self.test_cases.test_create_role) # 5. 搜索数据 self.run_test("搜索用户", self.test_cases.test_search_users) # 计算测试摘要 total_time = (time.time() - self.start_time) * 1000 total = len(self.test_results) passed = sum(1 for r in self.test_results if r.success) failed = total - passed skipped = 0 pass_rate = (passed / total * 100) if total > 0 else 0 summary = TestSummary( total=total, passed=passed, failed=failed, skipped=skipped, pass_rate=pass_rate, total_time=total_time, start_time=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.start_time)), end_time=time.strftime("%Y-%m-%d %H:%M:%S") ) # 记录摘要 self.logger.log_summary(total, passed, failed, skipped, total_time) # 生成报告 self._generate_reports(summary) return { "total": total, "passed": passed, "failed": failed, "skipped": skipped, "pass_rate": pass_rate, "total_time": total_time } def _generate_reports(self, summary: TestSummary) -> None: """ 生成测试报告 Args: summary: 测试摘要 """ self.logger.info("生成测试报告...") try: reports = self.report_generator.generate_all_reports(self.test_results, summary) for format_type, path in reports.items(): self.logger.info(f"✅ {format_type.upper()}报告已生成: {path}") except Exception as e: self.logger.error(f"生成报告失败: {str(e)}") def cleanup(self) -> None: """清理资源""" self.tester.close() self.logger.info("测试完成，资源已清理") def main(): """主函数""" print("="*60) print("前后端API交互自动化测试工具") print("="*60) print() # 显示配置信息 print("配置信息:") print(f" API基础URL: {config.api_base_url}") print(f" 超时时间: {config.api_timeout}秒") print(f" 最大重试次数: {config.api_max_retries}") print(f" 报告输出目录: {config.report_output_dir}") print(f" 报告格式: {', '.join(config.report_formats)}") print() # 创建测试运行器 runner = TestRunner() try: # 运行所有测试 results = runner.run_all_tests() # 显示最终结果 print() print("="*60) print("测试完成!") print("="*60) print(f"总计: {results['total']}") print(f"通过: {results['passed']}") print(f"失败: {results['failed']}") print(f"通过率: {results['pass_rate']:.2f}%") print(f"总耗时: {results['total_time']:.2f}ms") print("="*60) # 返回退出码 sys.exit(0 if results['failed'] == 0 else 1) except KeyboardInterrupt: print("\n\n测试被用户中断") sys.exit(130) except Exception as e: print(f"\n\n测试执行失败: {str(e)}") sys.exit(1) finally: runner.cleanup() if __name__ == "__main__": main()