#!/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": "登录", "url": "http://47.99.105.253:8084", "mode": "goal", # 目标驱动模式 "goal": "填入账号admin 密码password，登录成功", }, # 探索模式: AI 自主发现功能 { "name": "功能探索", "url": "http://47.99.105.253:8084", "mode": "explore", # 探索模式 "config": { "max_depth": 3, "max_clicks": 30, "dangerous_patterns": ["删除", "移除", "退出", "注销"], # 记录但不执行 "require_login": { # 需要先登录 "goal": "填入账号admin 密码password，登录成功" } } }, # 混合模式: 先执行目标，再探索 # { # "name": "登录后探索", # "url": "http://47.99.105.253:8084", # "mode": "hybrid", # "steps": [ # {"action": "goal", "goal": "登录"}, # {"action": "explore", "config": {"max_clicks": 10}} # ] # }, ] # ============================================================ # 测试执行器 # ============================================================ 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"] mode = case.get("mode", "goal") result = { "name": name, "url": url, "mode": mode, "status": "failed", } try: with WebTester(model=model, headless=headless) as tester: tester.goto(url) if mode == "goal": # 目标模式 goal = case.get("goal", "") test_result = tester.test(goal) result["status"] = "passed" result["steps"] = test_result["steps"] result["report"] = test_result["report"] elif mode == "explore": # 探索模式 config = case.get("config", {}) # 如果需要先登录 require_login = config.pop("require_login", None) if require_login: login_goal = require_login.get("goal", "") if login_goal: tester.test(login_goal) tester.browser.wait(1000) # 执行探索 explore_result = tester.explore(config) result["status"] = "passed" result["elements"] = len(explore_result.get("discovered_elements", [])) result["bugs"] = len(explore_result.get("bug_list", [])) result["report"] = explore_result.get("report", "") elif mode == "hybrid": # 混合模式 for step in case.get("steps", []): if step.get("action") == "goal": tester.test(step["goal"]) elif step.get("action") == "explore": tester.explore(step.get("config", {})) result["status"] = "passed" 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"] mode = case.get("mode", "goal") print(f"\n{'='*60}") print(f"🧪 [{i}/{len(TEST_CASES)}] {name}") print(f" URL: {url}") print(f" Mode: {mode}") print(f"{'='*60}") try: tester.goto(url) if mode == "goal": goal = case.get("goal", "") 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"], }) elif mode == "explore": config = case.get("config", {}).copy() # 如果需要先登录 require_login = config.pop("require_login", None) if require_login: login_goal = require_login.get("goal", "") if login_goal: print(f" 🔐 执行登录...") tester.test(login_goal) tester.browser.wait(1000) # 执行探索 print(f" 🔍 开始功能探索...") result = tester.explore(config) elements = len(result.get("discovered_elements", [])) bugs = len(result.get("bug_list", [])) print(f"✅ 探索完成: 发现 {elements} 个元素, {bugs} 个问题") print(f"📄 报告: {result.get('report', '')}") results.append({ "name": name, "status": "passed", "elements": elements, "bugs": bugs, "report": result.get("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)