Martino Ferrari
288 lines
12 KiB
Python
288 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
collect.py — Run the unit-test suites and gather coverage for the E2E report.
|
|
|
|
Produces two JSON artifacts in --out:
|
|
|
|
* ``unit_tests.json`` — per-suite {total, passed, failed, skipped, time_s, ok}
|
|
for the C++ GTest binary, the Go chain-client tests, and the Python
|
|
framework tests, plus grand totals.
|
|
* ``coverage.json`` — per-language {pct, avail, note} for Python (coverage.py),
|
|
Go (``go test -cover``) and C++ (lcov, best-effort: only when the build was
|
|
instrumented with .gcno files; otherwise reported unavailable).
|
|
|
|
Each suite is isolated: a missing toolchain or a failing suite is recorded, never
|
|
fatal, so the report always renders. Requires the MARTe env (LD_LIBRARY_PATH) to
|
|
already be exported for the GTest binary — the orchestrator does this.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import xml.etree.ElementTree as ET
|
|
|
|
|
|
def _run(cmd, cwd=None, env=None, timeout=600):
|
|
try:
|
|
p = subprocess.run(cmd, cwd=cwd, env=env, timeout=timeout,
|
|
capture_output=True, text=True)
|
|
return p.returncode, p.stdout, p.stderr
|
|
except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e:
|
|
return 127, "", str(e)
|
|
|
|
|
|
# ── C++ GTest ─────────────────────────────────────────────────────────────────
|
|
|
|
def gtest_suite(gtest_bin, work):
|
|
s = {"name": "C++ GTest", "lang": "cpp", "total": 0, "passed": 0,
|
|
"failed": 0, "skipped": 0, "time_s": 0.0, "ok": False, "avail": False}
|
|
if not gtest_bin or not os.path.exists(gtest_bin):
|
|
s["detail"] = "GTest binary not found"
|
|
return s
|
|
xml_p = os.path.join(work, "gtest.xml")
|
|
rc, out, err = _run([gtest_bin, f"--gtest_output=xml:{xml_p}"], timeout=900)
|
|
s["avail"] = True
|
|
if os.path.exists(xml_p):
|
|
try:
|
|
root = ET.parse(xml_p).getroot()
|
|
s["total"] = int(root.get("tests", 0))
|
|
s["failed"] = int(root.get("failures", 0)) + int(root.get("errors", 0))
|
|
s["skipped"] = int(root.get("disabled", 0)) + int(root.get("skipped", 0))
|
|
s["time_s"] = float(root.get("time", 0.0))
|
|
s["passed"] = s["total"] - s["failed"] - s["skipped"]
|
|
s["ok"] = s["failed"] == 0 and s["total"] > 0
|
|
except (ET.ParseError, ValueError) as e:
|
|
s["detail"] = f"xml parse: {e}"
|
|
else:
|
|
s["detail"] = (err or out or "no xml produced")[-200:]
|
|
return s
|
|
|
|
|
|
# ── Go ────────────────────────────────────────────────────────────────────────
|
|
|
|
def go_suite(client_dir, work):
|
|
s = {"name": "Go (chain-client)", "lang": "go", "total": 0, "passed": 0,
|
|
"failed": 0, "skipped": 0, "time_s": 0.0, "ok": False, "avail": False}
|
|
cov_p = os.path.join(work, "go_cover.out")
|
|
rc, out, err = _run(["go", "test", "-json", f"-coverprofile={cov_p}", "./..."],
|
|
cwd=client_dir)
|
|
if rc == 127:
|
|
s["detail"] = "go toolchain not found"
|
|
return s
|
|
s["avail"] = True
|
|
cov_pct = None
|
|
for line in out.splitlines():
|
|
try:
|
|
ev = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
act = ev.get("Action")
|
|
if act == "pass" and ev.get("Test"):
|
|
s["passed"] += 1
|
|
s["total"] += 1
|
|
elif act == "fail" and ev.get("Test"):
|
|
s["failed"] += 1
|
|
s["total"] += 1
|
|
elif act == "skip" and ev.get("Test"):
|
|
s["skipped"] += 1
|
|
s["total"] += 1
|
|
elif act == "output":
|
|
m = re.search(r"coverage:\s+([\d.]+)%", ev.get("Output", ""))
|
|
if m:
|
|
cov_pct = float(m.group(1))
|
|
s["ok"] = s["failed"] == 0 and s["total"] > 0
|
|
s["cov_pct"] = cov_pct
|
|
return s
|
|
|
|
|
|
# ── Python ──────────────────────────────────────────────────────────────────
|
|
|
|
def py_suite(chain_dir, work):
|
|
s = {"name": "Python (framework)", "lang": "python", "total": 0, "passed": 0,
|
|
"failed": 0, "skipped": 0, "time_s": 0.0, "ok": False, "avail": True}
|
|
cov_json = os.path.join(work, "py_cover.json")
|
|
env = dict(os.environ)
|
|
env["COVERAGE_FILE"] = os.path.join(work, ".coverage")
|
|
have_cov = _run(["coverage", "--version"])[0] == 0
|
|
if have_cov:
|
|
cmd = ["coverage", "run", f"--source={chain_dir}", "-m", "unittest",
|
|
"tests_py", "-v"]
|
|
else:
|
|
cmd = [sys.executable, "-m", "unittest", "tests_py", "-v"]
|
|
rc, out, err = _run(cmd, cwd=chain_dir, env=env)
|
|
text = out + "\n" + err
|
|
m = re.search(r"Ran (\d+) tests? in ([\d.]+)s", text)
|
|
if m:
|
|
s["total"] = int(m.group(1))
|
|
s["time_s"] = float(m.group(2))
|
|
fm = re.search(r"failures=(\d+)", text)
|
|
em = re.search(r"errors=(\d+)", text)
|
|
sm = re.search(r"skipped=(\d+)", text)
|
|
s["failed"] = (int(fm.group(1)) if fm else 0) + (int(em.group(1)) if em else 0)
|
|
s["skipped"] = int(sm.group(1)) if sm else 0
|
|
s["passed"] = s["total"] - s["failed"] - s["skipped"]
|
|
s["ok"] = s["failed"] == 0 and s["total"] > 0
|
|
pct = None
|
|
if have_cov:
|
|
_run(["coverage", "json", "-o", cov_json], cwd=chain_dir, env=env)
|
|
if os.path.exists(cov_json):
|
|
try:
|
|
cj = json.load(open(cov_json))
|
|
pct = round(cj["totals"]["percent_covered"], 1)
|
|
except (KeyError, ValueError):
|
|
pass
|
|
s["cov_pct"] = pct
|
|
return s
|
|
|
|
|
|
# ── C++ coverage (best-effort) ────────────────────────────────────────────────
|
|
|
|
def _parse_lcov_info(path, repo):
|
|
"""Parse an LCOV tracefile into per-file line/function coverage.
|
|
|
|
Returns (files, totals) where ``files`` is a list of
|
|
{path (repo-relative), lines_found, lines_hit, pct, funcs_found,
|
|
funcs_hit} sorted worst-covered first, and ``totals`` aggregates the
|
|
same line counts across all files. Only the line counters (DA/LF/LH)
|
|
and function counters (FNF/FNH) are read; branch data is ignored.
|
|
"""
|
|
files = []
|
|
cur = None
|
|
repo_abs = os.path.abspath(repo) + os.sep
|
|
try:
|
|
fh = open(path)
|
|
except OSError:
|
|
return [], {"lines_found": 0, "lines_hit": 0, "pct": None}
|
|
with fh:
|
|
for line in fh:
|
|
line = line.rstrip("\n")
|
|
if line.startswith("SF:"):
|
|
src = line[3:]
|
|
rel = src[len(repo_abs):] if src.startswith(repo_abs) else src
|
|
cur = {"path": rel, "lines_found": 0, "lines_hit": 0,
|
|
"funcs_found": 0, "funcs_hit": 0}
|
|
elif cur is None:
|
|
continue
|
|
elif line.startswith("LF:"):
|
|
cur["lines_found"] = int(line[3:] or 0)
|
|
elif line.startswith("LH:"):
|
|
cur["lines_hit"] = int(line[3:] or 0)
|
|
elif line.startswith("FNF:"):
|
|
cur["funcs_found"] = int(line[4:] or 0)
|
|
elif line.startswith("FNH:"):
|
|
cur["funcs_hit"] = int(line[4:] or 0)
|
|
elif line == "end_of_record":
|
|
lf = cur["lines_found"]
|
|
cur["pct"] = round(100.0 * cur["lines_hit"] / lf, 1) if lf else None
|
|
files.append(cur)
|
|
cur = None
|
|
tot_f = sum(f["lines_found"] for f in files)
|
|
tot_h = sum(f["lines_hit"] for f in files)
|
|
totals = {"lines_found": tot_f, "lines_hit": tot_h,
|
|
"pct": round(100.0 * tot_h / tot_f, 1) if tot_f else None}
|
|
files.sort(key=lambda f: (f["pct"] if f["pct"] is not None else 101.0,
|
|
f["path"]))
|
|
return files, totals
|
|
|
|
|
|
def cpp_coverage(repo, target):
|
|
cov = {"name": "C++", "avail": False, "pct": None, "files": [],
|
|
"note": "not instrumented (rebuild with --cpp-coverage)"}
|
|
build = os.path.join(repo, "Build", target)
|
|
gcno = []
|
|
for root, _, files in os.walk(build):
|
|
for fn in files:
|
|
if fn.endswith(".gcno"):
|
|
gcno.append(os.path.join(root, fn))
|
|
if not gcno:
|
|
return cov
|
|
# lcov 2.x is strict by default; tolerate the benign mismatches that arise
|
|
# from mixing instrumented project objects with non-instrumented MARTe2/STL
|
|
# headers, and never let a single bad file abort the whole capture.
|
|
ign = ["--ignore-errors",
|
|
"mismatch,source,gcov,unused,empty,negative,unsupported,inconsistent"]
|
|
raw = os.path.join(build, "coverage_raw.info")
|
|
rc, out, err = _run(["lcov", "--capture", "--directory", build,
|
|
"--output-file", raw, "--quiet"] + ign, timeout=900)
|
|
if rc != 0 or not os.path.exists(raw):
|
|
cov["note"] = "lcov capture failed: " + (err or out or "")[-160:]
|
|
return cov
|
|
# Keep only this repo's own sources so the number reflects project code,
|
|
# not the MARTe2 framework headers dragged in by templates/inlines.
|
|
info = os.path.join(build, "coverage.info")
|
|
rc2, _, e2 = _run(["lcov", "--extract", raw,
|
|
os.path.join(repo, "Source", "*"),
|
|
os.path.join(repo, "Test", "*"),
|
|
"--output-file", info, "--quiet"] + ign, timeout=300)
|
|
summ_file = info if (rc2 == 0 and os.path.exists(info)) else raw
|
|
# Parse the tracefile directly for per-file detail; this also yields the
|
|
# aggregate so the headline number and the per-file table are consistent.
|
|
fdetail, totals = _parse_lcov_info(summ_file, repo)
|
|
if totals["pct"] is not None:
|
|
cov.update(avail=True, pct=totals["pct"], files=fdetail,
|
|
lines_found=totals["lines_found"],
|
|
lines_hit=totals["lines_hit"],
|
|
note="lcov (project sources)" if summ_file == info else "lcov")
|
|
return cov
|
|
# Fall back to lcov --summary if the tracefile had no parseable line data.
|
|
rc3, summ, _ = _run(["lcov", "--summary", summ_file] + ign)
|
|
m = re.search(r"lines[.\s]+:\s+([\d.]+)%", summ)
|
|
if m:
|
|
cov.update(avail=True, pct=float(m.group(1)),
|
|
note="lcov (project sources)" if summ_file == info else "lcov")
|
|
else:
|
|
cov["note"] = "lcov summary unparsed"
|
|
return cov
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description="Collect unit tests + coverage")
|
|
p.add_argument("--repo", required=True)
|
|
p.add_argument("--target", default="x86-linux")
|
|
p.add_argument("--out", required=True)
|
|
p.add_argument("--work", default=None,
|
|
help="scratch dir for *.xml/cover files (default: --out)")
|
|
p.add_argument("--cpp-coverage", action="store_true")
|
|
args = p.parse_args()
|
|
os.makedirs(args.out, exist_ok=True)
|
|
work = args.work or args.out
|
|
os.makedirs(work, exist_ok=True)
|
|
chain_dir = os.path.dirname(os.path.abspath(__file__))
|
|
client_dir = os.path.join(chain_dir, "client")
|
|
gtest_bin = os.path.join(args.repo, "Build", args.target, "GTest", "MainGTest.ex")
|
|
|
|
suites = [gtest_suite(gtest_bin, work),
|
|
go_suite(client_dir, work),
|
|
py_suite(chain_dir, work)]
|
|
totals = {k: sum(s.get(k, 0) for s in suites)
|
|
for k in ("total", "passed", "failed", "skipped")}
|
|
ut = {"suites": suites, "totals": totals,
|
|
"ok": all(s["ok"] for s in suites if s["avail"])}
|
|
with open(os.path.join(args.out, "unit_tests.json"), "w") as f:
|
|
json.dump(ut, f, indent=2)
|
|
|
|
langs = []
|
|
py = next(s for s in suites if s["lang"] == "python")
|
|
go = next(s for s in suites if s["lang"] == "go")
|
|
langs.append({"name": "Python", "avail": py.get("cov_pct") is not None,
|
|
"pct": py.get("cov_pct"), "note": "coverage.py"})
|
|
langs.append({"name": "Go", "avail": go.get("cov_pct") is not None,
|
|
"pct": go.get("cov_pct"), "note": "go test -cover"})
|
|
cpp = cpp_coverage(args.repo, args.target) if args.cpp_coverage else \
|
|
{"name": "C++", "avail": False, "pct": None, "note": "skipped (use --cpp-coverage)"}
|
|
langs.append(cpp)
|
|
with open(os.path.join(args.out, "coverage.json"), "w") as f:
|
|
json.dump({"languages": langs}, f, indent=2)
|
|
|
|
print(f"unit_tests: {totals['passed']}/{totals['total']} pass "
|
|
f"({totals['failed']} fail, {totals['skipped']} skip)")
|
|
print("coverage: " + ", ".join(
|
|
f"{l['name']}={l['pct']}%" if l["pct"] is not None else f"{l['name']}=n/a"
|
|
for l in langs))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|