1# This file is part of Hypothesis, which may be found at
2# https://github.com/HypothesisWorks/hypothesis/
3#
4# Copyright the Hypothesis Authors.
5# Individual contributors are listed in AUTHORS.rst and the git log.
6#
7# This Source Code Form is subject to the terms of the Mozilla Public License,
8# v. 2.0. If a copy of the MPL was not distributed with this file, You can
9# obtain one at https://mozilla.org/MPL/2.0/.
10
11import functools
12import os
13import re
14import subprocess
15import sys
16import sysconfig
17import types
18from collections import defaultdict
19from collections.abc import Iterable
20from enum import IntEnum
21from functools import lru_cache, reduce
22from os import sep
23from pathlib import Path
24from typing import TYPE_CHECKING, Optional
25
26from hypothesis._settings import Phase, Verbosity
27from hypothesis.internal.compat import PYPY
28from hypothesis.internal.escalation import is_hypothesis_file
29
30if TYPE_CHECKING:
31 from typing import TypeAlias
32
33Location: "TypeAlias" = tuple[str, int]
34Branch: "TypeAlias" = tuple[Optional[Location], Location]
35Trace: "TypeAlias" = set[Branch]
36
37
38@functools.cache
39def should_trace_file(fname: str) -> bool:
40 # fname.startswith("<") indicates runtime code-generation via compile,
41 # e.g. compile("def ...", "<string>", "exec") in e.g. attrs methods.
42 return not (is_hypothesis_file(fname) or fname.startswith("<"))
43
44
45# where possible, we'll use 3.12's new sys.monitoring module for low-overhead
46# coverage instrumentation; on older python versions we'll use sys.settrace.
47# tool_id = 1 is designated for coverage, but we intentionally choose a
48# non-reserved tool id so we can co-exist with coverage tools.
49MONITORING_TOOL_ID = 3
50if sys.version_info[:2] >= (3, 12):
51 MONITORING_EVENTS = {sys.monitoring.events.LINE: "trace_line"}
52
53
54class Tracer:
55 """A super-simple branch coverage tracer."""
56
57 __slots__ = (
58 "_previous_location",
59 "_should_trace",
60 "_tried_and_failed_to_trace",
61 "branches",
62 )
63
64 def __init__(self, *, should_trace: bool) -> None:
65 self.branches: Trace = set()
66 self._previous_location: Optional[Location] = None
67 self._tried_and_failed_to_trace = False
68 self._should_trace = should_trace and self.can_trace()
69
70 @staticmethod
71 def can_trace() -> bool:
72 return (
73 (sys.version_info[:2] < (3, 12) and sys.gettrace() is None)
74 or (
75 sys.version_info[:2] >= (3, 12)
76 and sys.monitoring.get_tool(MONITORING_TOOL_ID) is None
77 )
78 ) and not PYPY
79
80 def trace(self, frame, event, arg):
81 try:
82 if event == "call":
83 return self.trace
84 elif event == "line":
85 fname = frame.f_code.co_filename
86 if should_trace_file(fname):
87 current_location = (fname, frame.f_lineno)
88 self.branches.add((self._previous_location, current_location))
89 self._previous_location = current_location
90 except RecursionError:
91 pass
92
93 def trace_line(self, code: types.CodeType, line_number: int) -> None:
94 fname = code.co_filename
95 if not should_trace_file(fname):
96 # this function is only called on 3.12+, but we want to avoid an
97 # assertion to that effect for performance.
98 return sys.monitoring.DISABLE # type: ignore
99
100 current_location = (fname, line_number)
101 self.branches.add((self._previous_location, current_location))
102 self._previous_location = current_location
103
104 def __enter__(self):
105 self._tried_and_failed_to_trace = False
106
107 if not self._should_trace:
108 return self
109
110 if sys.version_info[:2] < (3, 12):
111 sys.settrace(self.trace)
112 return self
113
114 try:
115 sys.monitoring.use_tool_id(MONITORING_TOOL_ID, "scrutineer")
116 except ValueError:
117 # another thread may have registered a tool for MONITORING_TOOL_ID
118 # since we checked in can_trace.
119 self._tried_and_failed_to_trace = True
120 return self
121
122 for event, callback_name in MONITORING_EVENTS.items():
123 sys.monitoring.set_events(MONITORING_TOOL_ID, event)
124 callback = getattr(self, callback_name)
125 sys.monitoring.register_callback(MONITORING_TOOL_ID, event, callback)
126
127 return self
128
129 def __exit__(self, *args, **kwargs):
130 if not self._should_trace:
131 return
132
133 if sys.version_info[:2] < (3, 12):
134 sys.settrace(None)
135 return
136
137 if self._tried_and_failed_to_trace:
138 return
139
140 sys.monitoring.free_tool_id(MONITORING_TOOL_ID)
141 for event in MONITORING_EVENTS:
142 sys.monitoring.register_callback(MONITORING_TOOL_ID, event, None)
143
144
145UNHELPFUL_LOCATIONS = (
146 # There's a branch which is only taken when an exception is active while exiting
147 # a contextmanager; this is probably after the fault has been triggered.
148 # Similar reasoning applies to a few other standard-library modules: even
149 # if the fault was later, these still aren't useful locations to report!
150 # Note: The list is post-processed, so use plain "/" for separator here.
151 "/contextlib.py",
152 "/inspect.py",
153 "/re.py",
154 "/re/__init__.py", # refactored in Python 3.11
155 "/warnings.py",
156 # Quite rarely, the first AFNP line is in Pytest's internals.
157 "/_pytest/**",
158 "/pluggy/_*.py",
159 # used by pytest for failure formatting in the terminal
160 "/pygments/lexer.py",
161 # used by pytest for failure formatting
162 "/difflib.py",
163 "/reprlib.py",
164 "/typing.py",
165 "/conftest.py",
166 "/pprint.py",
167)
168
169
170def _glob_to_re(locs: Iterable[str]) -> str:
171 """Translate a list of glob patterns to a combined regular expression.
172 Only the * and ** wildcards are supported, and patterns including special
173 characters will only work by chance."""
174 # fnmatch.translate is not an option since its "*" consumes path sep
175 return "|".join(
176 loc.replace(".", re.escape("."))
177 .replace("**", r".+")
178 .replace("*", r"[^/]+")
179 .replace("/", re.escape(sep))
180 + r"\Z" # right anchored
181 for loc in locs
182 )
183
184
185def get_explaining_locations(traces):
186 # Traces is a dict[interesting_origin | None, set[frozenset[tuple[str, int]]]]
187 # Each trace in the set might later become a Counter instead of frozenset.
188 if not traces:
189 return {}
190
191 unions = {origin: set().union(*values) for origin, values in traces.items()}
192 seen_passing = {None}.union(*unions.pop(None, set()))
193
194 always_failing_never_passing = {
195 origin: reduce(set.intersection, [set().union(*v) for v in values])
196 - seen_passing
197 for origin, values in traces.items()
198 if origin is not None
199 }
200
201 # Build the observed parts of the control-flow graph for each origin
202 cf_graphs = {origin: defaultdict(set) for origin in unions}
203 for origin, seen_arcs in unions.items():
204 for src, dst in seen_arcs:
205 cf_graphs[origin][src].add(dst)
206 assert cf_graphs[origin][None], "Expected start node with >=1 successor"
207
208 # For each origin, our explanation is the always_failing_never_passing lines
209 # which are reachable from the start node (None) without passing through another
210 # AFNP line. So here's a whatever-first search with early stopping:
211 explanations = defaultdict(set)
212 for origin in unions:
213 queue = {None}
214 seen = set()
215 while queue:
216 assert queue.isdisjoint(seen), f"Intersection: {queue & seen}"
217 src = queue.pop()
218 seen.add(src)
219 if src in always_failing_never_passing[origin]:
220 explanations[origin].add(src)
221 else:
222 queue.update(cf_graphs[origin][src] - seen)
223
224 # The last step is to filter out explanations that we know would be uninformative.
225 # When this is the first AFNP location, we conclude that Scrutineer missed the
226 # real divergence (earlier in the trace) and drop that unhelpful explanation.
227 filter_regex = re.compile(_glob_to_re(UNHELPFUL_LOCATIONS))
228 return {
229 origin: {loc for loc in afnp_locs if not filter_regex.search(loc[0])}
230 for origin, afnp_locs in explanations.items()
231 }
232
233
234# see e.g. https://docs.python.org/3/library/sysconfig.html#posix-user
235# for examples of these path schemes
236STDLIB_DIRS = {
237 Path(sysconfig.get_path("platstdlib")).resolve(),
238 Path(sysconfig.get_path("stdlib")).resolve(),
239}
240SITE_PACKAGES_DIRS = {
241 Path(sysconfig.get_path("purelib")).resolve(),
242 Path(sysconfig.get_path("platlib")).resolve(),
243}
244
245EXPLANATION_STUB = (
246 "Explanation:",
247 " These lines were always and only run by failing examples:",
248)
249
250
251class ModuleLocation(IntEnum):
252 LOCAL = 0
253 SITE_PACKAGES = 1
254 STDLIB = 2
255
256 @classmethod
257 @lru_cache(1024)
258 def from_path(cls, path: str) -> "ModuleLocation":
259 path = Path(path).resolve()
260 # site-packages may be a subdir of stdlib or platlib, so it's important to
261 # check is_relative_to for this before the stdlib.
262 if any(path.is_relative_to(p) for p in SITE_PACKAGES_DIRS):
263 return cls.SITE_PACKAGES
264 if any(path.is_relative_to(p) for p in STDLIB_DIRS):
265 return cls.STDLIB
266 return cls.LOCAL
267
268
269# show local files first, then site-packages, then stdlib
270def _sort_key(path: str, lineno: int) -> tuple[int, str, int]:
271 return (ModuleLocation.from_path(path), path, lineno)
272
273
274def make_report(explanations, *, cap_lines_at=5):
275 report = defaultdict(list)
276 for origin, locations in explanations.items():
277 locations = list(locations)
278 locations.sort(key=lambda v: _sort_key(v[0], v[1]))
279 report_lines = [f" {fname}:{lineno}" for fname, lineno in locations]
280 if len(report_lines) > cap_lines_at + 1:
281 msg = " (and {} more with settings.verbosity >= verbose)"
282 report_lines[cap_lines_at:] = [msg.format(len(report_lines[cap_lines_at:]))]
283 if report_lines: # We might have filtered out every location as uninformative.
284 report[origin] = list(EXPLANATION_STUB) + report_lines
285 return report
286
287
288def explanatory_lines(traces, settings):
289 if Phase.explain in settings.phases and sys.gettrace() and not traces:
290 return defaultdict(list)
291 # Return human-readable report lines summarising the traces
292 explanations = get_explaining_locations(traces)
293 max_lines = 5 if settings.verbosity <= Verbosity.normal else float("inf")
294 return make_report(explanations, cap_lines_at=max_lines)
295
296
297# beware the code below; we're using some heuristics to make a nicer report...
298
299
300@functools.lru_cache
301def _get_git_repo_root() -> Path:
302 try:
303 where = subprocess.run(
304 ["git", "rev-parse", "--show-toplevel"],
305 check=True,
306 timeout=10,
307 capture_output=True,
308 text=True,
309 encoding="utf-8",
310 ).stdout.strip()
311 except Exception: # pragma: no cover
312 return Path().absolute().parents[-1]
313 else:
314 return Path(where)
315
316
317def tractable_coverage_report(trace: Trace) -> dict[str, list[int]]:
318 """Report a simple coverage map which is (probably most) of the user's code."""
319 coverage: dict = {}
320 t = dict(trace)
321 for file, line in set(t.keys()).union(t.values()) - {None}: # type: ignore
322 # On Python <= 3.11, we can use coverage.py xor Hypothesis' tracer,
323 # so the trace will be empty and this line never run under coverage.
324 coverage.setdefault(file, set()).add(line) # pragma: no cover
325 stdlib_fragment = f"{os.sep}lib{os.sep}python3.{sys.version_info.minor}{os.sep}"
326 return {
327 k: sorted(v)
328 for k, v in coverage.items()
329 if stdlib_fragment not in k
330 and (p := Path(k)).is_relative_to(_get_git_repo_root())
331 and "site-packages" not in p.parts
332 }