ai-web-tester/tests/test_cases.py

#!/usr/bin/env python3
"""
测试用例模板 - 快速设计和运行多个测试（支持并行执行）
"""
import sys
sys.path.insert(0, ".")

from src import WebTester
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict, Any
import time


# ============================================================
# 测试用例配置
# ============================================================

TEST_CASES = [
    {
        "name": "Example.com 链接测试",
        "url": "http://47.99.105.253:8084",
        "goal": "填入账号admin 密码password，登录成功",
    },
    # 添加更多测试用例...
]


# ============================================================
# 测试执行器
# ============================================================

def run_single_case(case: Dict[str, Any], model: str = "claude", 
                    headless: bool = True) -> Dict[str, Any]:
    """运行单个测试用例（独立浏览器实例）"""
    name = case.get("name", "Unknown")
    url = case["url"]
    goal = case["goal"]
    
    result = {
        "name": name,
        "url": url,
        "goal": goal,
        "status": "failed",
    }
    
    try:
        with WebTester(model=model, headless=headless) as tester:
            tester.goto(url)
            test_result = tester.test(goal)
            result["status"] = "passed"
            result["steps"] = test_result["steps"]
            result["report"] = test_result["report"]
    except Exception as e:
        result["error"] = str(e)
    
    return result


def run_tests(model: str = "claude", headless: bool = False):
    """串行运行所有测试用例"""
    results = []
    
    with WebTester(model=model, headless=headless) as tester:
        for i, case in enumerate(TEST_CASES, 1):
            name = case.get("name", f"Test {i}")
            url = case["url"]
            goal = case["goal"]
            
            print(f"\n{'='*60}")
            print(f"🧪 [{i}/{len(TEST_CASES)}] {name}")
            print(f"   URL: {url}")
            print(f"   Goal: {goal}")
            print(f"{'='*60}")
            
            try:
                tester.goto(url)
                result = tester.test(goal)
                
                # 检查所有步骤是否成功
                all_passed = all(r.get("success", False) for r in result.get("results", []))
                failed_count = sum(1 for r in result.get("results", []) if not r.get("success", False))
                
                if all_passed:
                    print(f"✅ 完成: {result['steps']} 步骤")
                    status = "passed"
                else:
                    print(f"⚠️ 部分失败: {failed_count}/{result['steps']} 步骤失败")
                    status = "failed"
                
                print(f"📄 报告: {result['report']}")
                
                results.append({
                    "name": name,
                    "status": status,
                    "steps": result["steps"],
                    "report": result["report"],
                })
            except Exception as e:
                print(f"❌ 失败: {e}")
                results.append({
                    "name": name,
                    "status": "failed",
                    "error": str(e),
                })
    
    _print_summary(results)
    return results


def run_tests_parallel(model: str = "claude", max_workers: int = 3):
    """
    并行运行所有测试用例
    
    Args:
        model: AI 模型
        max_workers: 最大并行数（默认 3）
    """
    print(f"\n🚀 并行模式启动 (workers={max_workers})")
    print(f"📋 待执行测试: {len(TEST_CASES)} 个\n")
    
    results = []
    start_time = time.time()
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        future_to_case = {
            executor.submit(run_single_case, case, model, True): case
            for case in TEST_CASES
        }
        
        # 收集结果
        for future in as_completed(future_to_case):
            case = future_to_case[future]
            try:
                result = future.result()
                status = "✅" if result["status"] == "passed" else "❌"
                print(f"{status} {result['name']}")
                results.append(result)
            except Exception as e:
                print(f"❌ {case['name']}: {e}")
                results.append({
                    "name": case["name"],
                    "status": "failed",
                    "error": str(e),
                })
    
    elapsed = time.time() - start_time
    print(f"\n⏱️  总耗时: {elapsed:.1f}秒")
    
    _print_summary(results)
    return results


def _print_summary(results: List[Dict[str, Any]]):
    """打印测试总结"""
    print(f"\n{'='*60}")
    print("📊 测试总结")
    print(f"{'='*60}")
    passed = sum(1 for r in results if r["status"] == "passed")
    failed = len(results) - passed
    print(f"✅ 通过: {passed}")
    print(f"❌ 失败: {failed}")
    if results:
        print(f"📈 通过率: {passed/len(results)*100:.1f}%")


def run_single_test(url: str, goal: str, model: str = "claude"):
    """运行单个测试"""
    with WebTester(model=model) as tester:
        tester.goto(url)
        result = tester.test(goal)
        print(f"✅ 完成: {result['steps']} 步骤")
        print(f"📄 报告: {result['report']}")
        return result


# ============================================================
# 主入口
# ============================================================

if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="AI Web Tester - 测试用例运行器")
    parser.add_argument("--url", help="单个测试的 URL")
    parser.add_argument("--goal", help="单个测试的目标描述")
    parser.add_argument("--model", default="claude", choices=["claude", "openai"], help="AI 模型")
    parser.add_argument("--headless", action="store_true", help="无头模式运行")
    parser.add_argument("--parallel", action="store_true", help="并行执行测试")
    parser.add_argument("--workers", type=int, default=3, help="并行工作线程数")
    
    args = parser.parse_args()
    
    if args.url and args.goal:
        run_single_test(args.url, args.goal, args.model)
    elif args.parallel:
        run_tests_parallel(model=args.model, max_workers=args.workers)
    else:
        run_tests(model=args.model, headless=args.headless)