Coverage for vibefpmgui/coverage.py: 93%
165 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-01 20:51 +0000
1"""Test coverage (Build options → Coverage): which lines the tests ran.
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.
11No Qt here: `vibefpmgui.window.coverage` runs the collector on a pool thread.
12"""
14from __future__ import annotations
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
27from .console import NO_WINDOW
29FLAG = "--coverage" # gfortran's (compile and link)
30FILES_PER_GCOV = 10 # .gcda files given to each gcov run
33@dataclass
34class FunctionCoverage:
35 name: str
36 line: int
37 end_line: int
38 hits: int
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)
50 @property
51 def lines_found(self) -> int:
52 return len(self.lines)
54 @property
55 def lines_hit(self) -> int:
56 return sum(1 for n in self.lines.values() if n > 0)
58 @property
59 def functions_found(self) -> int:
60 return len(self.functions)
62 @property
63 def functions_hit(self) -> int:
64 return sum(1 for f in self.functions if f.hits > 0)
66 @property
67 def percent(self) -> float:
68 return 100.0 * self.lines_hit / self.lines_found if self.lines_found else 0.0
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
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
91PROCEDURE_RE = re.compile(r"^(?!\s*end\b)\s*(?:[\w\s(),=*:]*?\s)?(program|subroutine|function)\s+\w+",
92 re.IGNORECASE)
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
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 "")
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
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))}
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: 167 ↛ 176line 167 didn't jump to line 176 because the condition on line 167 was always true
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}"): 171 ↛ 174line 171 didn't jump to line 174 because the loop on line 171 didn't complete
172 if (folder / candidate).is_file(): 172 ↛ 171line 172 didn't jump to line 171 because the condition on line 172 was always true
173 return str(folder / candidate)
174 for candidate in sorted(folder.glob(f"*-gcov{ext}")):
175 return str(candidate)
176 return shutil.which("gcov")
179def gcov_command(gcov: str, data_files: list[Path]) -> list[str]:
180 return [gcov, "--stdout", "--json-format", "--branch-probabilities", *map(str, data_files)]
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."""
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()
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])
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)
226 def stop(self):
227 self._stopped = True
228 with self._lock:
229 for process in self._running:
230 process.kill()
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