everything-is-suitable/everything-is-suitable-test/test-tools/main_enhanced.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_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()