#!/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_enhanced import ( APITestCases, ParameterizedTestCases, BoundaryTestCases, ExceptionTestCases ) class EnhancedTestRunner: """增强测试运行器""" def __init__(self): """初始化测试运行器""" self.logger = LoggerFactory.get_logger( "enhanced_test_runner", config.logging_file, config.logging_level, config.logging_console ) self.tester = APITester(self.logger, auto_auth=True) self.api_test_cases = APITestCases(self.tester) self.parameterized_test_cases = ParameterizedTestCases(self.tester) self.boundary_test_cases = BoundaryTestCases(self.tester) self.exception_test_cases = ExceptionTestCases(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_basic_tests(self) -> Dict[str, bool]: """ 运行基础测试用例 Returns: 测试结果字典 """ self.logger.info("="*60) self.logger.info("运行基础测试用例") self.logger.info("="*60) return self.api_test_cases.run_all_tests() def run_parameterized_tests(self) -> Dict[str, Dict[str, bool]]: """ 运行参数化测试用例 Returns: 测试结果字典 """ self.logger.info("="*60) self.logger.info("运行参数化测试用例") self.logger.info("="*60) results = {} # 测试不同凭据登录 credentials = [ {"username": "admin", "password": "admin123"}, {"username": "test", "password": "test123"} ] results["不同凭据登录"] = self.parameterized_test_cases.test_login_with_different_credentials(credentials) # 测试不同页面大小 page_sizes = [10, 20, 50, 100] results["分页功能"] = self.parameterized_test_cases.test_pagination(page_sizes) return results def run_boundary_tests(self) -> Dict[str, Dict[str, bool]]: """ 运行边界条件测试用例 Returns: 测试结果字典 """ self.logger.info("="*60) self.logger.info("运行边界条件测试用例") self.logger.info("="*60) results = {} # 测试用户名长度边界 results["用户名长度边界"] = self.boundary_test_cases.test_user_name_length() # 测试页面大小边界 results["页面大小边界"] = self.boundary_test_cases.test_page_size_boundaries() return results def run_exception_tests(self) -> Dict[str, Dict[str, bool]]: """ 运行异常场景测试用例 Returns: 测试结果字典 """ self.logger.info("="*60) self.logger.info("运行异常场景测试用例") self.logger.info("="*60) results = {} # 测试无效凭据 results["无效凭据"] = self.exception_test_cases.test_invalid_credentials() # 测试缺少必填字段 results["缺少必填字段"] = self.exception_test_cases.test_missing_required_fields() # 测试未授权访问 results["未授权访问"] = self.exception_test_cases.test_unauthorized_access() # 测试无效HTTP方法 results["无效HTTP方法"] = self.exception_test_cases.test_invalid_http_methods() return results 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. 基础测试 basic_results = self.run_basic_tests() for test_name, passed in basic_results.items(): full_test_name = f"基础测试 - {test_name}" self._record_test_result(full_test_name, passed) # 2. 参数化测试 param_results = self.run_parameterized_tests() for category, results in param_results.items(): for test_name, passed in results.items(): full_test_name = f"参数化测试 - {category} - {test_name}" self._record_test_result(full_test_name, passed) # 3. 边界条件测试 boundary_results = self.run_boundary_tests() for category, results in boundary_results.items(): for test_name, passed in results.items(): full_test_name = f"边界条件测试 - {category} - {test_name}" self._record_test_result(full_test_name, passed) # 4. 异常场景测试 exception_results = self.run_exception_tests() for category, results in exception_results.items(): for test_name, passed in results.items(): full_test_name = f"异常场景测试 - {category} - {test_name}" self._record_test_result(full_test_name, passed) # 计算测试摘要 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 _record_test_result(self, test_name: str, passed: bool, response_time: float = 0) -> None: """ 记录测试结果 Args: test_name: 测试名称 passed: 是否通过 response_time: 响应时间 """ 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=response_time, 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) 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 = EnhancedTestRunner() 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()