FinDataHub
cd076018ba
## 统一入口 run_all_tests.py 跑一次全量测试要记住四件事，任何一条忘了结论就是错的： 1. 每套件用哪个 venv（findatahub_core 只装在部分 venv 里） 2. 必须注入 PYTHONPATH，否则验的是已安装的旧版本 3. Scripts/bin/tests 需要 httpx，选错 venv 在收集阶段直接报错 4. **不能并行**——两个 pytest 会互踩系统临时目录，报 pytest-of-root EEXIST 假失败（表现为几十个 errors，看着像大回归） 摩擦本身就是回归风险的来源，固化成脚本。四个设计点： - 解释器按能力选：每套件声明 requires，扫描 Services/*/.venv 挑第一个 满足的；挑不到 exit 2 并列出缺什么，绝不用错的 venv 硬跑出假结果 - 强制注入 PYTHONPATH，源码目录排最前 - **每套件独立 --basetemp**（.pytest-cache/<suite>/）：从根上消除 tmpdir 争用，不依赖"记得串行"，也不去删系统临时目录 - 摘要提取要求「数字+状态词」相邻且 stdout 优先，避免把 (Background on this error at: sqlalche.me/...) 当成统计行 实测 6 套件 14.9s；加 --include-swift --include-frontend 共 8 套件 37s。 **重要发现**：scripts 从 205 → 252——多的 47 条是 test_rss_fetcher.py， 此前因 akshare venv 缺 httpx 一直被 --ignore 跳过。这批用例一直存在且可用， 只是从没在本机跑过。 ## 真实 MySQL 集成测试 findatahub-core/tests/integration/，8 条用例。 为什么 fake 不够：test_db_replace_guard.py 用 Fake 游标验了代码路径，但 Fake 验不了三件事——事务是否真的回滚（Fake 的 rollback 只是置标志位）、 跨 chunk 失败是否留下半成品（Fake 不会真提交，这个场景永远复现不出来）、 熔断时表是否真的没动（需真实读行数）。 隔离：每次 fixture 新建 fdh_it_replace_<随机8位> 空库，用完 DROP； 库名带前缀且 DROP 前再断言。不读现有配置、不碰现有库。 无 MYSQL_PASSWORD 时整组 skip，CI 无 MySQL 不会变红。 **证伪验证**：把 replace() 临时改回「每 chunk 提交」旧行为，该套件报 「回滚不完整：表内行数既不是旧值 1000，说明留下了半成品数据」→ 测试有牙。 恢复后复跑 8 passed，生产库行数前后一致（news_flash 368346 / stock_daily 8158172），无 fdh_it_% 残留。 --with-integration 开关默认关闭（要真实库且会建临时库）。
473 lines
20 KiB
Python
473 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
# coding=utf-8
|
||
"""统一测试入口 —— 一条命令跑全部套件.
|
||
|
||
## 为什么需要它
|
||
|
||
本机开发有 5 个独立 venv（各服务的 bundle 部署需要），跑一次全量测试要：
|
||
|
||
1. 记住每个套件该用哪个解释器（`findatahub-core` 只装在部分 venv 里）；
|
||
2. 记住 `findatahub_core` 以**非 editable** 方式装进各 venv，
|
||
不注入 `PYTHONPATH` 就会验到**已安装的旧版本**，源码改动不生效；
|
||
3. 记住 `Scripts/bin/tests` 里 `test_rss_fetcher.py` 需要 `httpx`，
|
||
选错 venv 会在收集阶段直接报错；
|
||
4. 记住 **不能并行跑两个 pytest 进程**——它们会互踩系统临时目录，
|
||
报 `FileExistsError: pytest-of-root`，表现为几十个 errors 的假回归。
|
||
|
||
这四条中任何一条被忘记，得到的结论就是错的。摩擦本身就是回归风险的来源，
|
||
所以把它固化成脚本。
|
||
|
||
## 关键设计
|
||
|
||
- **每个套件独立 `--basetemp`**：放进项目内 `.pytest-cache/<suite>/`，
|
||
从根上消除 tmpdir 争用，不依赖"记得串行"。也不去删系统临时目录。
|
||
- **解释器按能力选，不写死**：每个套件声明它需要能 import 哪些模块，
|
||
脚本在候选 venv 里挑第一个满足的；挑不到就明确报错，而不是用错的 venv 硬跑。
|
||
- **强制注入 `PYTHONPATH`**：把源码目录排在已安装包之前，避免验旧版本。
|
||
- **纯 Python、跨平台**：不依赖 bash，Windows 上同样可用
|
||
（与 `paths.py` / `process_utils.py` 的迁移方向一致）。
|
||
|
||
## 用法
|
||
|
||
python3 Scripts/bin/run_all_tests.py # 全部 Python 套件
|
||
python3 Scripts/bin/run_all_tests.py --list # 只列出套件与选中的解释器
|
||
python3 Scripts/bin/run_all_tests.py -k newsflash # 透传给 pytest 的 -k
|
||
python3 Scripts/bin/run_all_tests.py --only core,backend
|
||
python3 Scripts/bin/run_all_tests.py --include-swift --include-frontend
|
||
|
||
退出码：全部通过 → 0；任一失败 → 1；环境不满足（找不到解释器）→ 2。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
# 所有套件共用同一个 basetemp 根，但各自一个子目录（互不干扰）
|
||
BASETEMP_ROOT = ROOT / ".pytest-cache"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════
|
||
# 套件声明
|
||
# ══════════════════════════════════════════════════════════
|
||
|
||
@dataclass
|
||
class Suite:
|
||
"""一个可独立执行的 pytest 套件。"""
|
||
|
||
key: str
|
||
name: str
|
||
cwd: Path # 工作目录（相对 ROOT）
|
||
target: str # pytest 目标（相对 cwd）
|
||
# 需要排到 PYTHONPATH 最前面的源码目录（相对 ROOT）
|
||
source_paths: list[Path] = field(default_factory=list)
|
||
# 必须能 import 的模块；脚本据此挑选解释器
|
||
requires: list[str] = field(default_factory=lambda: ["pytest"])
|
||
note: str = ""
|
||
|
||
|
||
# 顺序即执行顺序。高频、纯离线的放前面，出错能更早暴露。
|
||
SUITES: list[Suite] = [
|
||
Suite(
|
||
key="core",
|
||
name="findatahub-core（数据层基类）",
|
||
cwd=ROOT / "Services/findatahub-core",
|
||
target="tests",
|
||
# 必须排在最前：该包以非 editable 装进各 venv，
|
||
# 否则测到的是已安装的旧版本（历史踩坑）
|
||
source_paths=[ROOT / "Services/findatahub-core/src"],
|
||
note="含 replace 收缩熔断回归",
|
||
),
|
||
Suite(
|
||
key="tushare",
|
||
name="tushare_fetcher（采集器）",
|
||
cwd=ROOT / "Services/tushare_fetcher",
|
||
target="tests/unit",
|
||
source_paths=[ROOT / "Services/findatahub-core/src"],
|
||
requires=["pytest", "findatahub_core"],
|
||
note="仅 unit：integration 需要真实 Tushare 额度",
|
||
),
|
||
Suite(
|
||
key="akshare",
|
||
name="akshare_fetcher（采集器）",
|
||
cwd=ROOT / "Services/akshare_fetcher",
|
||
target="tests",
|
||
source_paths=[ROOT / "Services/findatahub-core/src"],
|
||
requires=["pytest", "findatahub_core"],
|
||
),
|
||
Suite(
|
||
key="client",
|
||
name="findatahub-client（SDK）",
|
||
cwd=ROOT / "Services/findatahub-client",
|
||
target="tests",
|
||
# 测试 import 的是 `findatahub.client`（命名空间包），
|
||
# 必须把 client 自己的 src 排进 PYTHONPATH：
|
||
# 只写 findatahub_core 的话，akshare 的 venv 也能"满足依赖"却跑不起来
|
||
source_paths=[
|
||
ROOT / "Services/findatahub-core/src",
|
||
ROOT / "Services/findatahub-client/src",
|
||
],
|
||
requires=["pytest", "findatahub.client"],
|
||
),
|
||
Suite(
|
||
key="scripts",
|
||
name="Scripts/bin（脚本层）",
|
||
cwd=ROOT / "Scripts/bin",
|
||
target="tests",
|
||
# 该套件里 rss_fetcher 需要 httpx；缺了会在收集阶段直接报错
|
||
requires=["pytest", "httpx", "pymysql"],
|
||
note="含跨平台抽象层回归",
|
||
),
|
||
Suite(
|
||
key="backend",
|
||
name="kuaixun-service（后端）",
|
||
cwd=ROOT / "Services/kuaixun-service",
|
||
target="tests",
|
||
requires=["pytest", "fastapi", "httpx"],
|
||
note="含 CORS / 时间基准回归",
|
||
),
|
||
]
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════
|
||
# 解释器探测
|
||
# ══════════════════════════════════════════════════════════
|
||
|
||
def _venv_python(venv_dir: Path) -> Path | None:
|
||
"""返回 venv 内的解释器路径（兼容 POSIX 与 Windows 布局）。"""
|
||
for rel in ("bin/python", "bin/python3", "Scripts/python.exe"):
|
||
p = venv_dir / rel
|
||
if p.exists():
|
||
return p
|
||
return None
|
||
|
||
|
||
def candidate_interpreters() -> list[tuple[str, Path]]:
|
||
"""候选解释器，按优先级排列：各服务自带 venv 在前，当前解释器兜底。"""
|
||
cands: list[tuple[str, Path]] = []
|
||
for venv in sorted((ROOT / "Services").glob("*/.venv")):
|
||
py = _venv_python(venv)
|
||
if py:
|
||
cands.append((f"{venv.parent.name}/.venv", py))
|
||
cands.append(("当前解释器", Path(sys.executable)))
|
||
return cands
|
||
|
||
|
||
def _can_import(python: Path, module: str, pythonpath: str) -> bool:
|
||
"""探测某解释器能否 import 指定模块（含注入的 PYTHONPATH）。"""
|
||
env = dict(os.environ)
|
||
if pythonpath:
|
||
env["PYTHONPATH"] = pythonpath
|
||
try:
|
||
res = subprocess.run(
|
||
[str(python), "-c", f"import {module}"],
|
||
capture_output=True, timeout=30, env=env, cwd=str(ROOT),
|
||
)
|
||
return res.returncode == 0
|
||
except (OSError, subprocess.SubprocessError):
|
||
return False
|
||
|
||
|
||
def pick_interpreter(suite: Suite, cands: list[tuple[str, Path]]) -> tuple[str, Path] | None:
|
||
"""选第一个能满足 suite.requires 的解释器。"""
|
||
pythonpath = os.pathsep.join(str(p) for p in suite.source_paths)
|
||
for label, py in cands:
|
||
if all(_can_import(py, m, pythonpath) for m in suite.requires):
|
||
return label, py
|
||
return None
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════
|
||
# 执行
|
||
# ══════════════════════════════════════════════════════════
|
||
|
||
@dataclass
|
||
class Result:
|
||
suite: Suite
|
||
interpreter: str
|
||
returncode: int
|
||
seconds: float
|
||
summary: str
|
||
output: str
|
||
|
||
|
||
def _ensure_dir(path: Path) -> None:
|
||
"""建目录，容忍已存在。
|
||
|
||
不直接用 ``mkdir(exist_ok=True)``：本机沙箱对 mkdir 做了拦截，
|
||
即使目标已存在也可能抛 ``FileExistsError``（本项目已多次踩到）。
|
||
先判 is_dir 再兜住异常，两条路径都安全。
|
||
"""
|
||
if path.is_dir():
|
||
return
|
||
try:
|
||
path.mkdir(parents=True, exist_ok=True)
|
||
except FileExistsError:
|
||
pass
|
||
|
||
|
||
def run_suite(suite: Suite, label: str, python: Path, extra_args: list[str]) -> Result:
|
||
"""执行单个套件。"""
|
||
basetemp = BASETEMP_ROOT / suite.key
|
||
# pytest 会先清空 --basetemp 目标目录；它位于项目内，删的是自己的缓存
|
||
_ensure_dir(basetemp)
|
||
|
||
env = dict(os.environ)
|
||
if suite.source_paths:
|
||
env["PYTHONPATH"] = os.pathsep.join(str(p) for p in suite.source_paths)
|
||
# 统一 UTF-8：Windows 控制台默认 cp936，中文用例名会乱码甚至报编码错
|
||
env["PYTHONUTF8"] = "1"
|
||
env["PYTHONIOENCODING"] = "utf-8"
|
||
|
||
cmd = [
|
||
str(python), "-m", "pytest", suite.target,
|
||
"-q", "--basetemp", str(basetemp),
|
||
*extra_args,
|
||
]
|
||
|
||
started = time.time()
|
||
try:
|
||
proc = subprocess.run(
|
||
cmd, cwd=str(suite.cwd), env=env,
|
||
capture_output=True, text=True, timeout=1800,
|
||
)
|
||
elapsed = time.time() - started
|
||
stdout = proc.stdout or ""
|
||
stderr = proc.stderr or ""
|
||
output = stdout + stderr
|
||
return Result(suite, label, proc.returncode, elapsed,
|
||
_extract_summary(stdout, stderr), output)
|
||
except subprocess.TimeoutExpired:
|
||
return Result(suite, label, 124, time.time() - started,
|
||
"超时（1800s）", "")
|
||
except OSError as e:
|
||
return Result(suite, label, 125, time.time() - started,
|
||
f"启动失败：{e}", "")
|
||
|
||
|
||
# pytest 统计行的形态："132 passed, 2 skipped in 3.55s" / "1 error in 0.25s"
|
||
# 需覆盖的边界：
|
||
# - "-k 过滤到零用例" → "134 deselected in 0.32s"（exit 5，无 passed/failed）
|
||
# - "no tests ran"：同样 exit 5
|
||
# - xfailed / xpassed：backend 套件有 xfail
|
||
_SUMMARY_RE = re.compile(
|
||
r"\b\d+ (?:passed|failed|error|errors|skipped|deselected|xfailed|xpassed"
|
||
r"|warning|warnings|rerun|reruns)\b|no tests ran"
|
||
)
|
||
|
||
|
||
def _extract_summary(stdout: str, stderr: str = "") -> str:
|
||
"""从 pytest 输出里取统计行。
|
||
|
||
必须**优先看 stdout、且要求「数字 + 状态词」相邻**：
|
||
应用自身的 stderr 日志里常出现形如
|
||
``(Background on this error at: https://sqlalche.me/...)`` 的行，
|
||
只按 "error" 关键词匹配会把这种日志误当成统计结果
|
||
（首版脚本就把 backend 套件的摘要取成了 SQLAlchemy 的报错提示）。
|
||
"""
|
||
for text in (stdout, stderr):
|
||
for line in reversed(text.strip().splitlines()):
|
||
s = line.strip().strip("=· ")
|
||
if s and _SUMMARY_RE.search(s):
|
||
return s
|
||
return "（无统计输出）"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════
|
||
# Swift / 前端（可选）
|
||
# ══════════════════════════════════════════════════════════
|
||
|
||
def run_swift(extra_args: list[str]) -> Result:
|
||
suite = Suite(key="swift", name="Swift 单元测试", cwd=ROOT, target="Tests")
|
||
started = time.time()
|
||
if not shutil.which("swift"):
|
||
return Result(suite, "-", 127, 0, "未安装 swift", "")
|
||
# SwiftPM 自身要调 sandbox-exec，本机沙箱下必须 --disable-sandbox
|
||
cmd = ["swift", "test", "--disable-sandbox", *extra_args]
|
||
try:
|
||
proc = subprocess.run(cmd, cwd=str(ROOT), capture_output=True, text=True, timeout=1800)
|
||
out = (proc.stdout or "") + (proc.stderr or "")
|
||
executed = [l for l in out.splitlines() if "Executed" in l and "failure" in l]
|
||
summary = executed[-1].strip() if executed else ("通过" if proc.returncode == 0 else "失败")
|
||
return Result(suite, "swift", proc.returncode, time.time() - started, summary, out)
|
||
except (OSError, subprocess.SubprocessError) as e:
|
||
return Result(suite, "swift", 125, time.time() - started, f"启动失败：{e}", "")
|
||
|
||
|
||
# ANSI 转义：vitest / pytest 在管道输出里也可能带颜色码，
|
||
# 不去掉会破坏行首匹配（首版前端摘要因此退化成"通过"，看不到真实数量）
|
||
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
|
||
|
||
|
||
def run_frontend(extra_args: list[str]) -> Result:
|
||
cwd = ROOT / "Services/kuaixun-frontend"
|
||
suite = Suite(key="frontend", name="前端（vitest）", cwd=cwd, target="tests")
|
||
started = time.time()
|
||
npx = shutil.which("npx")
|
||
if not npx or not (cwd / "node_modules").exists():
|
||
return Result(suite, "-", 127, 0, "未安装 npx 或 node_modules", "")
|
||
cmd = [npx, "vitest", "run", "--reporter=dot", *extra_args]
|
||
try:
|
||
proc = subprocess.run(cmd, cwd=str(cwd), capture_output=True, text=True, timeout=1800)
|
||
out = _ANSI_RE.sub("", (proc.stdout or "") + (proc.stderr or ""))
|
||
# vitest 汇总行形如 " Tests 44 passed (44)"
|
||
summary = ""
|
||
for line in out.splitlines():
|
||
if re.match(r"^\s*Tests\s+\d+", line):
|
||
summary = line.strip()
|
||
if not summary:
|
||
summary = "通过" if proc.returncode == 0 else "失败"
|
||
return Result(suite, "npx", proc.returncode, time.time() - started, summary, out)
|
||
except (OSError, subprocess.SubprocessError) as e:
|
||
return Result(suite, "npx", 125, time.time() - started, f"启动失败：{e}", "")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════
|
||
# 入口
|
||
# ══════════════════════════════════════════════════════════
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(
|
||
description="统一测试入口：一条命令跑全部套件",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="示例：\n %(prog)s\n %(prog)s --list\n %(prog)s -k newsflash\n"
|
||
" %(prog)s --only core,backend --include-swift",
|
||
)
|
||
parser.add_argument("--list", action="store_true", help="只列出套件与选中的解释器")
|
||
parser.add_argument("--only", default="", help="只跑指定套件，逗号分隔（key）")
|
||
parser.add_argument("--skip", default="", help="跳过指定套件，逗号分隔（key）")
|
||
parser.add_argument("--include-swift", action="store_true", help="同时跑 Swift 测试")
|
||
parser.add_argument("--include-frontend", action="store_true", help="同时跑前端测试")
|
||
parser.add_argument(
|
||
"--with-integration", action="store_true",
|
||
help="同时跑需要真实 MySQL 的集成测试（会自建临时库，需 MYSQL_PASSWORD）",
|
||
)
|
||
parser.add_argument("--verbose", action="store_true",
|
||
help="失败时打印完整输出（刻意不设 -v：留给 pytest 的 -v）")
|
||
# 未知参数一律透传给 pytest。
|
||
# 不用 nargs=REMAINDER：argparse 在 "先有 --only、后有 -k" 的组合下
|
||
# 会把 -k 当成本脚本的未知选项而直接退出 2（已实测）。
|
||
args, extra = parser.parse_known_args(argv)
|
||
extra = [a for a in extra if a != "--"]
|
||
|
||
only = {s.strip() for s in args.only.split(",") if s.strip()}
|
||
skip = {s.strip() for s in args.skip.split(",") if s.strip()}
|
||
|
||
selected = [s for s in SUITES
|
||
if (not only or s.key in only) and s.key not in skip]
|
||
|
||
unknown = only - {s.key for s in SUITES}
|
||
if unknown:
|
||
print(f"未知套件：{', '.join(sorted(unknown))}；可用："
|
||
f"{', '.join(s.key for s in SUITES)}", file=sys.stderr)
|
||
return 2
|
||
|
||
cands = candidate_interpreters()
|
||
|
||
# ── 选解释器 ──
|
||
planned: list[tuple[Suite, str, Path]] = []
|
||
unresolved: list[Suite] = []
|
||
for suite in selected:
|
||
picked = pick_interpreter(suite, cands)
|
||
if picked is None:
|
||
unresolved.append(suite)
|
||
else:
|
||
planned.append((suite, picked[0], picked[1]))
|
||
|
||
if args.list:
|
||
print(f"项目根：{ROOT}\n")
|
||
print(f"{'套件':<10} {'解释器':<28} {'需要模块'}")
|
||
print("-" * 78)
|
||
for suite in selected:
|
||
picked = pick_interpreter(suite, cands)
|
||
label = picked[0] if picked else "❌ 无可用解释器"
|
||
print(f"{suite.key:<10} {label:<28} {', '.join(suite.requires)}")
|
||
if suite.note:
|
||
print(f"{'':<10} └─ {suite.note}")
|
||
return 0 if not unresolved else 2
|
||
|
||
if unresolved:
|
||
print("以下套件找不到满足依赖的解释器：", file=sys.stderr)
|
||
for suite in unresolved:
|
||
print(f" - {suite.key}（{suite.name}）：需要 "
|
||
f"{', '.join(suite.requires)}", file=sys.stderr)
|
||
print("\n可尝试：在对应服务目录执行 `uv sync` 补齐依赖。", file=sys.stderr)
|
||
return 2
|
||
|
||
# ── 串行执行 ──
|
||
_ensure_dir(BASETEMP_ROOT)
|
||
if args.with_integration and not os.environ.get("MYSQL_PASSWORD"):
|
||
print(
|
||
"⚠️ 已启用 --with-integration，但未设置 MYSQL_PASSWORD：\n"
|
||
" 集成测试会整体 skip（不算失败）。如需真跑：\n"
|
||
" set -a; . ~/.config/findatahub/env; set +a\n",
|
||
file=sys.stderr,
|
||
)
|
||
results: list[Result] = []
|
||
total = len(planned) + int(args.include_swift) + int(args.include_frontend)
|
||
idx = 0
|
||
|
||
print(f"项目根：{ROOT}")
|
||
print(f"套件数：{total}（每个套件独立 --basetemp，天然互不干扰）\n")
|
||
|
||
for suite, label, python in planned:
|
||
idx += 1
|
||
print(f"[{idx}/{total}] {suite.name} … ", end="", flush=True)
|
||
# 集成测试默认排除：它要真实 MySQL，且会建临时库。
|
||
# 只对 core 套件生效（目前唯一含 integration 子目录的套件）。
|
||
suite_extra = list(extra)
|
||
if suite.key == "core" and not args.with_integration:
|
||
suite_extra.append("--ignore=tests/integration")
|
||
r = run_suite(suite, label, python, suite_extra)
|
||
results.append(r)
|
||
status = "✓" if r.returncode == 0 else "✗"
|
||
print(f"{status} {r.summary} ({r.seconds:.1f}s, {r.interpreter})")
|
||
|
||
if args.include_swift:
|
||
idx += 1
|
||
print(f"[{idx}/{total}] Swift 单元测试 … ", end="", flush=True)
|
||
r = run_swift(extra)
|
||
results.append(r)
|
||
print(f"{'✓' if r.returncode == 0 else '✗'} {r.summary} ({r.seconds:.1f}s)")
|
||
|
||
if args.include_frontend:
|
||
idx += 1
|
||
print(f"[{idx}/{total}] 前端（vitest）… ", end="", flush=True)
|
||
r = run_frontend(extra)
|
||
results.append(r)
|
||
print(f"{'✓' if r.returncode == 0 else '✗'} {r.summary} ({r.seconds:.1f}s)")
|
||
|
||
# ── 汇总 ──
|
||
failed = [r for r in results if r.returncode != 0]
|
||
print("\n" + "=" * 78)
|
||
print(f"{'套件':<12} {'状态':<6} {'耗时':>8} 统计")
|
||
print("-" * 78)
|
||
for r in results:
|
||
status = "PASS" if r.returncode == 0 else "FAIL"
|
||
print(f"{r.suite.key:<12} {status:<6} {r.seconds:>7.1f}s {r.summary}")
|
||
print("=" * 78)
|
||
|
||
if failed:
|
||
print(f"\n❌ {len(failed)} 个套件失败：{', '.join(r.suite.key for r in failed)}")
|
||
for r in failed:
|
||
if args.verbose and r.output:
|
||
print(f"\n───── {r.suite.key} 输出 ─────")
|
||
print(r.output[-6000:])
|
||
return 1
|
||
|
||
total_seconds = sum(r.seconds for r in results)
|
||
print(f"\n✅ 全部 {len(results)} 个套件通过（合计 {total_seconds:.1f}s）")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|