183 lines
7.2 KiB
Python
183 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M1b 测试执行器：真实跑 unittest 并把完整日志落盘（QC #9 取证）。
|
||
|
||
对应 QC 退回意见 #9：「仅 py_compile 通过，无真实执行通过证据」。
|
||
本工具用 unittest.TextTestRunner 真实执行 tests/test_m1b_*.py，
|
||
把 stdout/stderr 全量写入
|
||
projects/pbls/deliverables/m1b/test_logs/{suite}.log
|
||
并产出汇总 JSON
|
||
projects/pbls/deliverables/m1b/test_report.json
|
||
退出码 = 失败数 + 错误数（0 表示全绿）。
|
||
|
||
用法：
|
||
python3 tools/m1b_run_tests.py # 跑全部 M1b 测试
|
||
python3 tools/m1b_run_tests.py --suite ext_ref
|
||
python3 tools/m1b_run_tests.py -v 3
|
||
"""
|
||
|
||
import argparse
|
||
import io
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
# --- M1b sys.path bootstrap: modules/ 下各包互为兄弟仓库，需逐个入 path ---
|
||
_M1B_MOD_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
||
_M1B_MODULES_DIR = os.path.abspath(os.path.join(_M1B_MOD_ROOT, ".."))
|
||
_M1B_CANDIDATES = [_M1B_MOD_ROOT, _M1B_MODULES_DIR]
|
||
try:
|
||
for _d in sorted(os.listdir(_M1B_MODULES_DIR)):
|
||
_sub = os.path.join(_M1B_MODULES_DIR, _d)
|
||
if os.path.isdir(_sub) and not _d.startswith("."):
|
||
_M1B_CANDIDATES.append(_sub)
|
||
except OSError:
|
||
pass
|
||
for _p in _M1B_CANDIDATES:
|
||
if _p not in sys.path:
|
||
sys.path.insert(0, _p)
|
||
# --- end bootstrap ---
|
||
|
||
import time
|
||
import traceback
|
||
import unittest
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
MOD_ROOT = os.path.dirname(HERE)
|
||
REPO_ROOT = os.path.abspath(os.path.join(MOD_ROOT, "..", ".."))
|
||
TESTS_DIR = os.path.join(MOD_ROOT, "tests")
|
||
LOG_DIR = os.path.join(REPO_ROOT, "projects", "pbls", "deliverables",
|
||
"m1b", "test_logs")
|
||
REPORT = os.path.join(REPO_ROOT, "projects", "pbls", "deliverables",
|
||
"m1b", "test_report.json")
|
||
|
||
SUITES = {
|
||
"ext_ref": "test_m1b_ext_ref.py",
|
||
"realdb": "test_m1b_realdb.py",
|
||
"closure": "test_m1b_import_closure.py",
|
||
}
|
||
|
||
for p in (REPO_ROOT, MOD_ROOT, TESTS_DIR):
|
||
if p not in sys.path:
|
||
sys.path.insert(0, p)
|
||
|
||
|
||
def run_suite(name, filename, verbosity=2):
|
||
"""执行单个测试文件，返回 (result_dict, log_text)。"""
|
||
buf = io.StringIO()
|
||
old_out, old_err = sys.stdout, sys.stderr
|
||
started = time.time()
|
||
tests_run = failures = errors = skipped = 0
|
||
case_results = []
|
||
fatal = None
|
||
try:
|
||
sys.stdout = buf
|
||
sys.stderr = buf
|
||
loader = unittest.TestLoader()
|
||
suite = loader.discover(TESTS_DIR, pattern=filename, top_level_dir=TESTS_DIR)
|
||
runner = unittest.TextTestRunner(stream=buf, verbosity=verbosity)
|
||
res = runner.run(suite)
|
||
tests_run = res.testsRun
|
||
failures = len(res.failures)
|
||
errors = len(res.errors)
|
||
skipped = len(res.skipped)
|
||
for case, tb in res.failures:
|
||
case_results.append({"case": str(case), "status": "FAIL", "trace": tb})
|
||
for case, tb in res.errors:
|
||
case_results.append({"case": str(case), "status": "ERROR", "trace": tb})
|
||
for case, reason in res.skipped:
|
||
case_results.append({"case": str(case), "status": "SKIP",
|
||
"trace": str(reason)})
|
||
ok = res.wasSuccessful()
|
||
except Exception: # noqa: BLE001 - 装载失败也要留证据
|
||
ok = False
|
||
fatal = traceback.format_exc()
|
||
buf.write("\n[FATAL] suite load/run error:\n" + fatal)
|
||
finally:
|
||
sys.stdout, sys.stderr = old_out, old_err
|
||
elapsed = round(time.time() - started, 3)
|
||
log = buf.getvalue()
|
||
return {
|
||
"suite": name,
|
||
"file": "tests/%s" % filename,
|
||
"ok": bool(ok),
|
||
"tests_run": tests_run,
|
||
"failures": failures,
|
||
"errors": errors,
|
||
"skipped": skipped,
|
||
"passed": max(tests_run - failures - errors - skipped, 0),
|
||
"elapsed_sec": elapsed,
|
||
"fatal": fatal,
|
||
"cases": case_results,
|
||
}, log
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(description="M1b test runner (real execution)")
|
||
ap.add_argument("--suite", choices=list(SUITES.keys()),
|
||
help="只跑指定套件（缺省跑全部）")
|
||
ap.add_argument("-v", "--verbosity", type=int, default=2)
|
||
args = ap.parse_args(argv)
|
||
|
||
os.makedirs(LOG_DIR, exist_ok=True)
|
||
targets = ([args.suite] if args.suite else list(SUITES.keys()))
|
||
results = []
|
||
for name in targets:
|
||
fn = SUITES[name]
|
||
if not os.path.exists(os.path.join(TESTS_DIR, fn)):
|
||
results.append({"suite": name, "file": "tests/%s" % fn, "ok": False,
|
||
"tests_run": 0, "failures": 0, "errors": 1,
|
||
"skipped": 0, "passed": 0, "elapsed_sec": 0,
|
||
"fatal": "test file missing", "cases": []})
|
||
continue
|
||
res, log = run_suite(name, fn, verbosity=args.verbosity)
|
||
log_path = os.path.join(LOG_DIR, "%s.log" % name)
|
||
with open(log_path, "w", encoding="utf-8") as fh:
|
||
fh.write("=== M1b test suite: %s (%s) ===\n" % (name, fn))
|
||
fh.write("=== executed_at: %s ===\n" % time.strftime("%Y-%m-%d %H:%M:%S"))
|
||
fh.write("=== python: %s ===\n" % sys.version.replace("\n", " "))
|
||
fh.write("=== result: %s | run=%d pass=%d fail=%d error=%d skip=%d "
|
||
"| %.3fs ===\n\n" % (
|
||
"PASS" if res["ok"] else "FAIL", res["tests_run"],
|
||
res["passed"], res["failures"], res["errors"],
|
||
res["skipped"], res["elapsed_sec"]))
|
||
fh.write(log)
|
||
fh.write("\n=== END OF LOG ===\n")
|
||
res["log"] = os.path.relpath(log_path, REPO_ROOT)
|
||
res["log_bytes"] = os.path.getsize(log_path)
|
||
results.append(res)
|
||
print("[%s] %s run=%d pass=%d fail=%d error=%d skip=%d (%.3fs) -> %s" % (
|
||
"PASS" if res["ok"] else "FAIL", name, res["tests_run"], res["passed"],
|
||
res["failures"], res["errors"], res["skipped"], res["elapsed_sec"],
|
||
res["log"]))
|
||
|
||
total_run = sum(r["tests_run"] for r in results)
|
||
total_fail = sum(r["failures"] for r in results)
|
||
total_err = sum(r["errors"] for r in results)
|
||
total_skip = sum(r["skipped"] for r in results)
|
||
summary = {
|
||
"tool": "tools/m1b_run_tests.py",
|
||
"milestone": "M1b",
|
||
"executed_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"python": sys.version.split()[0],
|
||
"real_execution": True,
|
||
"suites": results,
|
||
"totals": {"run": total_run,
|
||
"passed": total_run - total_fail - total_err - total_skip,
|
||
"failures": total_fail, "errors": total_err,
|
||
"skipped": total_skip},
|
||
"all_green": total_fail == 0 and total_err == 0 and total_run > 0,
|
||
"log_dir": os.path.relpath(LOG_DIR, REPO_ROOT),
|
||
}
|
||
os.makedirs(os.path.dirname(REPORT), exist_ok=True)
|
||
with open(REPORT, "w", encoding="utf-8") as fh:
|
||
fh.write(json.dumps(summary, ensure_ascii=False, indent=2, default=str) + "\n")
|
||
print(json.dumps({"all_green": summary["all_green"], "totals": summary["totals"],
|
||
"report": os.path.relpath(REPORT, REPO_ROOT)},
|
||
ensure_ascii=False, indent=2))
|
||
return 0 if summary["all_green"] else (total_fail + total_err)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|