everything-is-suitable/everything-is-suitable-test/test-tools/main.py

#!/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()