#!/usr/bin/env python3
"""
Code coverage harness for libakstdlib.

Reports which lines, branches and functions of the library the CTest suite
actually executed. Coverage answers "what did the tests touch"; the mutation
harness in scripts/mutation_test.py answers the stronger question "would the
tests notice if it broke". They are complementary: a line can be covered and
still have no assertion behind it, so an uncovered line is definitely untested
while a covered one is only maybe tested.

Requires a build configured with -DAKSL_COVERAGE=ON, which compiles the library
and its tests with --coverage. That emits a .gcno next to every object at build
time and a .gcda next to it at run time; this script feeds those to gcov and
aggregates its JSON output.

There are no third-party dependencies (Python stdlib + gcc's own gcov), so this
works anywhere the project already builds -- no lcov, gcovr or genhtml needed.

Usage:
    scripts/coverage.py --build build-coverage [options]

    --build DIR         build tree to read .gcda from (default: build-coverage)
    --source-root DIR   repo root (default: parent of this script's dir)
    --gcov PROG         gcov binary to use (default: $GCOV or gcov)
    --include PREFIX    only report sources under this repo-relative prefix;
                        repeatable. Default: src, include
    --exclude PREFIX    drop sources under this repo-relative prefix;
                        repeatable. Default: tests, deps, build
    --zero              delete every .gcda in the build tree, then exit. Run
                        before the suite so counters do not accumulate across
                        runs.
    --run-tests         run ctest in the build tree first (implies --zero)
    --threshold PCT     exit non-zero if total line coverage < PCT (default 0)
    --branch-threshold PCT
                        exit non-zero if total branch coverage < PCT
    --summary-only      omit the per-file uncovered-line listing
    --output PATH       also write the text report to PATH (CTest hides the
                        output of a passing test, so the in-suite report lands
                        in <build>/coverage-summary.txt this way)
    --cobertura PATH    also write a Cobertura XML report (for CI publishers)
"""

import argparse
import glob
import json
import os
import subprocess
import sys
import xml.sax.saxutils as saxutils

DEFAULT_INCLUDE = ["src", "include"]
DEFAULT_EXCLUDE = ["tests", "deps", "build"]


# --------------------------------------------------------------------------- #
# gcov invocation and parsing
# --------------------------------------------------------------------------- #

class FileCoverage:
    """Accumulated coverage for one source file, keyed by line number.

    A single source file can be represented in several .gcda files -- a header
    compiled into every test binary, or a .c file linked more than once -- so
    counts are summed across all of them rather than overwritten.
    """

    def __init__(self, path):
        self.path = path              # repo-relative
        self.lines = {}               # line number -> execution count
        self.branches = {}            # line number -> list of branch counts
        self.functions = {}           # function name -> execution count

    def add_line(self, lineno, count):
        self.lines[lineno] = self.lines.get(lineno, 0) + count

    def add_branches(self, lineno, counts):
        cur = self.branches.setdefault(lineno, [0] * len(counts))
        # Defensive: different translation units should agree on the branch
        # count for a line, but never let a mismatch raise.
        if len(cur) != len(counts):
            if len(counts) > len(cur):
                cur.extend([0] * (len(counts) - len(cur)))
            counts = counts + [0] * (len(cur) - len(counts))
        for i, c in enumerate(counts):
            cur[i] += c

    def add_function(self, name, count):
        self.functions[name] = self.functions.get(name, 0) + count

    # -- derived totals ---------------------------------------------------- #

    @property
    def lines_total(self):
        return len(self.lines)

    @property
    def lines_covered(self):
        return sum(1 for c in self.lines.values() if c > 0)

    @property
    def branches_total(self):
        return sum(len(b) for b in self.branches.values())

    @property
    def branches_covered(self):
        return sum(1 for b in self.branches.values() for c in b if c > 0)

    @property
    def functions_total(self):
        return len(self.functions)

    @property
    def functions_covered(self):
        return sum(1 for c in self.functions.values() if c > 0)

    def uncovered_lines(self):
        return sorted(n for n, c in self.lines.items() if c == 0)

    def uncovered_functions(self):
        return sorted(n for n, c in self.functions.items() if c == 0)


def pct(covered, total):
    """Coverage percentage. A file with nothing instrumented counts as 100%,
    matching gcov/lcov: there is nothing there to leave untested."""
    return 100.0 * covered / total if total else 100.0


def find_gcda(build):
    return sorted(glob.glob(os.path.join(build, "**", "*.gcda"), recursive=True))


def run_gcov(gcov, gcda):
    """Run gcov on one .gcda and return the parsed JSON, or None on failure.

    gcov resolves the matching .gcno relative to the .gcda, so it is invoked
    from the .gcda's directory with a bare filename.
    """
    workdir = os.path.dirname(gcda) or "."
    cmd = [gcov, "--json-format", "--stdout", "--branch-probabilities",
           os.path.basename(gcda)]
    try:
        proc = subprocess.run(cmd, cwd=workdir, stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE)
    except OSError as exc:
        sys.stderr.write(f"warning: cannot run {gcov}: {exc}\n")
        return None
    if proc.returncode != 0:
        sys.stderr.write(f"warning: gcov failed on {gcda}:\n"
                         f"{proc.stderr.decode(errors='replace')}")
        return None
    text = proc.stdout.decode(errors="replace").strip()
    if not text:
        return None
    try:
        return json.loads(text)
    except ValueError as exc:
        sys.stderr.write(f"warning: unparseable gcov JSON for {gcda}: {exc}\n")
        return None


def rel_to_root(path, cwd, root):
    """Resolve a path from gcov's JSON to a repo-relative path, or None if it
    falls outside the repo (libc headers, the toolchain, an out-of-tree dep)."""
    if not os.path.isabs(path):
        path = os.path.join(cwd, path)
    path = os.path.realpath(path)
    try:
        rel = os.path.relpath(path, root)
    except ValueError:                                # different drive (Windows)
        return None
    if rel.startswith(os.pardir):
        return None
    return rel.replace(os.sep, "/")


def selected(rel, includes, excludes):
    """True if a repo-relative path passes the include/exclude prefix filters.
    Excludes win, so `--include src --exclude src/generated` behaves sanely."""
    def under(prefix):
        return rel == prefix or rel.startswith(prefix.rstrip("/") + "/")
    if any(under(e) for e in excludes):
        return False
    if not includes:
        return True
    return any(under(i) for i in includes)


def collect(build, root, gcov, includes, excludes):
    """Aggregate coverage for every selected source file in the build tree."""
    files = {}
    gcda_files = find_gcda(build)
    for gcda in gcda_files:
        data = run_gcov(gcov, gcda)
        if not data:
            continue
        cwd = data.get("current_working_directory") or os.path.dirname(gcda)
        for entry in data.get("files", []):
            rel = rel_to_root(entry.get("file", ""), cwd, root)
            if rel is None or not selected(rel, includes, excludes):
                continue
            fc = files.setdefault(rel, FileCoverage(rel))
            for line in entry.get("lines", []):
                lineno = line.get("line_number")
                if lineno is None:
                    continue
                fc.add_line(lineno, line.get("count", 0))
                branches = line.get("branches") or []
                if branches:
                    fc.add_branches(lineno, [b.get("count", 0) for b in branches])
            for fn in entry.get("functions", []):
                name = fn.get("demangled_name") or fn.get("name")
                if name:
                    fc.add_function(name, fn.get("execution_count", 0))
    return files, len(gcda_files)


# --------------------------------------------------------------------------- #
# Reporting
# --------------------------------------------------------------------------- #

def line_ranges(numbers):
    """Compress [3,4,5,9,11,12] into ['3-5', '9', '11-12'] for readability."""
    out = []
    start = prev = None
    for n in numbers:
        if start is None:
            start = prev = n
        elif n == prev + 1:
            prev = n
        else:
            out.append(str(start) if start == prev else f"{start}-{prev}")
            start = prev = n
    if start is not None:
        out.append(str(start) if start == prev else f"{start}-{prev}")
    return out


def report(files, summary_only):
    """Render the text report. Returns (totals dict, report text)."""
    names = sorted(files)
    width = max([len(n) for n in names] + [len("TOTAL")])
    out = []

    header = (f"  {'file'.ljust(width)}  {'lines':>15}  {'branches':>15}  "
              f"{'functions':>15}")
    out.append("=" * len(header))
    out.append("CODE COVERAGE SUMMARY")
    out.append("=" * len(header))
    out.append(header)
    out.append("  " + "-" * (len(header) - 2))

    def row(name, lc, lt, bc, bt, fc_, ft):
        cell = lambda c, t: f"{pct(c, t):5.1f}% {c:4d}/{t:<4d}"
        out.append(f"  {name.ljust(width)}  {cell(lc, lt):>15}  "
                   f"{cell(bc, bt):>15}  {cell(fc_, ft):>15}")

    tot = [0, 0, 0, 0, 0, 0]
    for name in names:
        f = files[name]
        vals = (f.lines_covered, f.lines_total, f.branches_covered,
                f.branches_total, f.functions_covered, f.functions_total)
        row(name, *vals)
        tot = [t + v for t, v in zip(tot, vals)]

    out.append("  " + "-" * (len(header) - 2))
    row("TOTAL", *tot)

    if not summary_only:
        for name in names:
            f = files[name]
            gaps = f.uncovered_lines()
            missing_fns = f.uncovered_functions()
            if not gaps and not missing_fns:
                continue
            out.append(f"\n{name}: uncovered (each one is a missing test)")
            if missing_fns:
                out.append("  functions never called:")
                for fn in missing_fns:
                    out.append(f"    {fn}")
            if gaps:
                out.append(f"  lines ({len(gaps)}): "
                           + ", ".join(line_ranges(gaps)))

    text = "\n".join(out) + "\n"
    sys.stdout.write(text)
    return text, {
        "lines_covered": tot[0], "lines_total": tot[1],
        "branches_covered": tot[2], "branches_total": tot[3],
        "functions_covered": tot[4], "functions_total": tot[5],
    }


def write_cobertura(path, files, root, totals):
    """Emit a minimal Cobertura report. One <class> per source file, which is
    what CI coverage publishers expect for C."""
    esc = saxutils.quoteattr
    lr = pct(totals["lines_covered"], totals["lines_total"]) / 100.0
    br = pct(totals["branches_covered"], totals["branches_total"]) / 100.0
    out = ['<?xml version="1.0" encoding="UTF-8"?>',
           f'<coverage line-rate="{lr:.4f}" branch-rate="{br:.4f}" '
           f'lines-covered="{totals["lines_covered"]}" '
           f'lines-valid="{totals["lines_total"]}" '
           f'branches-covered="{totals["branches_covered"]}" '
           f'branches-valid="{totals["branches_total"]}" '
           'complexity="0" version="0" timestamp="0">',
           '  <sources>', f'    <source>{saxutils.escape(root)}</source>',
           '  </sources>', '  <packages>',
           '    <package name="akstdlib" '
           f'line-rate="{lr:.4f}" branch-rate="{br:.4f}" complexity="0">',
           '      <classes>']
    for name in sorted(files):
        f = files[name]
        flr = pct(f.lines_covered, f.lines_total) / 100.0
        fbr = pct(f.branches_covered, f.branches_total) / 100.0
        cls = os.path.basename(name)
        out.append(f'        <class name={esc(cls)} filename={esc(name)} '
                   f'line-rate="{flr:.4f}" branch-rate="{fbr:.4f}" '
                   'complexity="0">')
        out.append('          <methods/>')
        out.append('          <lines>')
        for lineno in sorted(f.lines):
            count = f.lines[lineno]
            brs = f.branches.get(lineno) or []
            if brs:
                taken = sum(1 for c in brs if c > 0)
                cond = (f' branch="true" condition-coverage='
                        f'{esc(f"{pct(taken, len(brs)):.0f}% ({taken}/{len(brs)})")}')
            else:
                cond = ' branch="false"'
            out.append(f'            <line number="{lineno}" '
                       f'hits="{count}"{cond}/>')
        out.append('          </lines>')
        out.append('        </class>')
    out += ['      </classes>', '    </package>', '  </packages>',
            '</coverage>']
    with open(path, "w") as fh:
        fh.write("\n".join(out) + "\n")


# --------------------------------------------------------------------------- #
# Driver
# --------------------------------------------------------------------------- #

def zero_counters(build):
    """Delete accumulated .gcda. gcov counters are cumulative across runs, so
    without this a report mixes in every earlier suite run (and any test binary
    run by hand), which quietly overstates coverage."""
    removed = 0
    for gcda in find_gcda(build):
        try:
            os.remove(gcda)
            removed += 1
        except OSError as exc:
            sys.stderr.write(f"warning: cannot remove {gcda}: {exc}\n")
    return removed


def main():
    try:
        sys.stdout.reconfigure(line_buffering=True)
    except (AttributeError, ValueError):
        pass
    here = os.path.dirname(os.path.abspath(__file__))
    default_root = os.path.dirname(here)

    ap = argparse.ArgumentParser(description="Code coverage for libakstdlib")
    ap.add_argument("--build", default="build-coverage")
    ap.add_argument("--source-root", default=default_root)
    ap.add_argument("--gcov", default=os.environ.get("GCOV", "gcov"))
    ap.add_argument("--include", action="append", default=None)
    ap.add_argument("--exclude", action="append", default=None)
    ap.add_argument("--zero", action="store_true")
    ap.add_argument("--run-tests", action="store_true")
    ap.add_argument("--threshold", type=float, default=0.0)
    ap.add_argument("--branch-threshold", type=float, default=0.0)
    ap.add_argument("--summary-only", action="store_true")
    ap.add_argument("--output", default=None,
                    help="also write the text report to this path")
    ap.add_argument("--cobertura", default=None)
    args = ap.parse_args()

    root = os.path.realpath(args.source_root)
    build = os.path.abspath(args.build)
    includes = args.include if args.include is not None else DEFAULT_INCLUDE
    if args.exclude is not None:
        excludes = args.exclude
    else:
        # Excludes win over includes, so a *default* exclude must not silently
        # cancel an explicit --include: `--include tests` has to report tests/
        # even though tests/ is excluded by default.
        excludes = [e for e in DEFAULT_EXCLUDE
                    if not any(selected(e, [i], []) for i in (args.include or []))]

    if not os.path.isdir(build):
        sys.stderr.write(f"error: no such build directory: {build}\n"
                         "Configure one with:\n"
                         "  cmake -S . -B build-coverage -DAKSL_COVERAGE=ON\n"
                         "  cmake --build build-coverage\n")
        return 2

    if args.zero and not args.run_tests:
        n = zero_counters(build)
        print(f"Cleared {n} .gcda counter file(s) in {build}")
        return 0

    if args.run_tests:
        n = zero_counters(build)
        print(f"Cleared {n} .gcda counter file(s) in {build}")
        print("Running ctest ...")
        rc = subprocess.call(["ctest", "--test-dir", build,
                              "--output-on-failure"])
        if rc != 0:
            # Report anyway: partial coverage from a red suite is still useful,
            # and the ctest output above already shows what failed.
            sys.stderr.write(f"warning: ctest exited {rc}; "
                             "reporting coverage from a failing suite\n")
        print()

    files, n_gcda = collect(build, root, args.gcov, includes, excludes)
    if not n_gcda:
        sys.stderr.write(
            f"error: no .gcda files under {build}.\n"
            "Either the build was not configured with -DAKSL_COVERAGE=ON, or\n"
            "the test suite has not been run yet (try --run-tests).\n")
        return 2
    if not files:
        sys.stderr.write(
            f"error: {n_gcda} .gcda file(s) found, but none of the covered "
            "sources matched\nthe include/exclude filters "
            f"(include={includes}, exclude={excludes}).\n")
        return 2

    text, totals = report(files, args.summary_only)

    # CTest swallows the output of a passing test, so also drop the report on
    # disk: that is what makes the in-suite coverage_report entry useful without
    # having to re-run ctest under -V.
    if args.output:
        path = os.path.abspath(args.output)
        try:
            with open(path, "w") as fh:
                fh.write(text)
            print(f"\nReport written to: {path}")
        except OSError as exc:
            sys.stderr.write(f"warning: cannot write {path}: {exc}\n")

    if args.cobertura:
        path = os.path.abspath(args.cobertura)
        write_cobertura(path, files, root, totals)
        print(f"\nCobertura report written to: {path}")

    line_pct = pct(totals["lines_covered"], totals["lines_total"])
    branch_pct = pct(totals["branches_covered"], totals["branches_total"])
    failed = False
    if args.threshold > 0 and line_pct < args.threshold:
        print(f"\nFAIL: line coverage {line_pct:.1f}% < "
              f"threshold {args.threshold:.1f}%")
        failed = True
    if args.branch_threshold > 0 and branch_pct < args.branch_threshold:
        print(f"\nFAIL: branch coverage {branch_pct:.1f}% < "
              f"threshold {args.branch_threshold:.1f}%")
        failed = True
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())