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

Coverage measures which parts of the library the CTest suite actually executes.
It is the complement to mutation testing (scripts/mutation_test.py): coverage
finds code the tests never reach, mutation testing finds code the tests reach
but do not really check.

What is measured is the library's own translation units: src/error.c and the
generated src/errno.c. The macros in the public header are deliberately not
measured -- GCC attributes an expanded macro to its call site, so header logic
would be reported as lines of the test that used it. Use mutation testing
(--target include/akerror.tmpl.h) to check how well those macros are tested.

This harness has no third-party dependencies -- just gcov, which ships with the
compiler, plus the project's normal cmake/ctest toolchain. By default it
configures its own instrumented build directory so the ordinary build tree is
left alone and coverage counters can never be stale.

Usage:
    scripts/coverage.py [options]

    --source-root DIR     repo root (default: parent of this script's dir)
    --build-dir DIR       instrumented build dir (default: <root>/build/coverage)
    --no-configure        reuse the build dir as-is (do not cmake/build)
    --no-run              report existing counters (do not reset and run ctest)
    --threshold PCT       exit non-zero if line coverage < PCT (default: 0 = off)
    --branch-threshold P  exit non-zero if branch coverage < P (default: 0 = off)
    --junit PATH          write a JUnit XML report to this path
    --max-uncovered N     uncovered lines to list per file (default: 40, 0 = all)
    --exclude SUBSTR      skip reported paths containing SUBSTR; repeatable
    --gcov PROG           gcov program (default: $GCOV or "gcov")
"""

import argparse
import collections
import json
import os
import subprocess
import sys


# --------------------------------------------------------------------------- #
# Running the instrumented suite
# --------------------------------------------------------------------------- #

def configure_and_build(cmake, root, build, jobs):
    """Configure an instrumented build dir and build everything in it."""
    cmds = [
        [cmake, "-S", root, "-B", build, "-DAKERR_COVERAGE=ON"],
        [cmake, "--build", build] + (["--parallel", str(jobs)] if jobs else []),
    ]
    for cmd in cmds:
        print("+ " + " ".join(cmd))
        if subprocess.call(cmd) != 0:
            return False
    return True


def reset_counters(build):
    """Delete accumulated .gcda files so each run reports one suite run."""
    removed = 0
    for dirpath, _dirs, files in os.walk(build):
        for name in files:
            if name.endswith(".gcda"):
                os.unlink(os.path.join(dirpath, name))
                removed += 1
    if removed:
        print(f"Reset {removed} coverage data file(s).")


def run_ctest(build, ctest):
    """Run the suite. Returns True if every test passed.

    Coverage counters are flushed at exit(), and the tests that are expected to
    fail exit via the library's unhandled-error handler (exit(), not abort()),
    so WILL_FAIL tests still contribute their counters.
    """
    cmd = [ctest, "--output-on-failure"]
    print("+ " + " ".join(cmd) + f"   (in {build})")
    return subprocess.call(cmd, cwd=build) == 0


# --------------------------------------------------------------------------- #
# Collecting gcov data
# --------------------------------------------------------------------------- #

class FileCov:
    """Merged coverage for one source file, across every object that built it.

    A file compiled into more than one object -- or a header included by several
    translation units -- is reported once, with counts summed. Branches are
    merged by (line, index within line), so differing expansions of the same
    line contribute the union of their branches.
    """

    def __init__(self):
        self.lines = collections.Counter()      # lineno -> execution count
        self.branches = collections.Counter()   # (lineno, idx) -> taken count
        self.funcs = collections.Counter()      # (name, start_line) -> count

    def merge(self, entry):
        for ln in entry.get("lines", []):
            no = ln["line_number"]
            self.lines[no] += ln.get("count", 0)
            for idx, br in enumerate(ln.get("branches", [])):
                if br.get("throw"):
                    continue
                self.branches[(no, idx)] += br.get("count", 0)
        for fn in entry.get("functions", []):
            key = (fn.get("name", "?"), fn.get("start_line", 0))
            self.funcs[key] += fn.get("execution_count", 0)

    @staticmethod
    def _ratio(counter):
        total = len(counter)
        hit = sum(1 for v in counter.values() if v > 0)
        return hit, total

    def line_stats(self):
        return self._ratio(self.lines)

    def branch_stats(self):
        return self._ratio(self.branches)

    def func_stats(self):
        return self._ratio(self.funcs)

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


def find_notes(build):
    """Every .gcno in the build tree: one per instrumented translation unit."""
    notes = []
    for dirpath, _dirs, files in os.walk(build):
        for name in files:
            if name.endswith(".gcno"):
                notes.append(os.path.join(dirpath, name))
    return sorted(notes)


def gcov_json(gcov, note, cwd):
    """Run gcov on one .gcno and return its parsed JSON, or None on failure.

    --stdout keeps gcov from littering .gcov files in the build tree. A .gcno
    with no matching .gcda still reports, with all counts zero, which is the
    correct answer for a translation unit no test executed.
    """
    cmd = [gcov, "--branch-probabilities", "--json-format", "--stdout", note]
    proc = subprocess.Popen(cmd, cwd=cwd, stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE)
    out, err = proc.communicate()
    if proc.returncode != 0:
        sys.stderr.write(f"warning: {' '.join(cmd)} failed:\n"
                         f"{err.decode(errors='replace')}")
        return None
    try:
        return json.loads(out.decode(errors="replace"))
    except ValueError as exc:
        sys.stderr.write(f"warning: unparseable gcov output for {note}: {exc}\n")
        return None


def display_path(path, build, root):
    """Label a source: build-relative for generated files, else repo-relative.

    Returns None for anything outside both trees (system headers, toolchain
    internals). Generated sources are labelled relative to the build dir so the
    report and its thresholds do not change with the build dir's location.
    """
    for base in (build, root):
        if path.startswith(base + os.sep):
            return os.path.relpath(path, base)
    return None


def collect(gcov, build, root, excludes):
    """Merge gcov data for every instrumented unit into {display path: FileCov}."""
    covs = {}
    for note in find_notes(build):
        data = gcov_json(gcov, note, build)
        if data is None:
            continue
        # Paths in the report are relative to the directory the unit was
        # compiled in, which gcov records in the notes file.
        compile_dir = data.get("current_working_directory") or build
        for entry in data.get("files", []):
            path = entry.get("file", "")
            if not os.path.isabs(path):
                path = os.path.join(compile_dir, path)
            display = display_path(os.path.realpath(path), build, root)
            if display is None:
                continue  # system headers, toolchain internals
            if any(x in display for x in excludes):
                continue
            covs.setdefault(display, FileCov()).merge(entry)
    return covs


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

def pct(hit, total):
    return 100.0 * hit / total if total else 100.0


def fmt_ratio(hit, total):
    if not total:
        return f"{'-':>9}      -"
    return f"{hit:>4}/{total:<4} {pct(hit, total):5.1f}%"


def compress(numbers):
    """[1,2,3,7,9,10] -> '1-3, 7, 9-10' for readable uncovered-line lists."""
    out, start, prev = [], None, None
    for n in numbers:
        if start is None:
            start = prev = n
        elif n == prev + 1:
            prev = n
        else:
            out.append(f"{start}" if start == prev else f"{start}-{prev}")
            start = prev = n
    if start is not None:
        out.append(f"{start}" if start == prev else f"{start}-{prev}")
    return ", ".join(out)


def report(covs, max_uncovered):
    """Print the per-file table and uncovered detail; return overall percentages."""
    width = max([len(p) for p in covs] + [len("TOTAL")])
    print("\n" + "=" * 72)
    print("CODE COVERAGE SUMMARY")
    print("=" * 72)
    print(f"  {'FILE':<{width}}  {'LINES':^15}  {'BRANCHES':^15}  FUNCS")

    totals = [0, 0, 0, 0, 0, 0]  # lines hit/total, branches hit/total, funcs
    for path in sorted(covs):
        cov = covs[path]
        lh, lt = cov.line_stats()
        bh, bt = cov.branch_stats()
        fh, ft = cov.func_stats()
        for i, v in enumerate((lh, lt, bh, bt, fh, ft)):
            totals[i] += v
        print(f"  {path:<{width}}  {fmt_ratio(lh, lt)}  {fmt_ratio(bh, bt)}  "
              f"{fh}/{ft}")

    lh, lt, bh, bt, fh, ft = totals
    print(f"  {'-' * width}  {'-' * 15}  {'-' * 15}  -----")
    print(f"  {'TOTAL':<{width}}  {fmt_ratio(lh, lt)}  {fmt_ratio(bh, bt)}  "
          f"{fh}/{ft}")

    for path in sorted(covs):
        missing = covs[path].uncovered_lines()
        if not missing:
            continue
        shown = missing if not max_uncovered else missing[:max_uncovered]
        more = "" if len(shown) == len(missing) else \
            f" ... (+{len(missing) - len(shown)} more)"
        print(f"\n  uncovered in {path} ({len(missing)} line(s)):")
        print(f"    {compress(shown)}{more}")

    return pct(lh, lt), pct(bh, bt)


def _xml_escape(text):
    return (str(text).replace("&", "&amp;").replace("<", "&lt;")
            .replace(">", "&gt;").replace('"', "&quot;"))


def write_junit(path, covs, line_threshold, branch_threshold):
    """One <testcase> per file per metric; below-threshold is a <failure>."""
    cases = []
    for f in sorted(covs):
        cov = covs[f]
        cases.append((f, "lines", cov.line_stats(), line_threshold))
        cases.append((f, "branches", cov.branch_stats(), branch_threshold))
    for metric, thr, stats in (("lines", line_threshold,
                                [c.line_stats() for c in covs.values()]),
                               ("branches", branch_threshold,
                                [c.branch_stats() for c in covs.values()])):
        cases.append(("TOTAL", metric,
                      (sum(h for h, _t in stats), sum(t for _h, t in stats)),
                      thr))

    fails = sum(1 for _f, _m, (h, t), thr in cases
                if t and thr > 0 and pct(h, t) < thr)
    out = ['<?xml version="1.0" encoding="UTF-8"?>',
           f'<testsuites name="coverage" tests="{len(cases)}" '
           f'failures="{fails}">',
           f'  <testsuite name="coverage" tests="{len(cases)}" '
           f'failures="{fails}">']
    for f, metric, (hit, total), thr in cases:
        name = _xml_escape(f"{f} {metric}")
        detail = _xml_escape(f"{hit}/{total} ({pct(hit, total):.1f}%)"
                             if total else "no data")
        out.append(f'    <testcase name="{name}" '
                   f'classname="coverage.{_xml_escape(metric)}" time="0">')
        if total and thr > 0 and pct(hit, total) < thr:
            out.append(f'      <failure message="{metric} coverage '
                       f'{pct(hit, total):.1f}% &lt; threshold {thr:.1f}%">'
                       f'{detail}</failure>')
        else:
            out.append(f'      <system-out>{detail}</system-out>')
        out.append('    </testcase>')
    out.append('  </testsuite>')
    out.append('</testsuites>')
    with open(path, "w") as fh:
        fh.write("\n".join(out) + "\n")


def main():
    # Line-buffer stdout so progress is visible live under CI / the cmake target.
    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 libakerror")
    ap.add_argument("--source-root", default=default_root)
    ap.add_argument("--build-dir", default=None)
    ap.add_argument("--configure", dest="configure", action="store_true",
                    default=True)
    ap.add_argument("--no-configure", dest="configure", action="store_false")
    ap.add_argument("--run", dest="run", action="store_true", default=True)
    ap.add_argument("--no-run", dest="run", action="store_false")
    ap.add_argument("--threshold", type=float, default=0.0,
                    help="fail if line coverage is below this percentage")
    ap.add_argument("--branch-threshold", type=float, default=0.0,
                    help="fail if branch coverage is below this percentage")
    ap.add_argument("--junit", default=None,
                    help="write a JUnit XML report to this path")
    ap.add_argument("--max-uncovered", type=int, default=40)
    ap.add_argument("--exclude", action="append", default=None,
                    help="skip reported paths containing this substring")
    ap.add_argument("--gcov", default=os.environ.get("GCOV", "gcov"))
    ap.add_argument("--cmake", default=os.environ.get("CMAKE", "cmake"))
    ap.add_argument("--ctest", default=os.environ.get("CTEST", "ctest"))
    ap.add_argument("-j", "--jobs", type=int, default=0)
    args = ap.parse_args()

    root = os.path.realpath(args.source_root)
    build = os.path.realpath(args.build_dir or os.path.join(root, "build",
                                                            "coverage"))
    # The tests exercise the library; their own source is not what we measure.
    excludes = args.exclude if args.exclude is not None else ["tests/"]

    if args.configure:
        if not configure_and_build(args.cmake, root, build, args.jobs):
            sys.stderr.write("\nInstrumented build FAILED; aborting.\n")
            return 2
    elif not os.path.isdir(build):
        sys.stderr.write(f"No build dir at {build} (drop --no-configure).\n")
        return 2

    tests_ok = True
    if args.run:
        reset_counters(build)
        tests_ok = run_ctest(build, args.ctest)
        if not tests_ok:
            sys.stderr.write("\nwarning: some tests FAILED; coverage below is "
                             "still reported, but the run is not green.\n")

    covs = collect(args.gcov, build, root, excludes)
    if not covs:
        sys.stderr.write("No coverage data found. Was the build instrumented "
                         "(-DAKERR_COVERAGE=ON) and the suite run?\n")
        return 2

    line_pct, branch_pct = report(covs, args.max_uncovered)

    if args.junit:
        junit_path = os.path.abspath(args.junit)
        write_junit(junit_path, covs, args.threshold, args.branch_threshold)
        print(f"\nJUnit report written to: {junit_path}")

    rc = 0
    if not tests_ok:
        print("\nFAIL: the CTest suite did not pass.")
        rc = 1

    # Thresholds gate every file as well as the total: a small, well-covered
    # file (the generated status-name table) must not mask a regression in a
    # bigger one. Files with no branches at all are not branch-gated.
    checks = [("total", "line", line_pct, args.threshold),
              ("total", "branch", branch_pct, args.branch_threshold)]
    for path in sorted(covs):
        lh, lt = covs[path].line_stats()
        bh, bt = covs[path].branch_stats()
        checks.append((path, "line", pct(lh, lt), args.threshold))
        if bt:
            checks.append((path, "branch", pct(bh, bt), args.branch_threshold))
    for path, metric, value, threshold in checks:
        if threshold > 0 and value < threshold:
            print(f"\nFAIL: {path} {metric} coverage {value:.1f}% < threshold "
                  f"{threshold:.1f}%")
            rc = 1
    return rc


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