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