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/.
11"""Observability tools to spit out analysis-ready tables, one row per test case."""
13import base64
14import dataclasses
15import json
16import math
17import os
18import sys
19import threading
20import time
21import warnings
22from collections.abc import Callable, Generator
23from contextlib import contextmanager
24from dataclasses import dataclass
25from datetime import date, timedelta
26from functools import lru_cache
27from pathlib import Path
28from threading import Lock
29from typing import (
30 TYPE_CHECKING,
31 Any,
32 Literal,
33 Optional,
34 TypeAlias,
35 Union,
36 cast,
37)
39from hypothesis.configuration import storage_directory
40from hypothesis.errors import HypothesisWarning
41from hypothesis.internal.conjecture.choice import (
42 BooleanConstraints,
43 BytesConstraints,
44 ChoiceConstraintsT,
45 ChoiceNode,
46 ChoiceT,
47 ChoiceTypeT,
48 FloatConstraints,
49 IntegerConstraints,
50 StringConstraints,
51)
52from hypothesis.internal.escalation import InterestingOrigin
53from hypothesis.internal.floats import float_to_int
54from hypothesis.internal.intervalsets import IntervalSet
55from hypothesis.utils.deprecation import note_deprecation
57if TYPE_CHECKING:
58 from hypothesis.internal.conjecture.data import ConjectureData, Spans, Status
61Observation: TypeAlias = Union["InfoObservation", "TestCaseObservation"]
62CallbackThreadT: TypeAlias = Callable[[Observation], None]
63# for all_threads=True, we pass the thread id as well.
64CallbackAllThreadsT: TypeAlias = Callable[[Observation, int], None]
65CallbackT: TypeAlias = CallbackThreadT | CallbackAllThreadsT
67# thread_id: list[callback]
68_callbacks: dict[int | None, list[CallbackThreadT]] = {}
69# callbacks where all_threads=True was set
70_callbacks_all_threads: list[CallbackAllThreadsT] = []
73@dataclass(slots=True, frozen=False)
74class PredicateCounts:
75 satisfied: int = 0
76 unsatisfied: int = 0
78 def update_count(self, *, condition: bool) -> None:
79 if condition:
80 self.satisfied += 1
81 else:
82 self.unsatisfied += 1
85def _choice_to_json(choice: ChoiceT | None) -> Any:
86 if choice is None:
87 return None
88 # see the note on the same check in to_jsonable for why we cast large
89 # integers to floats.
90 if (
91 isinstance(choice, int)
92 and not isinstance(choice, bool)
93 and abs(choice) >= 2**63
94 ):
95 return ["integer", str(choice)]
96 elif isinstance(choice, bytes):
97 return ["bytes", base64.b64encode(choice).decode()]
98 elif isinstance(choice, float) and math.isnan(choice):
99 # handle nonstandard nan bit patterns. We don't need to do this for -0.0
100 # vs 0.0 since json doesn't normalize -0.0 to 0.0.
101 return ["float", float_to_int(choice)]
102 return choice
105def choices_to_json(choices: tuple[ChoiceT, ...]) -> list[Any]:
106 return [_choice_to_json(choice) for choice in choices]
109def _constraints_to_json(
110 choice_type: ChoiceTypeT, constraints: ChoiceConstraintsT
111) -> dict[str, Any]:
112 constraints = constraints.copy()
113 if choice_type == "integer":
114 constraints = cast(IntegerConstraints, constraints)
115 return {
116 "min_value": _choice_to_json(constraints["min_value"]),
117 "max_value": _choice_to_json(constraints["max_value"]),
118 "weights": (
119 None
120 if constraints["weights"] is None
121 # wrap up in a list, instead of a dict, because json dicts
122 # require string keys
123 else [
124 (_choice_to_json(k), v) for k, v in constraints["weights"].items()
125 ]
126 ),
127 "shrink_towards": _choice_to_json(constraints["shrink_towards"]),
128 }
129 elif choice_type == "float":
130 constraints = cast(FloatConstraints, constraints)
131 return {
132 "min_value": _choice_to_json(constraints["min_value"]),
133 "max_value": _choice_to_json(constraints["max_value"]),
134 "allow_nan": constraints["allow_nan"],
135 "smallest_nonzero_magnitude": constraints["smallest_nonzero_magnitude"],
136 }
137 elif choice_type == "string":
138 constraints = cast(StringConstraints, constraints)
139 assert isinstance(constraints["intervals"], IntervalSet)
140 return {
141 "intervals": constraints["intervals"].intervals,
142 "min_size": _choice_to_json(constraints["min_size"]),
143 "max_size": _choice_to_json(constraints["max_size"]),
144 }
145 elif choice_type == "bytes":
146 constraints = cast(BytesConstraints, constraints)
147 return {
148 "min_size": _choice_to_json(constraints["min_size"]),
149 "max_size": _choice_to_json(constraints["max_size"]),
150 }
151 elif choice_type == "boolean":
152 constraints = cast(BooleanConstraints, constraints)
153 return {
154 "p": constraints["p"],
155 }
156 else:
157 raise NotImplementedError(f"unknown choice type {choice_type}")
160def nodes_to_json(nodes: tuple[ChoiceNode, ...]) -> list[dict[str, Any]]:
161 return [
162 {
163 "type": node.type,
164 "value": _choice_to_json(node.value),
165 "constraints": _constraints_to_json(node.type, node.constraints),
166 "was_forced": node.was_forced,
167 }
168 for node in nodes
169 ]
172@dataclass(slots=True, frozen=True)
173class ObservationMetadata:
174 traceback: str | None
175 reproduction_decorator: str | None
176 notes: list[str]
177 predicates: dict[str, PredicateCounts]
178 backend: dict[str, Any]
179 sys_argv: list[str]
180 os_getpid: int
181 imported_at: float
182 data_status: "Status"
183 phase: str
184 interesting_origin: InterestingOrigin | None
185 status_reason_location: str | None
186 choice_nodes: tuple[ChoiceNode, ...] | None
187 choice_spans: Optional["Spans"]
189 def to_json(self) -> dict[str, Any]:
190 data = {
191 "traceback": self.traceback,
192 "reproduction_decorator": self.reproduction_decorator,
193 "notes": self.notes,
194 "predicates": self.predicates,
195 "backend": self.backend,
196 "sys.argv": self.sys_argv,
197 "os.getpid()": self.os_getpid,
198 "imported_at": self.imported_at,
199 "data_status": self.data_status,
200 "phase": self.phase,
201 "interesting_origin": self.interesting_origin,
202 "status_reason_location": self.status_reason_location,
203 "choice_nodes": (
204 None if self.choice_nodes is None else nodes_to_json(self.choice_nodes)
205 ),
206 "choice_spans": (
207 None
208 if self.choice_spans is None
209 else [
210 (
211 # span.label is an int, but cast to string to avoid conversion
212 # to float (and loss of precision) for large label values.
213 #
214 # The value of this label is opaque to consumers anyway, so its
215 # type shouldn't matter as long as it's consistent.
216 str(span.label),
217 span.start,
218 span.end,
219 span.discarded,
220 )
221 for span in self.choice_spans
222 ]
223 ),
224 }
225 # check that we didn't forget one
226 assert len(data) == len(dataclasses.fields(self))
227 return data
230@dataclass(slots=True, frozen=True)
231class BaseObservation:
232 type: Literal["test_case", "info", "alert", "error"]
233 property: str
234 run_start: float
237InfoObservationType = Literal["info", "alert", "error"]
238TestCaseStatus = Literal["gave_up", "passed", "failed"]
241@dataclass(slots=True, frozen=True)
242class InfoObservation(BaseObservation):
243 type: InfoObservationType
244 title: str
245 content: str | dict
248@dataclass(slots=True, frozen=True)
249class TestCaseObservation(BaseObservation):
250 __test__ = False # no! bad pytest!
252 type: Literal["test_case"]
253 status: TestCaseStatus
254 status_reason: str
255 representation: str
256 arguments: dict
257 how_generated: str
258 features: dict
259 coverage: dict[str, list[int]] | None
260 timing: dict[str, float]
261 metadata: ObservationMetadata
264def add_observability_callback(f: CallbackT, /, *, all_threads: bool = False) -> None:
265 """
266 Adds ``f`` as a callback for |observability|. ``f``
267 should accept one argument, which is an observation. Whenever Hypothesis
268 produces a new observation, it calls each callback with that observation.
270 If Hypothesis tests are being run from multiple threads, callbacks are tracked
271 per-thread. In other words, ``add_observability_callback(f)`` only adds ``f``
272 as an observability callback for observations produced on that thread.
274 If ``all_threads=True`` is passed, ``f`` will instead be registered as a
275 callback for all threads. This means it will be called for observations
276 generated by all threads, not just the thread which registered ``f`` as a
277 callback. In this case, ``f`` will be passed two arguments: the first is the
278 observation, and the second is the integer thread id from
279 :func:`python:threading.get_ident` where that observation was generated.
281 We recommend against registering ``f`` as a callback for both ``all_threads=True``
282 and the default ``all_threads=False``, due to unclear semantics with
283 |remove_observability_callback|.
284 """
285 if all_threads:
286 _callbacks_all_threads.append(cast(CallbackAllThreadsT, f))
287 return
289 thread_id = threading.get_ident()
290 if thread_id not in _callbacks:
291 _callbacks[thread_id] = []
293 _callbacks[thread_id].append(cast(CallbackThreadT, f))
296def remove_observability_callback(f: CallbackT, /) -> None:
297 """
298 Removes ``f`` from the |observability| callbacks.
300 If ``f`` is not in the list of observability callbacks, silently do nothing.
302 If running under multiple threads, ``f`` will only be removed from the
303 callbacks for this thread.
304 """
305 if f in _callbacks_all_threads:
306 _callbacks_all_threads.remove(f)
308 thread_id = threading.get_ident()
309 if thread_id not in _callbacks:
310 return
312 callbacks = _callbacks[thread_id]
313 if f in callbacks:
314 callbacks.remove(f)
316 if not callbacks:
317 del _callbacks[thread_id]
320def observability_enabled() -> bool:
321 """
322 Returns whether or not Hypothesis considers |observability|
323 to be enabled. Observability is enabled if there is at least one observability
324 callback present.
326 Callers might use this method to determine whether they should compute an
327 expensive representation that is only used under observability, for instance
328 by |alternative backends|.
329 """
330 return bool(_callbacks) or bool(_callbacks_all_threads)
333@contextmanager
334def with_observability_callback(
335 f: Callable[[Observation], None], /, *, all_threads: bool = False
336) -> Generator[None, None, None]:
337 """
338 A simple context manager which calls |add_observability_callback| on ``f``
339 when it enters and |remove_observability_callback| on ``f`` when it exits.
340 """
341 add_observability_callback(f, all_threads=all_threads)
342 try:
343 yield
344 finally:
345 remove_observability_callback(f)
348def deliver_observation(observation: Observation) -> None:
349 thread_id = threading.get_ident()
351 for callback in _callbacks.get(thread_id, []):
352 callback(observation)
354 for callback in _callbacks_all_threads:
355 callback(observation, thread_id)
358class _TestcaseCallbacks:
359 def __bool__(self):
360 self._note_deprecation()
361 return bool(_callbacks)
363 def _note_deprecation(self):
364 note_deprecation(
365 "hypothesis.internal.observability.TESTCASE_CALLBACKS is deprecated. "
366 "Replace TESTCASE_CALLBACKS.append with add_observability_callback, "
367 "TESTCASE_CALLBACKS.remove with remove_observability_callback, and "
368 "bool(TESTCASE_CALLBACKS) with observability_enabled().",
369 since="2025-08-01",
370 has_codemod=False,
371 )
373 def append(self, f):
374 self._note_deprecation()
375 add_observability_callback(f)
377 def remove(self, f):
378 self._note_deprecation()
379 remove_observability_callback(f)
382#: .. warning::
383#:
384#: Deprecated in favor of |add_observability_callback|,
385#: |remove_observability_callback|, and |observability_enabled|.
386#:
387#: |TESTCASE_CALLBACKS| remains a thin compatibility
388#: shim which forwards ``.append``, ``.remove``, and ``bool()`` to those
389#: three methods. It is not an attempt to be fully compatible with the previous
390#: ``TESTCASE_CALLBACKS = []``, so iteration or other usages will not work
391#: anymore. Please update to using the new methods instead.
392#:
393#: |TESTCASE_CALLBACKS| will eventually be removed.
394TESTCASE_CALLBACKS = _TestcaseCallbacks()
397def make_testcase(
398 *,
399 run_start: float,
400 property: str,
401 data: "ConjectureData",
402 how_generated: str,
403 representation: str = "<unknown>",
404 timing: dict[str, float],
405 arguments: dict | None = None,
406 coverage: dict[str, list[int]] | None = None,
407 phase: str | None = None,
408 backend_metadata: dict[str, Any] | None = None,
409 status: (
410 Union[TestCaseStatus, "Status"] | None
411 ) = None, # overrides automatic calculation
412 status_reason: str | None = None, # overrides automatic calculation
413 # added to calculated metadata. If keys overlap, the value from this `metadata`
414 # is used
415 metadata: dict[str, Any] | None = None,
416) -> TestCaseObservation:
417 from hypothesis.core import reproduction_decorator
418 from hypothesis.internal.conjecture.data import Status
420 # We should only be sending observability reports for datas that have finished
421 # being modified.
422 assert data.frozen
424 if status_reason is not None:
425 pass
426 elif data.interesting_origin:
427 status_reason = str(data.interesting_origin)
428 elif phase == "shrink" and data.status == Status.OVERRUN:
429 status_reason = "exceeded size of current best test case"
430 elif data.status == Status.OVERRUN:
431 status_reason = (
432 str(data.events.pop("gave up because", ""))
433 or "exceeded maximum test case size"
434 )
435 else:
436 status_reason = str(data.events.pop("gave up because", ""))
438 status_map: dict[Status, TestCaseStatus] = {
439 Status.OVERRUN: "gave_up",
440 Status.INVALID: "gave_up",
441 Status.VALID: "passed",
442 Status.INTERESTING: "failed",
443 }
445 if status is not None and isinstance(status, Status):
446 status = status_map[status]
447 if status is None:
448 status = status_map[data.status]
450 status_reason_location: str | None = None
451 if (origin := data.interesting_origin) is not None:
452 if origin.filename is not None:
453 status_reason_location = f"{origin.filename}:{origin.lineno}"
454 elif status != "failed":
455 status_reason_location = data.invalid_location
457 return TestCaseObservation(
458 type="test_case",
459 status=status,
460 status_reason=status_reason,
461 representation=representation,
462 arguments={
463 k.removeprefix("generate:"): v for k, v in (arguments or {}).items()
464 },
465 how_generated=how_generated, # iid, mutation, etc.
466 features={
467 **{
468 f"target:{k}".strip(":"): v for k, v in data.target_observations.items()
469 },
470 **data.events,
471 },
472 coverage=coverage,
473 timing=timing,
474 metadata=ObservationMetadata(
475 **{
476 "traceback": data.expected_traceback,
477 "reproduction_decorator": (
478 reproduction_decorator(data.choices) if status == "failed" else None
479 ),
480 "notes": list(data.notes),
481 "predicates": dict(data._observability_predicates),
482 "backend": backend_metadata or {},
483 "data_status": data.status,
484 "phase": phase,
485 "interesting_origin": data.interesting_origin,
486 "status_reason_location": status_reason_location,
487 "choice_nodes": data.nodes if OBSERVABILITY_CHOICES else None,
488 "choice_spans": data.spans if OBSERVABILITY_CHOICES else None,
489 **_system_metadata(),
490 # unpack last so it takes precedence for duplicate keys
491 **(metadata or {}),
492 }
493 ),
494 run_start=run_start,
495 property=property,
496 )
499_WROTE_TO: set[Path] = set()
500_deliver_to_file_lock = Lock()
503def _deliver_to_file(
504 observation: Observation, thread_id: int
505) -> None: # pragma: no cover
506 from hypothesis.strategies._internal.utils import to_jsonable
508 kind = "testcases" if observation.type == "test_case" else "info"
509 observed_dir = storage_directory("observed")
510 observed_dir.create_if_missing()
511 observation_p = observed_dir.path / f"{date.today().isoformat()}_{kind}.jsonl"
513 observation_bytes = (
514 json.dumps(to_jsonable(observation, avoid_realization=False)) + "\n"
515 )
516 # only allow one conccurent file write to avoid write races. This is likely to make
517 # HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY quite slow under threading. A queue
518 # would be an improvement, but that requires a background thread, and I
519 # would prefer to avoid a thread in the single-threaded case. We could
520 # switch over to a queue if we detect multithreading, but it's tricky to get
521 # right.
522 with _deliver_to_file_lock:
523 _WROTE_TO.add(observation_p)
524 with observation_p.open(mode="a") as f:
525 f.write(observation_bytes)
528_imported_at = time.time()
531@lru_cache
532def _system_metadata() -> dict[str, Any]:
533 return {
534 "sys_argv": sys.argv,
535 "os_getpid": os.getpid(),
536 "imported_at": _imported_at,
537 }
540#: If ``False``, do not collect coverage information when observability is enabled.
541#:
542#: This is exposed both for performance (as coverage collection can be slow on
543#: Python 3.11 and earlier) and size (if you do not use coverage information,
544#: you may not want to store it in-memory).
545OBSERVABILITY_COLLECT_COVERAGE = (
546 "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY_NOCOVER" not in os.environ
547)
548#: If ``True``, include the ``metadata.choice_nodes`` and ``metadata.spans`` keys
549#: in test case observations.
550#:
551#: ``False`` by default. ``metadata.choice_nodes`` and ``metadata.spans`` can be
552#: a substantial amount of data, and so must be opted-in to, even when
553#: observability is enabled.
554#:
555#: .. warning::
556#:
557#: EXPERIMENTAL AND UNSTABLE. We are actively working towards a better
558#: interface for this as of June 2025, and this attribute may disappear or
559#: be renamed without notice.
560#:
561OBSERVABILITY_CHOICES = "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY_CHOICES" in os.environ
563if OBSERVABILITY_COLLECT_COVERAGE is False and (
564 sys.version_info[:2] >= (3, 12)
565): # pragma: no cover
566 warnings.warn(
567 "Coverage data collection should be quite fast in Python 3.12 or later "
568 "so there should be no need to turn coverage reporting off.",
569 HypothesisWarning,
570 stacklevel=2,
571 )
573if (
574 "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY" in os.environ
575 or OBSERVABILITY_COLLECT_COVERAGE is False
576): # pragma: no cover
577 add_observability_callback(_deliver_to_file, all_threads=True)
579 # Remove files more than a week old, to cap the size on disk
580 max_age = (date.today() - timedelta(days=8)).isoformat()
581 for p in storage_directory("observed", intent_to_write=False).path.glob("*.jsonl"):
582 if p.stem < max_age: # pragma: no branch
583 p.unlink(missing_ok=True)