run_all_tests.py

#!/usr/bin/env python3
"""
运行所有测试的统一脚本

这个脚本会按顺序运行所有测试套件：
1. 端到端集成测试
2. 错误自愈测试
3. 多轮对话测试
4. 本地环境验证
5. AgentCore 部署验证（可选）

使用方法:
    python run_all_tests.py                    # 运行所有测试（除了部署验证）
    python run_all_tests.py --all              # 运行所有测试（包括部署验证）
    python run_all_tests.py --suite e2e        # 只运行端到端测试
    python run_all_tests.py --suite error      # 只运行错误自愈测试
    python run_all_tests.py --suite multi      # 只运行多轮对话测试
    python run_all_tests.py --suite local      # 只运行本地验证
    python run_all_tests.py --suite deploy     # 只运行部署验证
"""

import sys
import subprocess
import argparse
import logging
from datetime import datetime
from nl2sql_agent import setup_logging


# 配置日志
setup_logging(level="INFO")
logger = logging.getLogger(__name__)


def print_separator(char="=", length=80):
    """打印分隔线"""
    print(char * length)


def print_header(title: str):
    """打印标题"""
    print()
    print_separator()
    print(f" {title}")
    print_separator()
    print()


def run_test_suite(script_name: str, suite_name: str) -> bool:
    """
    运行测试套件
    
    参数:
        script_name: 测试脚本名称
        suite_name: 测试套件名称
        
    返回:
        是否成功
    """
    print_header(f"运行测试套件: {suite_name}")
    
    logger.info(f"执行脚本: {script_name}")
    logger.info(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()
    
    try:
        result = subprocess.run(
            ["python", script_name],
            timeout=300  # 5 分钟超时
        )
        
        print()
        logger.info(f"结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        
        if result.returncode == 0:
            logger.info(f"✓ {suite_name} 测试通过")
            return True
        else:
            logger.error(f"✗ {suite_name} 测试失败（返回码: {result.returncode}）")
            return False
            
    except subprocess.TimeoutExpired:
        logger.error(f"✗ {suite_name} 测试超时")
        return False
    except Exception as e:
        logger.error(f"✗ {suite_name} 测试发生异常: {str(e)}")
        return False


def main():
    """
    主函数
    """
    # 解析命令行参数
    parser = argparse.ArgumentParser(
        description="运行所有测试套件",
        formatter_class=argparse.RawDescriptionHelpFormatter
    )
    
    parser.add_argument(
        "--suite",
        "-s",
        choices=["e2e", "error", "multi", "local", "deploy", "all"],
        help="指定要运行的测试套件"
    )
    
    parser.add_argument(
        "--all",
        "-a",
        action="store_true",
        help="运行所有测试（包括部署验证）"
    )
    
    parser.add_argument(
        "--skip-deploy",
        action="store_true",
        help="跳过部署验证测试"
    )
    
    args = parser.parse_args()
    
    # 打印标题
    print_separator("=")
    print(" NL2SQL Agent - 测试套件")
    print_separator("=")
    print()
    print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()
    
    # 定义测试套件
    test_suites = [
        ("test_e2e.py", "端到端集成测试", "e2e", False),
        ("test_error_recovery.py", "错误自愈测试", "error", False),
        ("test_multi_turn.py", "多轮对话测试", "multi", False),
        ("test_local_validation.py", "本地环境验证", "local", False),
        ("test_agentcore_deployment.py", "AgentCore 部署验证", "deploy", True),
    ]
    
    # 确定要运行的测试
    suites_to_run = []
    
    if args.suite:
        # 运行指定的测试套件
        if args.suite == "all":
            suites_to_run = test_suites
        else:
            suites_to_run = [s for s in test_suites if s[2] == args.suite]
    elif args.all:
        # 运行所有测试
        suites_to_run = test_suites
    else:
        # 默认：运行除部署验证外的所有测试
        suites_to_run = [s for s in test_suites if not s[3] or not args.skip_deploy]
    
    if not suites_to_run:
        logger.error("没有找到要运行的测试套件")
        sys.exit(1)
    
    # 显示将要运行的测试
    logger.info("将要运行以下测试套件:")
    for script, name, _, _ in suites_to_run:
        logger.info(f"  - {name} ({script})")
    print()
    
    # 运行测试
    results = []
    
    for script, name, suite_id, is_optional in suites_to_run:
        try:
            result = run_test_suite(script, name)
            results.append((name, result, is_optional))
        except Exception as e:
            logger.error(f"运行 {name} 时发生异常: {str(e)}")
            results.append((name, False, is_optional))
    
    # 输出测试结果摘要
    print()
    print_separator("=")
    print(" 测试结果摘要")
    print_separator("=")
    print()
    
    passed = 0
    failed = 0
    optional_failed = 0
    
    for name, result, is_optional in results:
        if result:
            status = "✓ 通过"
            passed += 1
        else:
            if is_optional:
                status = "⚠ 失败（可选）"
                optional_failed += 1
            else:
                status = "✗ 失败"
                failed += 1
        
        logger.info(f"{status}: {name}")
    
    print()
    print_separator("=")
    logger.info(f"总计: {passed} 通过, {failed} 失败, {optional_failed} 可选失败")
    logger.info(f"结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print_separator("=")
    
    # 输出建议
    if failed > 0:
        print()
        logger.warning("部分测试失败，请检查上述错误信息")
        logger.info("提示:")
        logger.info("  - 确保数据库配置正确（rds_demo/db_config.txt）")
        logger.info("  - 确保数据库可访问")
        logger.info("  - 确保所有依赖已安装（pip install -r requirements.txt）")
        logger.info("  - 查看详细日志以了解失败原因")
    elif optional_failed > 0:
        print()
        logger.info("所有必需测试通过！")
        logger.info("部分可选测试失败（通常是部署相关的测试）")
        logger.info("如需运行部署测试，请确保:")
        logger.info("  - AgentCore CLI 已安装")
        logger.info("  - AWS 凭证已配置")
        logger.info("  - Agent 已部署到 AgentCore")
    else:
        print()
        logger.info("🎉 所有测试通过！")
    
    # 返回退出码
    sys.exit(0 if failed == 0 else 1)


if __name__ == "__main__":
    main()