Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/hypothesis/internal/scrutineer.py: 39%

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

138 statements  

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, TypeAlias 

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_extensions import Self 

32 

33Location: TypeAlias = tuple[str, int] 

34Branch: TypeAlias = tuple[Location | None, Location] 

35Trace: TypeAlias = frozenset[Branch] 

36 

37 

38@functools.cache 

39def should_trace_file(fname: str) -> bool: # pragma: no cover 

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 hasattr(sys, "monitoring"): # pragma: no branch # always true on Python >= 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 "_branches", 

59 "_previous_location", 

60 "_should_trace", 

61 "_tried_and_failed_to_trace", 

62 ) 

63 

64 def __init__(self, *, should_trace: bool) -> None: 

65 self._branches: set[Branch] = set() 

66 self._previous_location: Location | None = 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 if PYPY: 

73 return False 

74 if hasattr(sys, "monitoring"): 

75 return sys.monitoring.get_tool(MONITORING_TOOL_ID) is None 

76 return sys.gettrace() is None # pragma: no cover # only on Python < 3.12 

77 

78 @property 

79 def branches(self) -> Trace: 

80 return frozenset(self._branches) 

81 

82 def trace(self, frame, event, arg): # pragma: no cover # only on Python < 3.12 

83 try: 

84 if event == "call": 

85 return self.trace 

86 elif event == "line": 

87 fname = frame.f_code.co_filename 

88 if should_trace_file(fname): 

89 current_location = (fname, frame.f_lineno) 

90 self._branches.add((self._previous_location, current_location)) 

91 self._previous_location = current_location 

92 except RecursionError: 

93 pass 

94 

95 def trace_line( 

96 self, code: types.CodeType, line_number: int 

97 ) -> None: # pragma: no cover 

98 # sys.monitoring callbacks do not fire for other monitoring tools, so coverage.py 

99 # can't see this function. 

100 fname = code.co_filename 

101 if not should_trace_file(fname): 

102 # this function is only called on 3.12+, but we want to avoid an 

103 # assertion to that effect for performance. 

104 return sys.monitoring.DISABLE # type: ignore 

105 

106 current_location = (fname, line_number) 

107 self._branches.add((self._previous_location, current_location)) 

108 self._previous_location = current_location 

109 

110 def __enter__(self) -> "Self": 

111 self._tried_and_failed_to_trace = False 

112 

113 if not self._should_trace: 

114 return self 

115 

116 if not hasattr(sys, "monitoring"): # pragma: no cover # only on Python < 3.12 

117 sys.settrace(self.trace) 

118 return self 

119 

120 try: 

121 sys.monitoring.use_tool_id(MONITORING_TOOL_ID, "scrutineer") 

122 except ValueError: # pragma: no cover 

123 # another thread may have registered a tool for MONITORING_TOOL_ID 

124 # since we checked in can_trace; this is a rare race condition. 

125 self._tried_and_failed_to_trace = True 

126 return self 

127 

128 for event, callback_name in MONITORING_EVENTS.items(): 

129 sys.monitoring.set_events(MONITORING_TOOL_ID, event) 

130 callback = getattr(self, callback_name) 

131 sys.monitoring.register_callback(MONITORING_TOOL_ID, event, callback) 

132 

133 return self 

134 

135 def __exit__(self, *args, **kwargs): 

136 if not self._should_trace: 

137 return 

138 

139 if not hasattr(sys, "monitoring"): # pragma: no cover # only on Python < 3.12 

140 sys.settrace(None) 

141 return 

142 

143 if self._tried_and_failed_to_trace: 

144 return # pragma: no cover # only true after the race in __enter__ above 

145 

146 sys.monitoring.free_tool_id(MONITORING_TOOL_ID) 

147 for event in MONITORING_EVENTS: 

148 sys.monitoring.register_callback(MONITORING_TOOL_ID, event, None) 

149 

150 

151UNHELPFUL_LOCATIONS = ( 

152 # Note: The list is post-processed, so use plain "/" for separator here. 

153 # Quite rarely, the first always-failing line is in Pytest's internals. 

154 "/_pytest/**", 

155 "/pluggy/_*.py", 

156 # used by pytest for failure formatting in the terminal. 

157 # seen: pygments/lexer.py, pygments/formatters/, pygments/filter.py. 

158 "/pygments/*", 

159 "/conftest.py", 

160 # syrupy registers a pytest_assertrepr_compare hook, which only runs when 

161 # assertions fail — making it appear as always-failing-never-passing. 

162 "/syrupy/__init__.py", 

163) 

164 

165 

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 ) 

179 

180 

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 {} 

186 

187 unions = {origin: set().union(*values) for origin, values in traces.items()} 

188 seen_passing = {None}.union(*unions.pop(None, set())) 

189 

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 } 

196 

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" 

203 

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) 

219 

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: { 

226 loc 

227 for loc in afnp_locs 

228 if not filter_regex.search(loc[0]) 

229 # In addition to UNHELPFUL_LOCATIONS, we drop all stdlib locations. 

230 and ModuleLocation.from_path(loc[0]) is not ModuleLocation.STDLIB 

231 } 

232 for origin, afnp_locs in explanations.items() 

233 } 

234 

235 

236# see e.g. https://docs.python.org/3/library/sysconfig.html#posix-user 

237# for examples of these path schemes 

238def _stdlib_dirs(): 

239 return { 

240 Path(sysconfig.get_path("platstdlib")).resolve(), 

241 Path(sysconfig.get_path("stdlib")).resolve(), 

242 # Under Pyodide and other embedded pythons, the stdlib is imported from 

243 # a zipfile on sys.path, which sysconfig doesn't report. 

244 *(Path(p) for p in sys.path if p.endswith(".zip")), 

245 } 

246 

247 

248STDLIB_DIRS = _stdlib_dirs() 

249SITE_PACKAGES_DIRS = { 

250 Path(sysconfig.get_path("purelib")).resolve(), 

251 Path(sysconfig.get_path("platlib")).resolve(), 

252} 

253 

254EXPLANATION_STUB = ( 

255 "Explanation:", 

256 " These lines were always and only run by failing test cases:", 

257) 

258 

259 

260class ModuleLocation(IntEnum): 

261 LOCAL = 0 

262 SITE_PACKAGES = 1 

263 STDLIB = 2 

264 

265 @classmethod 

266 @lru_cache(1024) 

267 def from_path(cls, path: str) -> "ModuleLocation": 

268 path = Path(path).resolve() 

269 # site-packages may be a subdir of stdlib or platlib, so it's important to 

270 # check is_relative_to for this before the stdlib. 

271 if any(path.is_relative_to(p) for p in SITE_PACKAGES_DIRS): 

272 return cls.SITE_PACKAGES 

273 if any(path.is_relative_to(p) for p in STDLIB_DIRS): 

274 return cls.STDLIB 

275 return cls.LOCAL 

276 

277 

278# show local files first, then site-packages, then stdlib 

279def _sort_key(path: str, lineno: int) -> tuple[int, str, int]: 

280 return (ModuleLocation.from_path(path), path, lineno) 

281 

282 

283def make_report(explanations, *, cap_lines_at=5): 

284 report = defaultdict(list) 

285 for origin, locations in explanations.items(): 

286 locations = list(locations) 

287 locations.sort(key=lambda v: _sort_key(v[0], v[1])) 

288 report_lines = [f" {fname}:{lineno}" for fname, lineno in locations] 

289 if len(report_lines) > cap_lines_at + 1: 

290 msg = " (and {} more with settings.verbosity >= verbose)" 

291 report_lines[cap_lines_at:] = [msg.format(len(report_lines[cap_lines_at:]))] 

292 if report_lines: # We might have filtered out every location as uninformative. 

293 report[origin] = list(EXPLANATION_STUB) + report_lines 

294 return report 

295 

296 

297def explanatory_lines(traces, settings): 

298 if Phase.explain in settings.phases and sys.gettrace() and not traces: 

299 return defaultdict(list) 

300 # Return human-readable report lines summarising the traces 

301 explanations = get_explaining_locations(traces) 

302 max_lines = 5 if settings.verbosity <= Verbosity.normal else float("inf") 

303 return make_report(explanations, cap_lines_at=max_lines) 

304 

305 

306# beware the code below; we're using some heuristics to make a nicer report... 

307 

308 

309@functools.lru_cache 

310def _get_git_repo_root() -> Path: 

311 try: 

312 where = subprocess.run( 

313 ["git", "rev-parse", "--show-toplevel"], 

314 check=True, 

315 timeout=10, 

316 capture_output=True, 

317 text=True, 

318 encoding="utf-8", 

319 ).stdout.strip() 

320 except Exception: # pragma: no cover 

321 return Path().absolute().parents[-1] 

322 else: 

323 return Path(where) 

324 

325 

326def tractable_coverage_report(trace: Trace) -> dict[str, list[int]]: 

327 """Report a simple coverage map which is (probably most) of the user's code.""" 

328 coverage: dict = {} 

329 t = dict(trace) 

330 for file, line in set(t.keys()).union(t.values()) - {None}: # type: ignore 

331 coverage.setdefault(file, set()).add(line) 

332 stdlib_fragment = f"{os.sep}lib{os.sep}python3.{sys.version_info.minor}{os.sep}" 

333 return { 

334 k: sorted(v) 

335 for k, v in coverage.items() 

336 if stdlib_fragment not in k 

337 and (p := Path(k)).is_relative_to(_get_git_repo_root()) 

338 and "site-packages" not in p.parts 

339 }