strepsiades
253 lines
8.2 KiB
Python
253 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from auto_test_lib import load_yaml_config, now_slug
|
|
|
|
|
|
ALL_DOMAINS = [
|
|
"company",
|
|
"supplier",
|
|
"project",
|
|
"contract",
|
|
"contract_change",
|
|
"construction_task",
|
|
"construction_task_detail",
|
|
"material",
|
|
"material_detail",
|
|
"contract_settlement",
|
|
"contract_settlement_detail",
|
|
"personnel",
|
|
"personnel_detail",
|
|
"preparation",
|
|
"production",
|
|
"lar",
|
|
"lar_change",
|
|
"project_detail_data",
|
|
"units",
|
|
]
|
|
|
|
|
|
def _list_report_dirs(report_base: Path) -> set[Path]:
|
|
if not report_base.exists():
|
|
return set()
|
|
return {path for path in report_base.iterdir() if path.is_dir()}
|
|
|
|
|
|
def _extract_report_dir(stdout: str, stderr: str) -> Path | None:
|
|
combined = "\n".join(part for part in [stdout, stderr] if part)
|
|
match = re.search(r"报告已生成:\s*(.+?/report\.md)", combined)
|
|
if not match:
|
|
return None
|
|
return Path(match.group(1).strip()).parent
|
|
|
|
|
|
def _build_test_summary(domain_reports: list[dict]) -> dict:
|
|
cases: list[dict] = []
|
|
passed_domains = 0
|
|
failed_domains = 0
|
|
|
|
for domain in domain_reports:
|
|
if domain.get("passed"):
|
|
passed_domains += 1
|
|
else:
|
|
failed_domains += 1
|
|
for round_report in domain.get("rounds", []):
|
|
cases.append(
|
|
{
|
|
"nodeid": f"{domain['domain']}::{round_report['name']}",
|
|
"domain": domain["domain"],
|
|
"case": round_report["name"],
|
|
"purpose": round_report.get("purpose", ""),
|
|
"passed": bool(round_report.get("passed")),
|
|
"result": "PASSED" if round_report.get("passed") else "FAILED",
|
|
"returncode": round_report.get("returncode"),
|
|
"issue_count": len(round_report.get("issues", [])),
|
|
"issues": round_report.get("issues", []),
|
|
}
|
|
)
|
|
|
|
passed_cases = sum(1 for case in cases if case["passed"])
|
|
return {
|
|
"total_domains": len(domain_reports),
|
|
"passed_domains": passed_domains,
|
|
"failed_domains": failed_domains,
|
|
"total_cases": len(cases),
|
|
"passed_cases": passed_cases,
|
|
"failed_cases": len(cases) - passed_cases,
|
|
"cases": cases,
|
|
}
|
|
|
|
|
|
def _build_markdown_report(run_id: str, summary: dict, domain_reports: list[dict], run_records: list[dict]) -> str:
|
|
lines = [
|
|
f"# 自动测试聚合报告 {run_id}",
|
|
"",
|
|
"## 运行方式",
|
|
"",
|
|
"- 本报告由 `scripts/run_auto_test_all_domains.py` 聚合生成",
|
|
"- 每个 domain 独立初始化并单独执行，避免跨 domain 状态污染",
|
|
"",
|
|
"## Pytest 风格汇总",
|
|
"",
|
|
f"collected {summary['total_cases']} cases across {summary['total_domains']} domains",
|
|
"",
|
|
]
|
|
for case in summary["cases"]:
|
|
lines.append(f"- {case['nodeid']} {case['result']}")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
f"== {summary['passed_cases']} passed, {summary['failed_cases']} failed, {summary['passed_domains']} passed domains, {summary['failed_domains']} failed domains ==",
|
|
"",
|
|
"## Domain 汇总",
|
|
"",
|
|
"| Domain | 结果 | 单域报告 |",
|
|
"| --- | --- | --- |",
|
|
]
|
|
)
|
|
run_record_by_domain = {item["domain"]: item for item in run_records}
|
|
for domain in domain_reports:
|
|
run_record = run_record_by_domain.get(domain["domain"], {})
|
|
lines.append(
|
|
f"| {domain['domain']} | {'通过' if domain.get('passed') else '失败'} | {run_record.get('report_dir', '')} |"
|
|
)
|
|
lines.append("")
|
|
|
|
for domain in domain_reports:
|
|
lines.append(f"## Domain: {domain['domain']}")
|
|
if domain.get("description"):
|
|
lines.extend(["", domain["description"]])
|
|
lines.append("")
|
|
lines.append(f"- 结果: {'通过' if domain.get('passed') else '失败'}")
|
|
record = run_record_by_domain.get(domain["domain"], {})
|
|
if record.get("report_dir"):
|
|
lines.append(f"- 单域报告目录: {record['report_dir']}")
|
|
if record.get("returncode") not in (None, 0):
|
|
lines.append(f"- 单域脚本返回码: {record['returncode']}")
|
|
if domain.get("issues"):
|
|
lines.append("- 问题:")
|
|
for issue in domain["issues"]:
|
|
lines.append(f" - {issue}")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="一轮执行所有已纳入 auto-test 的 domain")
|
|
parser.add_argument(
|
|
"--config",
|
|
default="tests/fixtures/auto_sync/auto_test_config.yaml",
|
|
help="自动测试配置文件路径",
|
|
)
|
|
parser.add_argument(
|
|
"--skip-init",
|
|
action="store_true",
|
|
help="透传给 run_auto_test.py",
|
|
)
|
|
parser.add_argument(
|
|
"--clear-only",
|
|
action="store_true",
|
|
help="透传给 run_auto_test.py",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
runner = Path(__file__).with_name("run_auto_test.py")
|
|
config = load_yaml_config(args.config)
|
|
report_base = Path(config["paths"]["report_dir"])
|
|
|
|
aggregate_run_id = now_slug()
|
|
aggregate_report_dir = report_base / aggregate_run_id
|
|
aggregate_report_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
domain_reports: list[dict] = []
|
|
run_records: list[dict] = []
|
|
overall_returncode = 0
|
|
|
|
for domain in ALL_DOMAINS:
|
|
command = [sys.executable, str(runner), "--config", args.config, "--domain", domain]
|
|
if args.skip_init:
|
|
command.append("--skip-init")
|
|
if args.clear_only:
|
|
command.append("--clear-only")
|
|
|
|
before_dirs = _list_report_dirs(report_base)
|
|
result = subprocess.run(command, check=False, text=True, capture_output=True)
|
|
after_dirs = _list_report_dirs(report_base)
|
|
new_dirs = sorted(after_dirs - before_dirs)
|
|
|
|
report_dir = _extract_report_dir(result.stdout, result.stderr)
|
|
if report_dir is None and new_dirs:
|
|
report_dir = new_dirs[-1]
|
|
|
|
run_record = {
|
|
"domain": domain,
|
|
"returncode": result.returncode,
|
|
"report_dir": str(report_dir) if report_dir else None,
|
|
"stdout_tail": (result.stdout or "")[-4000:],
|
|
"stderr_tail": (result.stderr or "")[-4000:],
|
|
}
|
|
run_records.append(run_record)
|
|
|
|
if result.returncode != 0:
|
|
overall_returncode = result.returncode
|
|
|
|
if report_dir is None:
|
|
domain_reports.append(
|
|
{
|
|
"domain": domain,
|
|
"description": "",
|
|
"passed": False,
|
|
"issues": ["未找到单域报告目录"],
|
|
"rounds": [],
|
|
}
|
|
)
|
|
continue
|
|
|
|
domain_report_path = report_dir / f"{domain}.json"
|
|
if not domain_report_path.exists():
|
|
domain_reports.append(
|
|
{
|
|
"domain": domain,
|
|
"description": "",
|
|
"passed": False,
|
|
"issues": [f"缺少单域报告文件: {domain_report_path}"],
|
|
"rounds": [],
|
|
}
|
|
)
|
|
overall_returncode = overall_returncode or 1
|
|
continue
|
|
|
|
domain_reports.append(json.loads(domain_report_path.read_text(encoding="utf-8")))
|
|
|
|
summary = _build_test_summary(domain_reports)
|
|
detail = {
|
|
"mode": "isolated-per-domain",
|
|
"aggregate_run_id": aggregate_run_id,
|
|
"summary": summary,
|
|
"runs": run_records,
|
|
"domains": domain_reports,
|
|
}
|
|
(aggregate_report_dir / "summary.json").write_text(
|
|
json.dumps(detail, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
(aggregate_report_dir / "report.md").write_text(
|
|
_build_markdown_report(aggregate_run_id, summary, domain_reports, run_records),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
print(f"聚合报告已生成: {aggregate_report_dir / 'report.md'}")
|
|
if summary["failed_domains"]:
|
|
return overall_returncode or 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main()) |