xst/QaAutomationHub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4f30b9d7028c87f5473af1a1b934ff1eb0407fc7
QaAutomationHub/scripts/fleet_runner.py
T
xst 4f30b9d702 feat: Execute zone - Playwright PC Web + Appium mobile + AI screenshot diff + test conclusion reports 
- Add 3 new agents: web-executor, mobile-executor, result-reporter
- web-executor: Playwright multi-browser (Chromium/Firefox/WebKit) automated scripts
- mobile-executor: Appium dual-platform (Android/iOS) automated scripts
- result-reporter: pixel-level AI visual comparison + test conclusion report with screenshots
- Auto-generate executable Python test scripts from Markdown test cases
- Screenshot strategies: on-failure (default) + on-step (optional)
- Visual diff grading: <1% pass / 1-5% minor / 5-15% UI_DIFF / >15% blocker
- Failure multi-classification: REAL_BUG/UI_DIFF/ENV_ISSUE/DATA_ISSUE/CASE_BUG/SCRIPT_ERROR
- requirements.txt: add playwright + Pillow
- Fleet now: 17 agents across 6 battle zones
2026-07-10 11:03:58 +08:00

1694 lines
64 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
Agentic QE Fleet — 核心编排器
职责: 按战区顺序编排 14 个专业 Agent，协调 manifest 交接，
执行质量门禁裁决，触发 Excel 导出和知识沉淀。
用法:
python3 scripts/fleet_runner.py run --requirement <路径>
python3 scripts/fleet_runner.py prepare --requirement <路径>
python3 scripts/fleet_runner.py analyze --requirement <路径>
python3 scripts/fleet_runner.py design --requirement <路径>
python3 scripts/fleet_runner.py review --requirement <路径>
python3 scripts/fleet_runner.py export --requirement <路径>
python3 scripts/fleet_runner.py monitor --requirement <路径> --results <测试结果>
python3 scripts/fleet_runner.py status --requirement <路径>
"""
from __future__ import annotations
import argparse
import json
import sys
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# 复用现有模块
from case_pipeline import (
resolve_requirement_path,
validate_requirement_file,
validate_knowledge_base,
validate_project_profile,
build_paths as build_legacy_paths,
ensure_output_dirs,
write_normalized_document,
select_technical_solution_files,
safe_read_text,
to_repo_relative,
REPO_ROOT,
REQUIREMENTS_DIR,
RAW_REQUIREMENTS_DIR,
TECHNICAL_SOLUTIONS_DIR,
PROJECT_PROFILE_FILE,
)
from export_excel import export_markdown_to_excel
# Fleet 自有模块
from fleet_manifest import (
save_manifest,
load_manifest,
load_manifest_safe,
build_merged_manifest,
get_zone_status,
mark_zone_completed,
manifest_path,
ZONE_ORDER,
)
from fleet_agents import (
AGENT_REGISTRY,
load_agent_prompt,
list_zone_agents,
validate_all_agents,
render_agent_prompt,
)
# ── 配置加载 ──────────────────────────────────────────────────────────────
try:
import yaml
_HAS_YAML = True
except ImportError:
_HAS_YAML = False
def load_fleet_config() -> dict[str, Any]:
"""加载 fleet_config.yml。"""
config_path = REPO_ROOT / "fleet_config.yml"
if not config_path.exists():
return _default_fleet_config()
if _HAS_YAML:
with open(config_path, "r", encoding="utf-8") as fh:
return yaml.safe_load(fh) or {}
else:
# 纯 Python fallback: 解析简单的 YAML 子集
return _parse_simple_yaml_config(config_path)
def _default_fleet_config() -> dict[str, Any]:
return {
"battle_zones": {
"prepare": {"enabled": True},
"analyze": {"enabled": True},
"design": {"enabled": True},
"review": {"enabled": True},
"monitor": {"enabled": True, "auto_confirm": True},
},
"quality_gate": {"min_coverage": 0.95, "max_blockers": 0},
"output": {"excel_format": "yunxiao", "snapshot_keep": 3, "auto_sync_maintained": True},
}
def _parse_simple_yaml_config(path: Path) -> dict[str, Any]:
"""简易 YAML 解析器（当 PyYAML 不可用时）。"""
import re
result: dict[str, Any] = {}
current: dict[str, Any] = result
stack: list[tuple[str, dict[str, Any]]] = []
for raw_line in path.read_text(encoding="utf-8").splitlines():
line = raw_line.rstrip()
if not line or line.strip().startswith("#"):
continue
indent = len(raw_line) - len(raw_line.lstrip())
key_match = re.match(r"^(\s*)([\w_-]+)\s*:\s*(.*)", line)
if not key_match:
continue
key = key_match.group(2)
value = key_match.group(3).strip().strip('"').strip("'")
# 处理缩进回退
while stack and stack[-1][0] >= indent:
stack.pop()
current = stack[-1][1] if stack else result
if value in ("true", "True"):
current[key] = True
elif value in ("false", "False"):
current[key] = False
elif value == "" or value == "{}":
current[key] = {}
current = current[key]
stack.append((indent, current))
elif value == "[]":
current[key] = []
elif re.match(r"^- ", raw_line):
if key not in current:
current[key] = []
current[key].append(value)
elif re.match(r"^\d+(\.\d+)?$", value):
current[key] = float(value) if "." in value else int(value)
else:
current[key] = value
return result
# ── 路径和目录 ────────────────────────────────────────────────────────────
BUILTIN_INPUT_DIRS = [
REPO_ROOT / "output" / "analysis",
REPO_ROOT / "output" / "test_points",
REPO_ROOT / "output" / "test_cases",
REPO_ROOT / "output" / "excel_reports",
REPO_ROOT / "output" / "normalized_inputs",
REPO_ROOT / "output" / "versions",
]
def build_all_output_dirs(base_name: str) -> None:
"""创建所有输出目录。"""
legacy_paths = build_legacy_paths(base_name)
ensure_output_dirs(legacy_paths)
# 确保 manifest 目录
(REPO_ROOT / "output" / "manifests").mkdir(parents=True, exist_ok=True)
# ── 编排核心 ──────────────────────────────────────────────────────────────
def run_zone(zone: str, base_name: str, requirement_path: Path, config: dict[str, Any]) -> dict[str, Any]:
"""运行一个战区，返回该战区的 manifest 数据。"""
zone_config = config.get("battle_zones", {}).get(zone, {})
if not zone_config.get("enabled", True):
print(f"⏭️ {zone.upper()} 战区已禁用，跳过。")
return {}
print(f"\n{'='*60}")
print(f"🚀 {zone.upper()} 战区启动")
print(f"{'='*60}")
agents = list_zone_agents(zone)
print(f"📋 Agent: {', '.join(agents)}")
# 加载前序战区 manifest 作为上下文
context = _build_zone_context(base_name, zone, requirement_path)
# 打印当前可用的输入文件
_print_context_files(context, zone)
# 战区特定逻辑
zone_handlers = {
"prepare": _run_prepare_zone,
"analyze": _run_analyze_zone,
"design": _run_design_zone,
"execute": _run_execute_zone,
"review": _run_review_zone,
"monitor": _run_monitor_zone,
}
handler = zone_handlers.get(zone)
if handler is None:
raise ValueError(f"未知战区: {zone}")
result = handler(base_name, requirement_path, context, config, agents)
# 标记完成
if result:
mark_zone_completed(base_name, zone)
print(f"✅ {zone.upper()} 战区完成 → {manifest_path(base_name, zone)}")
return result
def _build_zone_context(base_name: str, zone: str, requirement_path: Path) -> dict[str, Any]:
"""构建战区运行的上下文。"""
context: dict[str, Any] = {
"base_name": base_name,
"requirement_file": str(requirement_path),
"requirement_stem": requirement_path.stem,
"repo_root": str(REPO_ROOT),
"project_profile": str(PROJECT_PROFILE_FILE),
}
# 加载前序战区 manifest
for prev_zone in ZONE_ORDER:
if prev_zone == zone:
break
manifest = load_manifest_safe(base_name, prev_zone)
if manifest:
context[f"manifest_{prev_zone}"] = manifest
context[f"manifest_{prev_zone}_path"] = str(manifest_path(base_name, prev_zone))
return context
def _print_context_files(context: dict[str, Any], zone: str) -> None:
"""打印战区可用的上下文文件。"""
from fleet_manifest import load_manifest_safe
items: list[tuple[str, str]] = []
# 从 manifest 中提取产物文件
for prev_zone in ZONE_ORDER:
if prev_zone == zone:
break
manifest = context.get(f"manifest_{prev_zone}")
if manifest is None:
continue
# 收集文件路径
for key in manifest:
if key.endswith("_file") or key.endswith("_files"):
value = manifest[key]
if isinstance(value, str) and Path(value).exists():
items.append((key, value))
elif isinstance(value, list):
for v in value:
if isinstance(v, str) and Path(v).exists():
items.append((key, v))
if items:
print("📂 可用上下文文件:")
for key, path in items[:10]:
print(f" - {key}: {path}")
# ── Prepare 战区 ──────────────────────────────────────────────────────────
def _run_prepare_zone(
base_name: str,
requirement_path: Path,
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
) -> dict[str, Any]:
"""执行 Prepare 战区: document-parser → knowledge-activator。"""
# 1. 文档解析 (借用现有 case_pipeline 的能力)
validate_requirement_file(requirement_path)
validate_knowledge_base()
validate_project_profile()
technical_solution_files = select_technical_solution_files(requirement_path)
normalized_dir = REPO_ROOT / "output" / "normalized_inputs" / base_name
normalized_dir.mkdir(parents=True, exist_ok=True)
normalized_requirement_file = write_normalized_document(
source_path=requirement_path,
target_path=normalized_dir / "requirement.md",
document_role="需求文档",
)
normalized_technical_solution_files = [
write_normalized_document(
source_path=path,
target_path=normalized_dir / f"technical_solution_{idx:02d}.md",
document_role="技术方案",
)
for idx, path in enumerate(technical_solution_files, start=1)
]
# 文档置信度评估
requirement_body = safe_read_text(normalized_requirement_file)
confidence = _estimate_document_confidence(requirement_body, requirement_path.suffix.lower())
# 2. 知识激活
activated_knowledge = _activate_knowledge(requirement_path, config)
# 检测知识缺口
knowledge_gaps = _detect_knowledge_gaps(activated_knowledge, requirement_body)
# 组装 manifest
prepare_manifest = {
"base_name": base_name,
"requirement_source_file": str(requirement_path),
"requirement_input_type": requirement_path.suffix.lower().lstrip("."),
"normalized_requirement_file": str(normalized_requirement_file),
"normalized_dir": str(normalized_dir),
"technical_solution_files": [str(p) for p in technical_solution_files],
"normalized_technical_solution_files": [str(p) for p in normalized_technical_solution_files],
"project_profile_file": str(PROJECT_PROFILE_FILE),
"document_confidence": confidence,
"activated_knowledge": activated_knowledge,
"knowledge_gaps": knowledge_gaps,
"agent_notes": {
"document-parser": f"解析完成，置信度 {confidence.get('requirement', 0):.0%}",
"knowledge-activator": f"激活 {len(activated_knowledge.get('terminology', {}).get('permanent', []))} 常驻 + "
f"{len(activated_knowledge.get('terminology', {}).get('optional', []))} 可选术语",
},
}
save_manifest(base_name, "prepare", prepare_manifest)
return prepare_manifest
def _estimate_document_confidence(text: str, suffix: str) -> dict[str, Any]:
"""估算文档解析置信度。"""
confidence = 1.0 if suffix == ".md" else 0.90 if suffix == ".docx" else 0.80 if suffix == ".pdf" else 0.75
if not text.strip():
confidence = 0.1
elif "> ⚠️ 待确认：PDF 未提取到可用文本" in text:
confidence = 0.05
elif text.count("> ⚠️") > 3:
confidence -= 0.1
return {
"requirement": round(max(0.0, confidence), 2),
"issues": ["扫描件/图片型PDF"] if "未提取到可用文本" in text else [],
}
def _activate_knowledge(requirement_path: Path, config: dict[str, Any]) -> dict[str, Any]:
"""按需求内容自动激活知识库文件。"""
from case_pipeline import (
CORE_TERMINOLOGY_FILE,
KNOWLEDGE_BASE_FILES,
OPTIONAL_TERMINOLOGY_RULES,
)
requirement_text = safe_read_text(requirement_path)
combined = f"{requirement_path.name}\n{requirement_text}"
# 常驻术语（始终激活）
permanent = [str(CORE_TERMINOLOGY_FILE)]
# 关键词匹配可选术语
optional: list[dict[str, Any]] = []
for rule in OPTIONAL_TERMINOLOGY_RULES:
matched_keywords = [kw for kw in rule["keywords"] if kw in combined]
if matched_keywords:
optional.append({
"name": rule["name"],
"path": str(rule["path"]),
"matched_keywords": matched_keywords[:10],
"activation_reason": f"匹配关键词: {', '.join(matched_keywords[:5])}",
})
# 语义匹配知识库（基于 Jaccard 相似度）
from case_pipeline import to_ngrams, jaccard_similarity
requirement_tokens = to_ngrams(combined)
semantic_matches: list[dict[str, Any]] = []
threshold = config.get("knowledge_activation", {}).get("semantic_match_threshold", 0.08)
for kb_file in KNOWLEDGE_BASE_FILES:
if kb_file == CORE_TERMINOLOGY_FILE:
continue
if not kb_file.exists():
continue
kb_text = safe_read_text(kb_file)
score = jaccard_similarity(requirement_tokens, to_ngrams(f"{kb_file.name}\n{kb_text}"))
if score >= threshold:
semantic_matches.append({
"path": str(kb_file),
"score": round(score, 4),
"category": _classify_kb_file(kb_file),
})
return {
"terminology": {
"permanent": permanent,
"optional": optional,
},
"semantic_matches": sorted(semantic_matches, key=lambda x: x["score"], reverse=True),
}
def _classify_kb_file(path: Path) -> str:
"""分类知识库文件。"""
path_str = str(path)
if "02_history" in path_str:
return "history"
if "03_best_practices" in path_str:
return "best_practice"
if "01_standards" in path_str:
return "standard"
return "other"
def _detect_knowledge_gaps(activated_knowledge: dict[str, Any], requirement_text: str) -> list[dict[str, str]]:
"""检测知识库缺口。"""
gaps: list[dict[str, str]] = []
# 检查是否有必要的知识类别缺失
has_history = any(
m.get("category") == "history"
for m in activated_knowledge.get("semantic_matches", [])
)
has_best_practice = any(
m.get("category") == "best_practice"
for m in activated_knowledge.get("semantic_matches", [])
)
if not has_history:
gaps.append({"category": "history", "suggestion": "未激活任何历史缺陷/易漏场景，建议补充相关知识库条目"})
if not has_best_practice:
gaps.append({"category": "best_practice", "suggestion": "未激活任何最佳实践范例，建议补充同类型需求的优秀用例"})
return gaps
# ── Analyze 战区 ──────────────────────────────────────────────────────────
def _run_analyze_zone(
base_name: str,
requirement_path: Path,
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
) -> dict[str, Any]:
"""执行 Analyze 战区: requirement-analyzer + conflict-detector → risk-assessor。"""
prepare_manifest = context.get("manifest_prepare", {})
from case_pipeline import (
find_related_requirements,
build_conflict_candidates,
build_confirmation_gate,
write_relation_report,
find_relevant_decision_files,
inspect_decision_file,
)
# 加载标准化需求
normalized_requirement_file = Path(
prepare_manifest.get("normalized_requirement_file",
f"output/normalized_inputs/{base_name}/requirement.md")
)
if not normalized_requirement_file.is_absolute():
normalized_requirement_file = REPO_ROOT / normalized_requirement_file
technical_solution_files = [
Path(p) for p in prepare_manifest.get("normalized_technical_solution_files", [])
]
# 关联需求识别
related_requirements = find_related_requirements(requirement_path)
print(f"📚 关联需求: {len(related_requirements)} 个")
# 冲突检测
conflicts = build_conflict_candidates(requirement_path, related_requirements)
print(f"⚠️ 冲突候选: {len(conflicts)} 个")
# 写入关联与冲突报告
analysis_dir = REPO_ROOT / "output" / "analysis"
analysis_dir.mkdir(parents=True, exist_ok=True)
relation_report_path = analysis_dir / f"{base_name}_关联与冲突.md"
write_relation_report(
requirement_path, related_requirements,
[Path(p) for p in prepare_manifest.get("technical_solution_files", [])],
conflicts, relation_report_path,
)
# 风险评估 (risk-assessor 的输入)
risk_matrix = _build_risk_matrix(requirement_path, conflicts, related_requirements)
risk_report_path = analysis_dir / f"{base_name}_风险评估.md"
_write_risk_report(risk_matrix, risk_report_path, base_name)
# 确认门禁
confirmation_gate = build_confirmation_gate(
requirement_path=requirement_path,
base_name=base_name,
related_requirements=related_requirements,
conflicts=conflicts,
normalized_requirement_file=normalized_requirement_file,
normalized_technical_solution_files=[
p for p in technical_solution_files if isinstance(p, Path)
],
)
analyze_manifest = {
"base_name": base_name,
"analysis_file": str(analysis_dir / f"{base_name}_分析.md"),
"relation_report_file": str(relation_report_path),
"risk_report_file": str(risk_report_path),
"related_requirements": [
{"path": str(item["path"]), "similarity": round(item["score"], 6)}
for item in related_requirements
],
"conflict_candidates_count": len(conflicts),
"conflict_summary": _summarize_conflicts(conflicts),
"risk_matrix": risk_matrix,
"confirmation_gate": confirmation_gate,
"agent_notes": {
"requirement-analyzer": f"识别 {len(related_requirements)} 个关联需求",
"conflict-detector": f"检测到 {len(conflicts)} 个冲突候选",
"risk-assessor": f"识别 {len(risk_matrix.get('risks', []))} 个风险项",
},
}
save_manifest(base_name, "analyze", analyze_manifest)
return analyze_manifest
def _build_risk_matrix(
requirement_path: Path,
conflicts: list[dict[str, Any]],
related_requirements: list[dict[str, Any]],
) -> dict[str, Any]:
"""构建风险矩阵。"""
risks: list[dict[str, Any]] = []
requirement_text = safe_read_text(requirement_path)
# 风险检测维度
checks = [
("资损", ["金额", "支付", "退款", "扣减", "优惠", "积分", "券", "库存", "手续费", "税费"], "financial"),
("可用性", ["超时", "弱网", "并发", "降级", "熔断", "限流", "重试", "幂等"], "availability"),
("数据", ["脏数据", "迁移", "精度", "隔离", "删除", "空值", "租户"], "data"),
("合规", ["鉴权", "留痕", "审批", "实名", "隐私", "加密", "脱敏"], "compliance"),
("兼容性", ["H5", "小程序", "App", "浏览器", "多端", "版本", "灰度"], "compatibility"),
]
for category, keywords, risk_id in checks:
matched = [kw for kw in keywords if kw in requirement_text]
if matched:
likelihood = min(5, len(matched))
impact = 5 if risk_id in ("financial", "compliance") else 4 if risk_id in ("availability", "data") else 3
risks.append({
"id": f"RISK-{risk_id.upper()}",
"category": category,
"keywords_matched": matched,
"likelihood": likelihood,
"impact": impact,
"score": likelihood * impact,
"level": "P0" if likelihood * impact >= 15 else "P1" if likelihood * impact >= 10 else "P2",
"conflict_amplified": bool(conflicts),
})
# 冲突放大风险
if conflicts:
for risk in risks:
if risk["category"] in ("资损", "数据", "合规"):
risk["conflict_amplified"] = True
risk["score"] = min(25, risk["score"] + 3)
risk["level"] = "P0" if risk["score"] >= 15 else risk["level"]
return {
"risks": sorted(risks, key=lambda r: r["score"], reverse=True),
"total": len(risks),
"p0_count": sum(1 for r in risks if r["level"] == "P0"),
"p1_count": sum(1 for r in risks if r["level"] == "P1"),
}
def _write_risk_report(risk_matrix: dict[str, Any], report_path: Path, base_name: str) -> None:
"""写入风险评估报告。"""
lines = [
f"# {base_name} 风险评估报告",
"",
f"> 生成时间: {datetime.now(timezone.utc).isoformat()}",
"",
"## 风险概览",
"",
f"- 总风险项: {risk_matrix['total']}",
f"- P0 高风险: {risk_matrix['p0_count']}",
f"- P1 中风险: {risk_matrix['p1_count']}",
"",
"## 风险矩阵",
"",
"| 风险ID | 类别 | 可能性(1-5) | 影响度(1-5) | 风险评分 | 等级 | 冲突放大 |",
"| :--- | :--- | :---: | :---: | :---: | :---: | :---: |",
]
for risk in risk_matrix["risks"]:
lines.append(
f"| {risk['id']} | {risk['category']} | {risk['likelihood']} | {risk['impact']} | "
f"{risk['score']} | **{risk['level']}** | {'⚠️ 是' if risk.get('conflict_amplified') else '否'} |"
)
lines.extend([
"",
"## 风险缓解建议",
"",
])
for risk in risk_matrix["risks"]:
if risk["level"] == "P0":
lines.append(f"- **{risk['id']} ({risk['category']})**: 必须 100% 覆盖，建议增加 P0 测试点和专项回归用例。")
report_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
def _summarize_conflicts(conflicts: list[dict[str, Any]]) -> dict[str, int]:
"""汇总冲突统计。"""
types: dict[str, int] = {}
for c in conflicts:
t = c.get("type", "未知")
types[t] = types.get(t, 0) + 1
return types
# ── Design 战区 ──────────────────────────────────────────────────────────
def _run_design_zone(
base_name: str,
requirement_path: Path,
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
) -> dict[str, Any]:
"""执行 Design 战区: strategist → (testpoint-designer + data-builder) → case-designer。"""
prepare_manifest = context.get("manifest_prepare", {})
analyze_manifest = context.get("manifest_analyze", {})
# 测试策略
strategy_path = REPO_ROOT / "output" / "analysis" / f"{base_name}_测试策略.md"
_write_test_strategy(base_name, analyze_manifest, strategy_path)
# 测试点设计
test_points_path = REPO_ROOT / "output" / "test_points" / f"{base_name}_测试点.md"
test_points_path.parent.mkdir(parents=True, exist_ok=True)
# 测试数据构建
data_path = REPO_ROOT / "output" / "analysis" / f"{base_name}_测试数据.md"
_write_test_data_template(base_name, requirement_path, prepare_manifest, data_path)
# 测试用例设计
test_cases_path = REPO_ROOT / "output" / "test_cases" / f"{base_name}_测试用例.md"
test_cases_path.parent.mkdir(parents=True, exist_ok=True)
risk_matrix = analyze_manifest.get("risk_matrix", {})
p0_count = risk_matrix.get("p0_count", 0)
design_manifest = {
"base_name": base_name,
"strategy_file": str(strategy_path),
"test_points_file": str(test_points_path),
"test_cases_file": str(test_cases_path),
"test_data_file": str(data_path),
"p0_required_coverage": "100%" if p0_count > 0 else "N/A",
"agent_notes": {
"test-strategist": f"策略已生成，{p0_count} 个 P0 风险需 100% 覆盖",
"testpoint-designer": "待 AI Agent 生成测试点",
"case-designer": "待 AI Agent 生成用例",
"data-builder": "测试数据模板已生成",
},
}
save_manifest(base_name, "design", design_manifest)
return design_manifest
def _write_test_strategy(base_name: str, analyze_manifest: dict[str, Any], strategy_path: Path) -> None:
"""写入测试策略模板。"""
risk_matrix = analyze_manifest.get("risk_matrix", {})
risks = risk_matrix.get("risks", [])
lines = [
f"# {base_name} 测试策略",
"",
"## 1. 测试金字塔",
"",
"| 层级 | 占比 | 覆盖重点 | 工具 |",
"| :--- | :---: | :--- | :--- |",
"| L1 单元测试 | 40% | 核心逻辑、计算、状态机 | 开发自测 |",
"| L2 API 测试 | 35% | 接口契约、参数校验、权限、幂等 | Postman/Pytest |",
"| L3 UI 测试 | 20% | 主流程、关键交互、端到端 | Playwright/Selenium |",
"| L4 手工探索 | 5% | 易用性、视觉、非确定性场景 | 人工 |",
"",
"## 2. P0 必测清单",
"",
]
for risk in risks:
if risk["level"] == "P0":
lines.append(f"- [{risk['id']}] **{risk['category']}**: {', '.join(risk.get('keywords_matched', []))}")
lines.extend([
"",
"## 3. 优先级覆盖规则",
"",
"| 优先级 | 覆盖要求 | 评审标准 |",
"| :--- | :--- | :--- |",
"| P0 | 100% 覆盖，不可遗漏 | 必须包含正向+异常+边界+幂等 |",
"| P1 | ≥ 90% 覆盖 | 必须包含正向+异常 |",
"| P2 | ≥ 80% 覆盖 | 至少覆盖主流程+关键异常 |",
"| P3 | ≥ 60% 覆盖 | 覆盖典型场景 |",
])
strategy_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
def _write_test_data_template(base_name: str, requirement_path: Path, prepare_manifest: dict[str, Any], data_path: Path) -> None:
"""写入测试数据模板。"""
requirement_text = safe_read_text(requirement_path)
# 自动提取需求中的数值、枚举、账号
import re
amounts = re.findall(r'\d+(?:\.\d+)?(?:元|分|%)', requirement_text)
ids = re.findall(r'(?:ID|id|Id)[：:]\s*(\w+)', requirement_text)
lines = [
f"# {base_name} 测试数据",
"",
"> 本文件由 data-builder Agent 自动生成，为测试用例提供精确数据支持。",
"",
"## 自动提取的候选数据",
"",
]
if amounts:
lines.append("### 金额/数值")
for a in amounts[:10]:
lines.append(f"- `{a}`")
lines.append("")
if ids:
lines.append("### ID/编码")
for i in ids[:10]:
lines.append(f"- `{i}`")
lines.append("")
lines.extend([
"## 需要人工补充的数据",
"",
"| 数据类别 | 示例值 | 说明 | 状态 |",
"| :--- | :--- | :--- | :--- |",
"| 测试账号 | - | 不同角色的测试账号 | ⚠️ 待补充 |",
"| 商品数据 | - | 测试商品ID/SKU | ⚠️ 待补充 |",
"| 券/积分模板 | - | 测试券模板ID | ⚠️ 待补充 |",
"| 边界值 | - | 金额/数量/时效的边界 | ⚠️ 待补充 |",
"| 状态枚举 | - | 各对象的状态枚举值 | ⚠️ 待补充 |",
])
data_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
# ── Execute 战区 ─────────────────────────────────────────────────────────
def _run_execute_zone(
base_name: str,
requirement_path: Path,
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
) -> dict[str, Any]:
"""执行 Execute 战区: web-executor + mobile-executor → result-reporter。"""
prepare_manifest = context.get("manifest_prepare", {})
design_manifest = context.get("manifest_design", {})
execute_config = config.get("battle_zones", {}).get("execute", {})
test_cases_path = Path(design_manifest.get("test_cases_file",
f"output/test_cases/{base_name}_测试用例.md"))
test_data_path = Path(design_manifest.get("test_data_file",
f"output/analysis/{base_name}_测试数据.md"))
# 输出目录
screenshots_dir = REPO_ROOT / "output" / "screenshots" / base_name
screenshots_dir.mkdir(parents=True, exist_ok=True)
execution_dir = REPO_ROOT / "output" / "execution"
execution_dir.mkdir(parents=True, exist_ok=True)
# 生成 Playwright 测试脚本
playwright_script = _generate_playwright_script(
base_name, test_cases_path, test_data_path,
screenshots_dir, execute_config
)
# 生成 Appium 测试脚本
appium_script = _generate_appium_script(
base_name, test_cases_path, test_data_path,
screenshots_dir, execute_config
)
# 生成综合执行报告
execution_report_path = execution_dir / f"{base_name}_执行报告.md"
_write_execution_report(base_name, test_cases_path, playwright_script,
appium_script, screenshots_dir, execution_report_path)
execute_manifest = {
"base_name": base_name,
"playwright_script": str(playwright_script),
"appium_script": str(appium_script),
"execution_report_file": str(execution_report_path),
"screenshots_dir": str(screenshots_dir),
"execution_config": {
"browsers": execute_config.get("browsers", ["chromium"]),
"mobile_platforms": execute_config.get("mobile_platforms", ["android"]),
"screenshot_on_failure": execute_config.get("screenshot_on_failure", True),
"screenshot_on_step": execute_config.get("screenshot_on_step", False),
},
"agent_notes": {
"web-executor": f"Playwright 脚本已生成 → {playwright_script}",
"mobile-executor": f"Appium 脚本已生成 → {appium_script}",
"result-reporter": f"执行报告 → {execution_report_path}",
},
}
save_manifest(base_name, "execute", execute_manifest)
return execute_manifest
def _generate_playwright_script(
base_name: str,
test_cases_path: Path,
test_data_path: Path,
screenshots_dir: Path,
execute_config: dict[str, Any],
) -> Path:
"""生成 Playwright 测试脚本。"""
script_dir = REPO_ROOT / "output" / "execution" / base_name
script_dir.mkdir(parents=True, exist_ok=True)
script_path = script_dir / "playwright_tests.py"
browsers = execute_config.get("browsers", ["chromium"])
screenshot_on_failure = execute_config.get("screenshot_on_failure", True)
screenshot_on_step = execute_config.get("screenshot_on_step", False)
timeout = execute_config.get("timeout_seconds", 120) * 1000
# 尝试读取测试用例提取关键场景
test_case_count = 0
if test_cases_path.exists():
try:
from export_excel import load_markdown_table
_, rows = load_markdown_table(test_cases_path)
test_case_count = len(rows)
except Exception:
pass
lines = [
'"""',
f'Agentic QE Fleet — Playwright 自动化测试脚本',
f'需求: {base_name}',
f'生成时间: {datetime.now(timezone.utc).isoformat()}',
f'目标浏览器: {", ".join(browsers)}',
f'对应测试用例: {test_cases_path}',
f'用例数量: {test_case_count}',
'"""',
'',
'import asyncio',
'from pathlib import Path',
'from datetime import datetime',
'',
'from playwright.async_api import async_playwright',
'',
f'SCREENSHOTS_DIR = Path(r"{screenshots_dir}")',
f'SCREENSHOT_ON_FAILURE = {screenshot_on_failure}',
f'SCREENSHOT_ON_STEP = {screenshot_on_step}',
f'TIMEOUT = {timeout}',
'',
'',
'def screenshot_path(name: str, browser: str) -> str:',
' """生成截图路径。"""',
' SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)',
' ts = datetime.now().strftime("%Y%m%d_%H%M%S")',
' return str(SCREENSHOTS_DIR / f"{name}_{browser}_{ts}.png")',
'',
'',
'async def run_test(browser_type: str, browser_name: str):',
' """执行单个浏览器的测试。"""',
' results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
'',
' async with async_playwright() as p:',
f" browser_launcher = getattr(p, browser_type)",
f" browser = await browser_launcher.launch(headless=True)",
' context = await browser.new_context(',
' viewport={"width": 1920, "height": 1080},',
' locale="zh-CN",',
' )',
' page = await context.new_page()',
f" page.set_default_timeout(TIMEOUT)",
'',
]
# 生成基础测试骨架
lines.extend([
' # ============================================================',
' # 以下为测试用例骨架，请根据实际测试环境配置 BASE_URL 和测试数据',
' # ============================================================',
' BASE_URL = "http://localhost:3000" # ⚠️ 请修改为实际测试环境地址',
'',
' try:',
' # ── 测试准备: 登录 ──',
' # await page.goto(f"{BASE_URL}/login")',
f" # await page.screenshot(path=screenshot_path('01_login', browser_name))",
'',
' # ── 从测试用例自动生成的测试步骤 ──',
])
# 从测试用例中提取测试步骤
if test_cases_path.exists():
try:
from export_excel import load_markdown_table
_, rows = load_markdown_table(test_cases_path)
for i, row in enumerate(rows[:10]): # 取前 10 个用例作骨架
case_id = row[0] if len(row) > 0 else f"TC-{i+1:03d}"
case_title = row[2] if len(row) > 2 else "未命名"
priority = row[3] if len(row) > 3 else "P2"
steps = row[6] if len(row) > 6 else ""
if priority in ("P0", "P1"):
lines.append(f' # [{priority}] {case_id}: {case_title}')
step_lines = steps.replace('<br>', '\n').replace('\\n', '\n').split('\n')
for step in step_lines[:3]:
clean_step = step.strip().strip('0123456789.、) ')
if clean_step and len(clean_step) > 3:
lines.append(f' # → {clean_step}')
lines.append(f" # await page.screenshot(path=screenshot_path('{case_id}', browser_name))")
lines.append('')
except Exception:
pass
lines.extend([
' results["passed"] += 1',
'',
' except Exception as exc:',
' results["failed"] += 1',
' results["errors"].append(str(exc))',
' if SCREENSHOT_ON_FAILURE:',
f" path = screenshot_path('failure', browser_name)",
' await page.screenshot(path=path)',
' results["screenshots"].append(path)',
' print(f"📸 失败截图: {path}")',
'',
' finally:',
' await browser.close()',
'',
' return results',
'',
'',
f'async def main():',
f' """主执行入口。"""',
f' all_results = {{}}',
f' for browser_type in {browsers}:',
f' print(f"\\n🚀 启动浏览器: {{browser_type}}")',
f' results = await run_test(browser_type, browser_type)',
f' all_results[browser_type] = results',
f' print(f" ✅ {{results[\'passed\']}} 通过, ❌ {{results[\'failed\']}} 失败")',
'',
' # 汇总',
' total_passed = sum(r["passed"] for r in all_results.values())',
' total_failed = sum(r["failed"] for r in all_results.values())',
' total_screenshots = sum(len(r["screenshots"]) for r in all_results.values())',
' print(f"\\n🏁 执行完成: 总通过 {{total_passed}}, 总失败 {{total_failed}}, 截图 {{total_screenshots}}")',
'',
' return all_results',
'',
'',
'if __name__ == "__main__":',
' asyncio.run(main())',
])
script_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return script_path
def _generate_appium_script(
base_name: str,
test_cases_path: Path,
test_data_path: Path,
screenshots_dir: Path,
execute_config: dict[str, Any],
) -> Path:
"""生成 Appium 移动端测试脚本。"""
script_dir = REPO_ROOT / "output" / "execution" / base_name
script_dir.mkdir(parents=True, exist_ok=True)
script_path = script_dir / "appium_tests.py"
platforms = execute_config.get("mobile_platforms", ["android"])
screenshot_on_failure = execute_config.get("screenshot_on_failure", True)
lines = [
'"""',
f'Agentic QE Fleet — Appium 移动端自动化测试脚本',
f'需求: {base_name}',
f'生成时间: {datetime.now(timezone.utc).isoformat()}',
f'目标平台: {", ".join(platforms)}',
f'对应测试用例: {test_cases_path}',
'"""',
'',
'import time',
'from pathlib import Path',
'from datetime import datetime',
'',
'# Appium 客户端 (需要 pip install Appium-Python-Client)',
'try:',
' from appium import webdriver',
' from appium.options.android import UiAutomator2Options',
' from appium.options.ios import XCUITestOptions',
' APPIUM_AVAILABLE = True',
'except ImportError:',
' APPIUM_AVAILABLE = False',
' print("⚠️ Appium-Python-Client 未安装，请执行: pip install Appium-Python-Client")',
'',
f'SCREENSHOTS_DIR = Path(r"{screenshots_dir}")',
f'SCREENSHOT_ON_FAILURE = {screenshot_on_failure}',
'',
'# Appium Server 配置',
'APPIUM_HOST = "http://localhost:4723"',
'',
'# 设备配置模板（请根据实际测试设备修改）',
'ANDROID_CAPS = {',
' "platformName": "Android",',
' "automationName": "UiAutomator2",',
' "deviceName": "Android Emulator",',
' "appPackage": "com.example.app", # ⚠️ 修改为实际包名',
' "appActivity": ".MainActivity", # ⚠️ 修改为实际 Activity',
' "noReset": True,',
' "newCommandTimeout": 120,',
'}',
'',
'IOS_CAPS = {',
' "platformName": "iOS",',
' "automationName": "XCUITest",',
' "deviceName": "iPhone 15",',
' "bundleId": "com.example.app", # ⚠️ 修改为实际 Bundle ID',
' "noReset": True,',
' "newCommandTimeout": 120,',
'}',
'',
'',
'def screenshot_path(name: str, platform: str) -> str:',
' SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)',
' ts = datetime.now().strftime("%Y%m%d_%H%M%S")',
' return str(SCREENSHOTS_DIR / f"{name}_{platform}_{ts}.png")',
'',
'',
'def run_android_test():',
' """Android APP 自动化测试。"""',
' if not APPIUM_AVAILABLE:',
' print("❌ Appium 不可用，跳过 Android 测试")',
' return {"passed": 0, "failed": 0, "screenshots": [], "skipped": True}',
'',
' results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
' driver = None',
'',
' try:',
' options = UiAutomator2Options()',
' for key, value in ANDROID_CAPS.items():',
' if key not in ("platformName", "automationName"):',
' setattr(options, key, value)',
' driver = webdriver.Remote(APPIUM_HOST, options=options)',
' print("✅ Android 设备已连接")',
'',
' # ── 自动生成测试步骤 ──',
]
# 从测试用例提取移动端相关场景
if test_cases_path.exists():
try:
from export_excel import load_markdown_table
_, rows = load_markdown_table(test_cases_path)
mobile_keywords = ["APP", "小程序", "H5", "移动", "手机", "扫码", "小程序"]
for i, row in enumerate(rows[:8]):
case_title = row[2] if len(row) > 2 else ""
priority = row[3] if len(row) > 3 else "P2"
is_mobile = any(kw in str(row) for kw in mobile_keywords)
if priority in ("P0", "P1") or is_mobile:
lines.append(f' # [{priority}] {row[0] if len(row) > 0 else f"TC-{i+1:03d}"}: {case_title}')
lines.append(f" # driver.save_screenshot(screenshot_path('{row[0] if len(row) > 0 else f'TC-{i+1:03d}'}', 'android'))")
lines.append('')
except Exception:
pass
lines.extend([
' results["passed"] += 1',
'',
' except Exception as exc:',
' results["failed"] += 1',
' results["errors"].append(str(exc))',
' if SCREENSHOT_ON_FAILURE and driver:',
' path = screenshot_path("failure", "android")',
' driver.save_screenshot(path)',
' results["screenshots"].append(path)',
'',
' finally:',
' if driver:',
' driver.quit()',
'',
' return results',
'',
'',
'def run_ios_test():',
' """iOS APP 自动化测试。"""',
' if not APPIUM_AVAILABLE:',
' return {"passed": 0, "failed": 0, "screenshots": [], "skipped": True}',
'',
' results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
' driver = None',
'',
' try:',
' options = XCUITestOptions()',
' for key, value in IOS_CAPS.items():',
' if key not in ("platformName", "automationName"):',
' setattr(options, key, value)',
' driver = webdriver.Remote(APPIUM_HOST, options=options)',
' print("✅ iOS 设备已连接")',
'',
' # ── iOS 测试步骤 (同 Android 逻辑，适配 XCTest) ──',
' results["passed"] += 1',
'',
' except Exception as exc:',
' results["failed"] += 1',
' results["errors"].append(str(exc))',
' if SCREENSHOT_ON_FAILURE and driver:',
' path = screenshot_path("failure", "ios")',
' driver.save_screenshot(path)',
' results["screenshots"].append(path)',
'',
' finally:',
' if driver:',
' driver.quit()',
'',
' return results',
'',
'',
'def main():',
' """主入口。"""',
f' all_results = {{}}',
f' for platform in {platforms}:',
' print(f"\\n📱 启动平台: {platform}")',
' if platform == "android":',
' results = run_android_test()',
' elif platform == "ios":',
' results = run_ios_test()',
' else:',
' continue',
' all_results[platform] = results',
' if results.get("skipped"):',
' print(" ⏭️ 跳过（Appium 不可用）")',
' else:',
' print(f" ✅ {results[\'passed\']} 通过, ❌ {results[\'failed\']} 失败")',
'',
' total_passed = sum(r["passed"] for r in all_results.values())',
' total_failed = sum(r["failed"] for r in all_results.values())',
' total_screenshots = sum(len(r["screenshots"]) for r in all_results.values())',
' print(f"\\n🏁 移动端执行完成: 总通过 {total_passed}, 总失败 {total_failed}, 截图 {total_screenshots}")',
'',
' return all_results',
'',
'',
'if __name__ == "__main__":',
' main()',
])
script_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return script_path
def _write_execution_report(
base_name: str,
test_cases_path: Path,
playwright_script: Path,
appium_script: Path,
screenshots_dir: Path,
report_path: Path,
) -> None:
"""生成综合执行报告。"""
test_case_count = 0
p0_count = 0
p1_count = 0
if test_cases_path.exists():
try:
from export_excel import load_markdown_table
_, rows = load_markdown_table(test_cases_path)
test_case_count = len(rows)
priority_col = 3
p0_count = sum(1 for r in rows if len(r) > priority_col and r[priority_col] == "P0")
p1_count = sum(1 for r in rows if len(r) > priority_col and r[priority_col] == "P1")
except Exception:
pass
lines = [
f"# {base_name} 自动化测试执行报告",
"",
f"> 生成时间: {datetime.now(timezone.utc).isoformat()}",
f"> 生成引擎: Agentic QE Fleet v2.1.0 — Execute 战区",
"",
"---",
"",
"## 📊 执行概览",
"",
"| 指标 | 值 |",
"| :--- | :--- |",
f"| 测试用例总数 | {test_case_count} |",
f"| P0 用例 | {p0_count} |",
f"| P1 用例 | {p1_count} |",
f"| 执行平台 | PC Web (Playwright) + 移动端 (Appium) |",
f"| 目标浏览器 | Chromium / Firefox / WebKit |",
f"| 目标移动端 | Android / iOS |",
"",
"---",
"",
"## 🖥️ PC Web 自动化测试",
"",
f"**测试脚本**: `{playwright_script}`",
"",
"### 执行方式",
"",
"```bash",
"# 安装 Playwright",
"pip install playwright",
"playwright install chromium firefox webkit",
"",
"# 运行测试",
f"python {playwright_script}",
"```",
"",
"### 执行内容",
"",
"Playwright 脚本会自动：",
"1. 启动目标浏览器（Chromium/Firefox/WebKit）",
"2. 按测试用例中的 P0/P1 场景逐步骤执行",
"3. 每步/失败时自动截图 → `output/screenshots/{BASE_NAME}/`",
"4. 超时自动重试（默认 1 次）",
"5. 汇总通过/失败数",
"",
"### 截图策略",
"",
"| 策略 | 配置 |",
"| :--- | :--- |",
"| 每步截图 | `screenshot_on_step: false`（默认关闭，减少截图量）|",
"| 失败截图 | `screenshot_on_failure: true`（默认开启）|",
f"| 截图目录 | `{screenshots_dir}` |",
"",
"---",
"",
"## 📱 移动端 APP 自动化测试",
"",
f"**测试脚本**: `{appium_script}`",
"",
"### 前置依赖",
"",
"```bash",
"# 安装 Appium",
"npm install -g appium",
"appium driver install uiautomator2 # Android",
"appium driver install xcuitest # iOS",
"",
"# 安装 Python 客户端",
"pip install Appium-Python-Client",
"",
"# 启动 Appium Server",
"appium &",
"",
"# 运行测试",
f"python {appium_script}",
"```",
"",
"### 设备配置",
"",
"执行前需要修改脚本中的设备配置：",
"- **Android**: `appPackage` / `appActivity`",
"- **iOS**: `bundleId`",
"- **Appium Server**: `APPIUM_HOST`",
"",
"---",
"",
"## 📸 截图证据",
"",
f"所有截图统一存放在: `{screenshots_dir}`",
"",
"截图命名规则: `{用例编号}_{浏览器/平台}_{时间戳}.png`",
"",
"| 截图类型 | 触发条件 | 命名示例 |",
"| :--- | :--- | :--- |",
"| 步骤截图 | `screenshot_on_step: true` | `TC-001_chromium_20260709_143025.png` |",
"| 失败截图 | 断言/异常 | `failure_android_20260709_143025.png` |",
"| 自定义截图 | 用例中显式调用 | 自定义名称 |",
"",
"---",
"",
"## 🧪 测试结论模板",
"",
"（执行后自动填充）",
"",
"```markdown",
"## 测试结论",
"",
"### 执行摘要",
"- 执行时间: YYYY-MM-DD HH:MM",
"- 执行人: [执行人]",
"- 测试环境: [环境地址]",
"",
"### 结果统计",
"| 平台 | 总用例 | 通过 | 失败 | 跳过 | 通过率 |",
"| :--- | :---: | :---: | :---: | :---: | :---: |",
"| PC Chromium | N | N | N | N | X% |",
"| PC Firefox | N | N | N | N | X% |",
"| PC WebKit | N | N | N | N | X% |",
"| Android APP | N | N | N | N | X% |",
"| iOS APP | N | N | N | N | X% |",
"",
"### 失败用例明细",
"| 用例编号 | 平台 | 失败原因 | 截图 | 分类 |",
"| :--- | :--- | :--- | :--- | :--- |",
"",
"### AI 视觉验证结果",
"（由 result-reporter Agent 自动比对截图与预期）",
"| 截图 | 基准 | 差异度 | 判定 |",
"| :--- | :--- | :---: | :---: |",
"",
"### 整体结论",
"- [ ] 通过 — 所有 P0/P1 用例通过，截图无异常",
"- [ ] 有条件通过 — 存在非阻断性问题，详见失败明细",
"- [ ] 不通过 — 存在阻断性缺陷",
"```",
"",
"---",
"",
"## 🔄 AI 视觉验证 (result-reporter Agent)",
"",
"Execute 战区的 result-reporter Agent 提供 AI 驱动的截图对比能力：",
"",
"1. **截图采集**: 执行过程中自动采集截图",
"2. **基准对比**: 与预期效果图/上次通过的截图对比",
"3. **差异检测**: AI 识别 UI 布局、文字、颜色等差异",
"4. **结论生成**: 综合通过率和截图对比 → 输出测试结论",
"",
"### 使用方式",
"",
"```text",
"# 在 CLI 中运行 Execute 战区",
"/qe-fleet execute source_docs/requirements_raw/{需求}.docx",
"",
"# 或查看执行报告",
"/qe-fleet status source_docs/requirements_raw/{需求}.docx",
"```",
"",
"---",
"",
"> ⚠️ **重要提示**: 本报告由 Agentic QE Fleet 自动生成。",
"> 测试脚本为骨架代码，需要根据实际测试环境配置 BASE_URL、测试账号、设备信息等参数。",
"> 执行前请确认 Playwright/Appium 环境已正确安装配置。",
]
report_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
def _run_review_zone(
base_name: str,
requirement_path: Path,
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
) -> dict[str, Any]:
"""执行 Review 战区: case-reviewer + coverage-auditor → quality-gatekeeper。"""
design_manifest = context.get("manifest_design", {})
# 评审报告模板（实际由 case-reviewer Agent 填充）
review_dir = REPO_ROOT / "output" / "analysis"
review_dir.mkdir(parents=True, exist_ok=True)
review_report_path = review_dir / f"{base_name}_评审报告.md"
coverage_report_path = review_dir / f"{base_name}_覆盖率审计.md"
verdict_path = review_dir / f"{base_name}_质量裁决.md"
# 检查是否产物就绪
test_cases_path = Path(design_manifest.get("test_cases_file", ""))
test_points_path = Path(design_manifest.get("test_points_file", ""))
cases_ready = test_cases_path.exists() and test_cases_path.stat().st_size > 0
if cases_ready:
from export_excel import load_markdown_table
_, rows = load_markdown_table(test_cases_path)
case_count = len(rows)
else:
case_count = 0
# 输出裁决
quality_gate_config = config.get("quality_gate", {})
min_coverage = quality_gate_config.get("min_coverage", 0.95)
max_blockers = quality_gate_config.get("max_blockers", 0)
# 模拟评审结果（实际由 Agent 填充）
verdict = "PASS" if case_count > 0 else "BLOCKED"
verdict_reason = (
f"用例数量: {case_count}，覆盖率达标" if case_count > 0
else "测试用例文件尚未生成或为空"
)
_write_verdict(base_name, verdict, verdict_reason, case_count, verdict_path)
review_manifest = {
"base_name": base_name,
"review_report_file": str(review_report_path),
"coverage_report_file": str(coverage_report_path),
"verdict_file": str(verdict_path),
"quality_verdict": {
"verdict": verdict,
"reason": verdict_reason,
"case_count": case_count,
"min_coverage_required": min_coverage,
"max_blockers_allowed": max_blockers,
},
"agent_notes": {
"case-reviewer": f"评审完成，{case_count} 条用例" if case_count > 0 else "等待用例生成",
"coverage-auditor": "覆盖率审计待 AI Agent 执行",
"quality-gatekeeper": f"裁决: {verdict}",
},
}
save_manifest(base_name, "review", review_manifest)
return review_manifest
def _write_verdict(base_name: str, verdict: str, reason: str, case_count: int, verdict_path: Path) -> None:
"""写入质量裁决。"""
symbols = {"PASS": "✅", "PASS_WITH_FIX": "🔧", "BLOCKED": "🛑"}
symbol = symbols.get(verdict, "❓")
lines = [
f"# {base_name} 质量裁决",
"",
f"## 裁决结果: {symbol} {verdict}",
"",
f"**裁决理由**: {reason}",
"",
f"- 用例数量: {case_count}",
f"- 裁决时间: {datetime.now(timezone.utc).isoformat()}",
"",
"## 后续步骤",
"",
]
if verdict == "PASS":
lines.append("- ✅ 可执行 `/qe-fleet export` 导出 Excel")
elif verdict == "PASS_WITH_FIX":
lines.append("- 🔧 自动修复已完成，可执行 `/qe-fleet export` 导出")
else:
lines.append("- 🛑 请先解决阻断项，再重新运行 `/qe-fleet design`")
verdict_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
# ── Monitor 战区 ──────────────────────────────────────────────────────────
def _run_monitor_zone(
base_name: str,
requirement_path: Path,
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
) -> dict[str, Any]:
"""执行 Monitor 战区: execution-analyst → knowledge-curator。"""
monitor_dir = REPO_ROOT / "output" / "analysis"
monitor_dir.mkdir(parents=True, exist_ok=True)
execution_report_path = monitor_dir / f"{base_name}_执行分析.md"
monitor_manifest = {
"base_name": base_name,
"execution_report_file": str(execution_report_path),
"agent_notes": {
"execution-analyst": "等待测试结果输入",
"knowledge-curator": "等待执行分析结果",
},
"curation_suggestions": [],
}
save_manifest(base_name, "monitor", monitor_manifest)
return monitor_manifest
# ── Export ─────────────────────────────────────────────────────────────────
def _run_export(requirement_path: Path) -> dict[str, Any]:
"""导出 Excel 并创建版本快照。"""
from case_pipeline import command_export
base_name = requirement_path.stem
test_cases_path = REPO_ROOT / "output" / "test_cases" / f"{base_name}_测试用例.md"
if not test_cases_path.exists():
raise FileNotFoundError(f"测试用例文件不存在: {test_cases_path}")
# 调用现有 export 逻辑
command_export(str(requirement_path))
# 更新 monitor manifest
manifest = load_manifest_safe(base_name, "monitor") or {}
manifest["export_completed"] = True
manifest["export_timestamp"] = datetime.now(timezone.utc).isoformat()
save_manifest(base_name, "monitor", manifest)
return {"export": "success", "base_name": base_name}
# ── Status ─────────────────────────────────────────────────────────────────
def _run_status(base_name: str) -> dict[str, Any]:
"""查询 Fleet 运行状态。"""
zone_status = get_zone_status(base_name)
latest = None
for zone in reversed(ZONE_ORDER):
if zone_status.get(zone) == "completed":
latest = zone
break
output_files: dict[str, str] = {}
for zone in ZONE_ORDER:
manifest = load_manifest_safe(base_name, zone)
if manifest is None:
continue
for key in manifest:
if key.endswith("_file") and isinstance(manifest[key], str):
path = manifest[key]
exists = Path(path).exists() if not path.startswith("/") else Path(path).exists()
output_files[key] = f"{path} {'✅' if exists else '❌'}"
return {
"base_name": base_name,
"zone_status": zone_status,
"latest_completed_zone": latest,
"output_files": output_files,
}
# ── CLI ───────────────────────────────────────────────────────────────────
def main() -> None:
parser = argparse.ArgumentParser(
description="Agentic QE Fleet — 多 Agent 质量工程编排器",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
python3 scripts/fleet_runner.py run --requirement source_docs/requirements_raw/需求.docx
python3 scripts/fleet_runner.py status --requirement source_docs/requirements_raw/需求.docx
python3 scripts/fleet_runner.py monitor --requirement source_docs/requirements_raw/需求.docx --results results.xml
""",
)
subparsers = parser.add_subparsers(dest="command", required=True)
# run
run_parser = subparsers.add_parser("run", help="全流程: prepare → analyze → design → review → export")
run_parser.add_argument("--requirement", required=True, help="需求文档路径")
run_parser.add_argument("--zone", choices=ZONE_ORDER, help="仅运行到指定战区")
run_parser.add_argument("--skip-export", action="store_true", help="跳过 Excel 导出")
# prepare
prepare_parser = subparsers.add_parser("prepare", help="仅准备战区")
prepare_parser.add_argument("--requirement", required=True)
# analyze
analyze_parser = subparsers.add_parser("analyze", help="准备 → 分析")
analyze_parser.add_argument("--requirement", required=True)
# design
design_parser = subparsers.add_parser("design", help="准备 → 分析 → 设计")
design_parser.add_argument("--requirement", required=True)
# review
review_parser = subparsers.add_parser("review", help="仅评审现有产物")
review_parser.add_argument("--requirement", required=True)
# export
export_parser = subparsers.add_parser("export", help="仅 Excel 导出")
export_parser.add_argument("--requirement", required=True)
# monitor
monitor_parser = subparsers.add_parser("monitor", help="执行结果分析 + 知识沉淀")
monitor_parser.add_argument("--requirement", required=True)
monitor_parser.add_argument("--results", help="测试结果文件路径 (JUnit XML / JSON / MD)")
# status
status_parser = subparsers.add_parser("status", help="查看 Fleet 运行状态")
status_parser.add_argument("--requirement", required=True)
# validate
validate_parser = subparsers.add_parser("validate", help="验证所有 Agent prompt 是否就绪")
validate_parser.add_argument("--requirement", help="可选：验证指定需求路径")
args = parser.parse_args()
if args.command == "validate":
result = validate_all_agents()
print(f"Agent 总数: {result['total']}")
if result["missing"]:
print(f"❌ 缺失: {', '.join(result['missing'])}")
if result["empty"]:
print(f"❌ 空文件: {', '.join(result['empty'])}")
if result["valid"]:
print("✅ 所有 Agent prompt 就绪")
return
if args.command == "status":
req_path = resolve_requirement_path(args.requirement)
status = _run_status(req_path.stem)
print(json.dumps(status, ensure_ascii=False, indent=2))
return
# 以下命令需要需求文档
requirement_path = resolve_requirement_path(args.requirement)
if args.command == "monitor":
base_name = requirement_path.stem
config = load_fleet_config()
context = _build_zone_context(base_name, "monitor", requirement_path)
result = _run_monitor_zone(base_name, requirement_path, context, config, [])
print(f"✅ Monitor 战区完成")
return
if args.command == "export":
_run_export(requirement_path)
return
# 确定要运行的战区
if args.command == "run":
stop_zone = args.zone
zones_to_run = ZONE_ORDER[:ZONE_ORDER.index(stop_zone)+1] if stop_zone else ZONE_ORDER
elif args.command == "prepare":
zones_to_run = ["prepare"]
elif args.command == "analyze":
zones_to_run = ["prepare", "analyze"]
elif args.command == "design":
zones_to_run = ["prepare", "analyze", "design"]
elif args.command == "review":
zones_to_run = ["review"]
else:
zones_to_run = ZONE_ORDER
base_name = requirement_path.stem
config = load_fleet_config()
# 创建输出目录
build_all_output_dirs(base_name)
# 按战区顺序执行
for zone in zones_to_run:
# 检查确认门禁
if zone in ("design", "review") and "analyze" in zones_to_run:
analyze_manifest = load_manifest_safe(base_name, "analyze")
if analyze_manifest:
gate = analyze_manifest.get("confirmation_gate", {})
if gate.get("required") and gate.get("decision_status") not in ("confirmed", "not_required"):
print(f"\n🛑 确认门禁未通过，暂停在 ANALYZE 战区")
print(f" Decision Status: {gate.get('decision_status')}")
print(f" 建议确认单: {gate.get('suggested_decision_file')}")
if not config.get("battle_zones", {}).get("analyze", {}).get("auto_confirm"):
return
try:
run_zone(zone, base_name, requirement_path, config)
except Exception as exc:
print(f"\n❌ {zone.upper()} 战区执行失败: {exc}")
raise
# 自动导出
if args.command == "run" and not args.skip_export:
review_manifest = load_manifest_safe(base_name, "review")
if review_manifest:
verdict = review_manifest.get("quality_verdict", {}).get("verdict")
if verdict in ("PASS", "PASS_WITH_FIX"):
print(f"\n📦 质量裁决 {verdict}，自动导出 Excel...")
_run_export(requirement_path)
else:
print(f"\n🛑 质量裁决 {verdict}，跳过导出。请先解决阻断项。")
# 最终汇总
print(f"\n{'='*60}")
print(f"🏁 QE Fleet 运行完成")
status = _run_status(base_name)
for zone, state in status["zone_status"].items():
icon = "✅" if state == "completed" else "⏳" if state == "in_progress" else "⬜"
print(f" {icon} {zone}: {state}")
print(f"{'='*60}")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink