vibefpmgui.coverage

  1"""Test coverage (Build options → Coverage): which lines the tests ran.
  2
  3With coverage on, fpm compiles with gfortran's ``--coverage`` (a build
  4folder of its own, as fpm does for any flags), and the test programs write
  5how often each line ran into ``.gcda`` files next to the objects. After a
  6test run, a `Collector` has ``gcov`` (the one that matches the compiler)
  7read them, as JSON, several at a time; `read_gcov` turns that into
  8`FileCoverage` objects: the editor's gutter shows covered and missed lines,
  9and the *Coverage* panel the totals and each file's numbers.
 10
 11No Qt here: `vibefpmgui.window.coverage` runs the collector on a pool thread.
 12"""
 13
 14from __future__ import annotations
 15
 16import json
 17import os
 18import re
 19import shutil
 20import subprocess
 21import threading
 22from collections.abc import Iterable
 23from concurrent.futures import ThreadPoolExecutor
 24from dataclasses import dataclass, field
 25from pathlib import Path
 26
 27from .console import NO_WINDOW
 28
 29FLAG = "--coverage"  # gfortran's (compile and link)
 30FILES_PER_GCOV = 10  # .gcda files given to each gcov run
 31
 32
 33@dataclass
 34class FunctionCoverage:
 35    name: str
 36    line: int
 37    end_line: int
 38    hits: int
 39
 40
 41@dataclass
 42class FileCoverage:
 43    path: Path
 44    lines: dict[int, int] = field(default_factory=dict)  # line -> times run (lines with code only)
 45    functions: list[FunctionCoverage] = field(default_factory=list)
 46    branches_found: int = 0
 47    branches_hit: int = 0
 48    line_branches: dict[int, tuple[int, int]] = field(default_factory=dict)  # line -> (taken, total)
 49
 50    @property
 51    def lines_found(self) -> int:
 52        return len(self.lines)
 53
 54    @property
 55    def lines_hit(self) -> int:
 56        return sum(1 for n in self.lines.values() if n > 0)
 57
 58    @property
 59    def functions_found(self) -> int:
 60        return len(self.functions)
 61
 62    @property
 63    def functions_hit(self) -> int:
 64        return sum(1 for f in self.functions if f.hits > 0)
 65
 66    @property
 67    def percent(self) -> float:
 68        return 100.0 * self.lines_hit / self.lines_found if self.lines_found else 0.0
 69
 70    def partial(self, line: int) -> bool:
 71        """The line ran, but not every one of its branches was taken."""
 72        taken, total = self.line_branches.get(line, (0, 0))
 73        return self.lines.get(line, 0) > 0 and taken < total
 74
 75    def missed(self) -> list[tuple[int, int]]:
 76        """The lines not run, as (first, last) ranges: lines in between
 77        without code don't break a range."""
 78        ranges: list[tuple[int, int]] = []
 79        for line in sorted(self.lines):
 80            if self.lines[line] > 0:
 81                start = None
 82                continue
 83            if ranges and start is not None:
 84                ranges[-1] = (ranges[-1][0], line)
 85            else:
 86                ranges.append((line, line))
 87            start = line
 88        return ranges
 89
 90
 91PROCEDURE_RE = re.compile(r"^(?!\s*end\b)\s*(?:[\w\s(),=*:]*?\s)?(program|subroutine|function)\s+\w+",
 92                          re.IGNORECASE)
 93
 94
 95def procedure_kind(lines: list[str], line: int) -> str | None:
 96    """What a function of the coverage starts at (1-based ``line``):
 97    "program", "subroutine" or "function" (gcov only knows gfortran's names
 98    for them), read from its first line; None if it doesn't say."""
 99    if not 1 <= line <= len(lines):
100        return None
101    text = lines[line - 1].split("!", 1)[0]
102    found = PROCEDURE_RE.match(text)
103    return found.group(1).lower() if found else None
104
105
106def ranges_text(ranges: list[tuple[int, int]], limit: int = 12) -> str:
107    """ "8, 10–15, 22" (the first ``limit`` ranges, then "…")."""
108    parts = [str(a) if a == b else f"{a}–{b}" for a, b in ranges[:limit]]
109    return ", ".join(parts) + (", …" if len(ranges) > limit else "")
110
111
112def read_gcov(outputs: Iterable[str]) -> dict[Path, FileCoverage]:
113    """gcov's JSON (``--stdout --json-format``: a document per line, one per
114    .gcda file), by source file (absolute paths). A file in several (built
115    for the library and for the tests, say) has its counts added."""
116    lines: dict[Path, dict[int, int]] = {}
117    functions: dict[Path, dict[tuple[str, int], list[int]]] = {}  # (name, start) -> [end, hits]
118    branches: dict[Path, dict[tuple, int]] = {}  # (line, function, branch) -> times taken
119    for text in outputs:
120        for raw in text.splitlines():
121            if not raw.startswith("{"):
122                continue
123            try:
124                doc = json.loads(raw)
125            except ValueError:
126                continue
127            folder = Path(doc.get("current_working_directory") or ".")  # (where it was compiled)
128            for entry in doc.get("files", []):
129                path = (folder / entry["file"]).resolve()
130                counts = lines.setdefault(path, {})
131                for line in entry.get("lines", []):
132                    number = line["line_number"]
133                    counts[number] = counts.get(number, 0) + max(line["count"], 0)
134                    taken = branches.setdefault(path, {})
135                    for n, branch in enumerate(b for b in line.get("branches", []) if not b.get("throw")):
136                        key = (number, line.get("function_name", ""), n)
137                        taken[key] = taken.get(key, 0) + branch["count"]
138                found = functions.setdefault(path, {})
139                for fn in entry.get("functions", []):
140                    end_hits = found.setdefault((fn["name"], fn["start_line"]), [fn["end_line"], 0])
141                    end_hits[1] += fn["execution_count"]
142    files = {}
143    for path, counts in lines.items():
144        per_line: dict[int, tuple[int, int]] = {}
145        for (number, _fn, _n), count in branches.get(path, {}).items():
146            taken, total = per_line.get(number, (0, 0))
147            per_line[number] = (taken + (count > 0), total + 1)
148        fns = sorted((FunctionCoverage(name, start, end, hits)
149                      for (name, start), (end, hits) in functions.get(path, {}).items()), key=lambda f: f.line)
150        files[path] = FileCoverage(path, counts, fns, sum(t for _h, t in per_line.values()),
151                                   sum(h for h, _t in per_line.values()), per_line)
152    return files
153
154
155def totals(files: list[FileCoverage]) -> dict[str, tuple[int, int]]:
156    """(hit, found) for lines, functions and branches, over ``files``."""
157    return {"lines": (sum(f.lines_hit for f in files), sum(f.lines_found for f in files)),
158            "functions": (sum(f.functions_hit for f in files), sum(f.functions_found for f in files)),
159            "branches": (sum(f.branches_hit for f in files), sum(f.branches_found for f in files))}
160
161
162def find_gcov(compiler: str) -> str | None:
163    """The gcov that goes with the compiler (the same GCC: a gcov of another
164    version can't read the counts): next to it, plain or with the same
165    prefix (``arm64-apple-darwin20.0.0-gcov``), else the one on the PATH."""
166    found = shutil.which(compiler or "gfortran")
167    if found:
168        folder = Path(found).resolve().parent
169        name, ext = Path(found).stem, Path(found).suffix  # (".exe" on Windows)
170        prefix = name[:-len("gfortran")] if name.endswith("gfortran") else ""
171        for candidate in (f"{prefix}gcov{ext}", f"gcov{ext}"):
172            if (folder / candidate).is_file():
173                return str(folder / candidate)
174        for candidate in sorted(folder.glob(f"*-gcov{ext}")):
175            return str(candidate)
176    return shutil.which("gcov")
177
178
179def gcov_command(gcov: str, data_files: list[Path]) -> list[str]:
180    return [gcov, "--stdout", "--json-format", "--branch-probabilities", *map(str, data_files)]
181
182
183class Collector:
184    """gcov reading every .gcda file under the project's build folder,
185    ``jobs`` runs at a time (default: one per CPU). `run` blocks (call it on
186    a pool thread); `stop`, from another thread, ends it early."""
187
188    def __init__(self, gcov: str, build_dir: Path, jobs: int | None = None):
189        self.gcov, self.build_dir = gcov, build_dir
190        self.jobs = jobs or os.cpu_count() or 1
191        self._stopped = False
192        self._running: set[subprocess.Popen] = set()
193        self._lock = threading.Lock()
194
195    def run(self) -> tuple[dict[Path, FileCoverage], str]:
196        """The coverage, and an error message ("" if none)."""
197        data = sorted(self.build_dir.rglob("*.gcda")) if self.build_dir.is_dir() else []
198        chunks = [data[i:i + FILES_PER_GCOV] for i in range(0, len(data), FILES_PER_GCOV)]
199        with ThreadPoolExecutor(self.jobs) as pool:
200            results = list(pool.map(self._gcov, chunks))
201        if self._stopped:
202            return {}, "stopped"
203        files = read_gcov(out for out, _err in results)
204        errors = [err for _out, err in results if err]
205        return files, ("" if files or not errors else errors[-1])
206
207    def _gcov(self, chunk: list[Path]) -> tuple[str, str]:
208        """One gcov run's output, and its error ("" if it worked)."""
209        if self._stopped:
210            return "", ""
211        try:
212            process = subprocess.Popen(
213                gcov_command(self.gcov, chunk), cwd=self.build_dir, stdout=subprocess.PIPE,
214                stderr=subprocess.PIPE, text=True, errors="replace",
215                **NO_WINDOW)
216        except OSError as error:
217            return "", str(error)
218        with self._lock:
219            self._running.add(process)
220        out, err = process.communicate()
221        with self._lock:
222            self._running.discard(process)
223        message = err.strip().splitlines()[-1] if err.strip() else f"gcov exited with {process.returncode}"
224        return out, ("" if process.returncode == 0 else message)
225
226    def stop(self):
227        self._stopped = True
228        with self._lock:
229            for process in self._running:
230                process.kill()
231
232
233def clear_counts(build_dir: Path) -> int:
234    """Remove the counts of earlier runs (.gcda files), so the next run's
235    coverage is its own. Returns how many there were."""
236    removed = 0
237    if build_dir.is_dir():
238        for path in build_dir.rglob("*.gcda"):
239            try:
240                path.unlink()
241                removed += 1
242            except OSError:
243                pass
244    return removed