94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
# Agent-N 全量回归可续跑驱动器：逐文件跑 pytest，结果追加 JSONL。
|
||
# 用法：.venv/Scripts/python.exe poc/pi-fallback/regression_driver.py <清单文件> <结果.jsonl> [预算秒] [单文件超时秒]
|
||
# 已记录在结果文件中的条目自动跳过；单文件超时记录为 TIMEOUT 后交由人工拆分。
|
||
import json
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
REPO = Path(__file__).resolve().parents[2]
|
||
PY = REPO / ".venv" / "Scripts" / "python.exe"
|
||
SUMMARY_RE = re.compile(
|
||
r"(?:(\d+) passed)?[,\s]*(?:(\d+) failed)?[,\s]*(?:(\d+) error[s]?)?[,\s]*"
|
||
r"(?:(\d+) skipped)?[,\s]*(?:(\d+) deselected)?")
|
||
|
||
|
||
def parse_summary(out: str) -> dict:
|
||
res = {"passed": 0, "failed": 0, "errors": 0, "skipped": 0}
|
||
for line in reversed(out.splitlines()):
|
||
line = line.strip()
|
||
if "passed" in line or "failed" in line or "error" in line:
|
||
m = SUMMARY_RE.search(line)
|
||
if m and ("in " in line and "s" in line):
|
||
res["passed"] = int(m.group(1) or 0)
|
||
res["failed"] = int(m.group(2) or 0)
|
||
res["errors"] = int(m.group(3) or 0)
|
||
res["skipped"] = int(m.group(4) or 0)
|
||
return res
|
||
return res
|
||
|
||
|
||
def failed_names(out: str) -> list[str]:
|
||
names = []
|
||
for line in out.splitlines():
|
||
s = line.strip()
|
||
if s.startswith(("FAILED ", "ERROR ")):
|
||
names.append(s[:300])
|
||
return names
|
||
|
||
|
||
def main() -> int:
|
||
list_file, results_file = Path(sys.argv[1]), Path(sys.argv[2])
|
||
budget = float(sys.argv[3]) if len(sys.argv) > 3 else 250.0
|
||
per_file_timeout = float(sys.argv[4]) if len(sys.argv) > 4 else 200.0
|
||
files = [x.strip() for x in list_file.read_text(encoding="utf-8").splitlines()
|
||
if x.strip()]
|
||
done = set()
|
||
if results_file.is_file():
|
||
for line in results_file.read_text(encoding="utf-8").splitlines():
|
||
if line.strip():
|
||
done.add(json.loads(line)["file"])
|
||
t0 = time.monotonic()
|
||
ran = 0
|
||
with open(results_file, "a", encoding="utf-8") as out:
|
||
for f in files:
|
||
if f in done:
|
||
continue
|
||
if time.monotonic() - t0 > budget:
|
||
print(f"[budget] 预算耗尽，停止（本轮完成 {ran} 个文件）")
|
||
break
|
||
ft0 = time.monotonic()
|
||
rec = {"file": f}
|
||
try:
|
||
proc = subprocess.run(
|
||
[str(PY), "-m", "pytest", f, "-q", "--tb=no", "-rfE",
|
||
"-p", "no:cacheprovider"],
|
||
cwd=str(REPO), capture_output=True, text=True,
|
||
timeout=per_file_timeout, check=False)
|
||
tail = proc.stdout + proc.stderr
|
||
s = parse_summary(tail)
|
||
rec.update(s)
|
||
rec["rc"] = proc.returncode
|
||
fails = failed_names(tail)
|
||
if fails:
|
||
rec["failedNames"] = fails
|
||
except subprocess.TimeoutExpired:
|
||
rec.update({"timeout": True, "passed": 0, "failed": 0,
|
||
"errors": 0, "skipped": 0})
|
||
rec["sec"] = round(time.monotonic() - ft0, 1)
|
||
out.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
||
out.flush()
|
||
ran += 1
|
||
status = ("TIMEOUT" if rec.get("timeout")
|
||
else f"passed={rec['passed']} failed={rec['failed']} "
|
||
f"errors={rec['errors']}")
|
||
print(f"[{status}] {f} ({rec['sec']}s)")
|
||
print(f"[done] 本轮 {ran} 个文件，累计 {len(done) + ran}/{len(files)}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|