test(coverage): a reproducible coverage preset, report and floor check

tools/coverage_floor.py

@@ -0,0 +1,282 @@
+#!/usr/bin/env python3
+"""Gate an lcov report for src/** against the project's coverage floors.
+
+The floors are the contract made visible in docs/coverage.md: every file under
+src/ must keep at least 80% of its instrumented lines executed by the unit
+suite, and src/ as a whole at least 90%. `tools/coverage.sh` runs this after it
+captures the report, so a regression fails the one command instead of quietly
+lowering a number in a document.
+
+The check is deliberately able to fail: `--per-file` and `--aggregate` exist so
+the failure path can be demonstrated (raise the floor above a file's real rate
+and watch it exit 1), and the same knobs are what a future task would tighten.
+
+Exit status: 0 when every floor holds, 1 when one does not, 2 when the report
+cannot be read at all (missing file, no src/ records, malformed record) -- a
+report this tool cannot find must never read as a pass.
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+# The floors docs/coverage.md states. Kept here rather than in the shell script
+# so the checker can be run on its own against any captured report.
+PER_FILE_FLOOR = 80.0
+AGGREGATE_FLOOR = 90.0
+
+
+@dataclass(frozen=True)
+class Record:
+    """One lcov SF..end_of_record block: a source file and what it executed."""
+
+    path: str
+    lines_found: int
+    lines_hit: int
+    functions_found: int
+    functions_hit: int
+
+    @property
+    def rate(self) -> float:
+        return 100.0 * self.lines_hit / self.lines_found if self.lines_found else 100.0
+
+    @property
+    def function_rate(self) -> float:
+        return (
+            100.0 * self.functions_hit / self.functions_found
+            if self.functions_found
+            else 100.0
+        )
+
+    @property
+    def display(self) -> str:
+        return relative_to_source_root(self.path)
+
+
+def relative_to_source_root(path: str) -> str:
+    """`/home/.../plotter/src/gl/loader.cpp` -> `src/gl/loader.cpp`."""
+    if path.startswith("src/"):
+        return path
+    marker = path.rfind("/src/")
+    return path[marker + 1 :] if marker >= 0 else path
+
+
+def parse_lcov(text: str) -> list[Record]:
+    """Every SF block in an lcov tracefile, in file order.
+
+    A record with no LF line is treated as having no instrumented lines, which
+    is what a header-only or string-literal-only translation unit reports -- it
+    is not a failure and not part of the aggregate either way.
+
+    Functions are counted from lcov 2.x's per-function records (`FNA:<index>,
+    <count>,<name>`) when they are there, because that is also what `lcov --list`
+    and `genhtml` count; `FNF`/`FNH` are the fallback for a tracefile written
+    without them. The two can disagree by a function or two -- a duplicated
+    function name collapses in FNF but not in the FNA list -- and the table here
+    exists to agree with the tools the reader will run next.
+    """
+    records: list[Record] = []
+    fields: dict[str, str] = {}
+    functions_found = 0
+    functions_hit = 0
+
+    def flush() -> None:
+        nonlocal functions_found, functions_hit
+        source = fields.get("SF")
+        found = functions_found or int(fields.get("FNF", 0))
+        hit = functions_hit or int(fields.get("FNH", 0))
+        lines_found = int(fields.get("LF", 0))
+        lines_hit = int(fields.get("LH", 0))
+        fields.clear()
+        functions_found = 0
+        functions_hit = 0
+        if source is None:
+            return
+        records.append(
+            Record(
+                path=source,
+                lines_found=lines_found,
+                lines_hit=lines_hit,
+                functions_found=found,
+                functions_hit=hit,
+            )
+        )
+
+    for raw_line in text.splitlines():
+        line = raw_line.strip()
+        if not line:
+            continue
+        if line == "end_of_record":
+            flush()
+            continue
+        key, separator, value = line.partition(":")
+        if not separator:
+            continue
+        if key == "FNA":
+            # FNA:<index>,<count>,<name> -- the name itself may hold a comma,
+            # so only the first two fields are split off.
+            _, count, _ = value.split(",", 2)
+            functions_found += 1
+            if int(count) > 0:
+                functions_hit += 1
+        elif key in {"SF", "LF", "LH", "FNF", "FNH"}:
+            fields[key] = value
+    flush()
+    return records
+
+
+def src_records(records: list[Record]) -> list[Record]:
+    """The records the floors apply to: the compiled bodies under src/."""
+    selected = [
+        record
+        for record in records
+        if record.path.startswith("src/") or "/src/" in record.path
+    ]
+    selected.sort(key=lambda record: record.display)
+    return selected
+
+
+def render_table(records: list[Record], markdown: bool) -> str:
+    aggregate_hit = sum(record.lines_hit for record in records)
+    aggregate_found = sum(record.lines_found for record in records)
+    aggregate_rate = 100.0 * aggregate_hit / aggregate_found if aggregate_found else 0.0
+    function_hit = sum(record.functions_hit for record in records)
+    function_found = sum(record.functions_found for record in records)
+
+    lines: list[str] = []
+    if markdown:
+        lines.append("| File | Lines | Line rate | Functions |")
+        lines.append("| --- | ---: | ---: | ---: |")
+    else:
+        lines.append(f"{'File':<40} {'Lines':>10} {'Rate':>8} {'Functions':>12}")
+
+    for record in records:
+        if markdown:
+            lines.append(
+                f"| `{record.display}` | {record.lines_hit}/{record.lines_found} | "
+                f"{record.rate:.1f}% | {record.function_rate:.1f}% |"
+            )
+        else:
+            lines.append(
+                f"{record.display:<40} "
+                f"{f'{record.lines_hit}/{record.lines_found}':>10} "
+                f"{f'{record.rate:.1f}%':>8} "
+                f"{f'{record.functions_hit}/{record.functions_found}':>12}"
+            )
+
+    if markdown:
+        function_rate = (
+            f"**{100.0 * function_hit / function_found:.1f}%**" if function_found else ""
+        )
+        lines.append(
+            f"| **src/ aggregate** | **{aggregate_hit}/{aggregate_found}** | "
+            f"**{aggregate_rate:.1f}%** | {function_rate} |"
+        )
+    else:
+        lines.append(
+            f"{'src/ aggregate':<40} {f'{aggregate_hit}/{aggregate_found}':>10} "
+            f"{f'{aggregate_rate:.1f}%':>8}"
+        )
+    return "\n".join(lines)
+
+
+def main(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(
+        description="Fail when src/** coverage drops below the project's floors.",
+    )
+    parser.add_argument("report", help="the lcov tracefile, e.g. build/cov/coverage.info")
+    parser.add_argument(
+        "--per-file",
+        type=float,
+        default=PER_FILE_FLOOR,
+        metavar="PCT",
+        help=f"minimum line rate for every file under src/ (default {PER_FILE_FLOOR:g})",
+    )
+    parser.add_argument(
+        "--aggregate",
+        type=float,
+        default=AGGREGATE_FLOOR,
+        metavar="PCT",
+        help=f"minimum line rate for src/ as a whole (default {AGGREGATE_FLOOR:g})",
+    )
+    parser.add_argument("--table", action="store_true", help="print every file, not just failures")
+    parser.add_argument(
+        "--format",
+        choices=("table", "markdown"),
+        default="table",
+        help="table layout; markdown is what docs/coverage.md carries",
+    )
+    args = parser.parse_args(argv)
+
+    report = Path(args.report)
+    if not report.is_file():
+        print(
+            f"coverage_floor: no report at {report} -- build and capture with "
+            "tools/coverage.sh first",
+            file=sys.stderr,
+        )
+        return 2
+
+    try:
+        records = src_records(parse_lcov(report.read_text(encoding="utf-8")))
+    except (OSError, ValueError) as error:
+        print(f"coverage_floor: cannot read {report}: {error}", file=sys.stderr)
+        return 2
+
+    if not records:
+        print(
+            f"coverage_floor: {report} holds no records under src/ -- the report is "
+            "empty or was filtered away; refusing to report a pass",
+            file=sys.stderr,
+        )
+        return 2
+
+    if args.table:
+        print(render_table(records, args.format == "markdown"))
+        print()
+
+    failures: list[str] = []
+    for record in records:
+        # LF:0 is a translation unit with nothing to execute (string literals
+        # only); it has no rate to hold to a floor.
+        if record.lines_found == 0:
+            continue
+        if record.lines_hit * 100.0 < args.per_file * record.lines_found:
+            failures.append(
+                f"  {record.display}: {record.rate:.1f}% lines "
+                f"({record.lines_hit}/{record.lines_found}) is below the "
+                f"{args.per_file:g}% per-file floor"
+            )
+
+    aggregate_hit = sum(record.lines_hit for record in records)
+    aggregate_found = sum(record.lines_found for record in records)
+    aggregate_rate = 100.0 * aggregate_hit / aggregate_found if aggregate_found else 0.0
+    if aggregate_hit * 100.0 < args.aggregate * aggregate_found:
+        failures.append(
+            f"  src/ aggregate: {aggregate_rate:.1f}% lines "
+            f"({aggregate_hit}/{aggregate_found}) is below the "
+            f"{args.aggregate:g}% aggregate floor"
+        )
+
+    if failures:
+        print(
+            f"coverage_floor: FAILED -- {len(failures)} floor(s) violated in {report}:",
+            file=sys.stderr,
+        )
+        print("\n".join(failures), file=sys.stderr)
+        return 1
+
+    print(
+        f"coverage_floor: ok -- {len(records)} files under src/, aggregate "
+        f"{aggregate_rate:.1f}% lines ({aggregate_hit}/{aggregate_found}), "
+        f"every file at or above {args.per_file:g}%, aggregate target {args.aggregate:g}%"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))