#!/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()