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