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 choice_nodes: tuple[ChoiceNode, ...] | None
186 choice_spans: Optional["Spans"]
188 def to_json(self) -> dict[str, Any]:
189 data = {
190 "traceback": self.traceback,
191 "reproduction_decorator": self.reproduction_decorator,
192 "notes": self.notes,
193 "predicates": self.predicates,
194 "backend": self.backend,
195 "sys.argv": self.sys_argv,
196 "os.getpid()": self.os_getpid,
197 "imported_at": self.imported_at,
198 "data_status": self.data_status,
199 "phase": self.phase,
200 "interesting_origin": self.interesting_origin,
201 "choice_nodes": (
202 None if self.choice_nodes is None else nodes_to_json(self.choice_nodes)
203 ),
204 "choice_spans": (
205 None
206 if self.choice_spans is None
207 else [
208 (
209 # span.label is an int, but cast to string to avoid conversion
210 # to float (and loss of precision) for large label values.
211 #
212 # The value of this label is opaque to consumers anyway, so its
213 # type shouldn't matter as long as it's consistent.
214 str(span.label),
215 span.start,
216 span.end,
217 span.discarded,
218 )
219 for span in self.choice_spans
220 ]
221 ),
222 }
223 # check that we didn't forget one
224 assert len(data) == len(dataclasses.fields(self))
225 return data
228@dataclass(slots=True, frozen=True)
229class BaseObservation:
230 type: Literal["test_case", "info", "alert", "error"]
231 property: str
232 run_start: float
235InfoObservationType = Literal["info", "alert", "error"]
236TestCaseStatus = Literal["gave_up", "passed", "failed"]
239@dataclass(slots=True, frozen=True)
240class InfoObservation(BaseObservation):
241 type: InfoObservationType
242 title: str
243 content: str | dict
246@dataclass(slots=True, frozen=True)
247class TestCaseObservation(BaseObservation):
248 __test__ = False # no! bad pytest!
250 type: Literal["test_case"]
251 status: TestCaseStatus
252 status_reason: str
253 representation: str
254 arguments: dict
255 how_generated: str
256 features: dict
257 coverage: dict[str, list[int]] | None
258 timing: dict[str, float]
259 metadata: ObservationMetadata
262def add_observability_callback(f: CallbackT, /, *, all_threads: bool = False) -> None:
263 """
264 Adds ``f`` as a callback for |observability|. ``f``
265 should accept one argument, which is an observation. Whenever Hypothesis
266 produces a new observation, it calls each callback with that observation.
268 If Hypothesis tests are being run from multiple threads, callbacks are tracked
269 per-thread. In other words, ``add_observability_callback(f)`` only adds ``f``
270 as an observability callback for observations produced on that thread.
272 If ``all_threads=True`` is passed, ``f`` will instead be registered as a
273 callback for all threads. This means it will be called for observations
274 generated by all threads, not just the thread which registered ``f`` as a
275 callback. In this case, ``f`` will be passed two arguments: the first is the
276 observation, and the second is the integer thread id from
277 :func:`python:threading.get_ident` where that observation was generated.
279 We recommend against registering ``f`` as a callback for both ``all_threads=True``
280 and the default ``all_threads=False``, due to unclear semantics with
281 |remove_observability_callback|.
282 """
283 if all_threads:
284 _callbacks_all_threads.append(cast(CallbackAllThreadsT, f))
285 return
287 thread_id = threading.get_ident()
288 if thread_id not in _callbacks:
289 _callbacks[thread_id] = []
291 _callbacks[thread_id].append(cast(CallbackThreadT, f))
294def remove_observability_callback(f: CallbackT, /) -> None:
295 """
296 Removes ``f`` from the |observability| callbacks.
298 If ``f`` is not in the list of observability callbacks, silently do nothing.
300 If running under multiple threads, ``f`` will only be removed from the
301 callbacks for this thread.
302 """
303 if f in _callbacks_all_threads:
304 _callbacks_all_threads.remove(f)
306 thread_id = threading.get_ident()
307 if thread_id not in _callbacks:
308 return
310 callbacks = _callbacks[thread_id]
311 if f in callbacks:
312 callbacks.remove(f)
314 if not callbacks:
315 del _callbacks[thread_id]
318def observability_enabled() -> bool:
319 """
320 Returns whether or not Hypothesis considers |observability|
321 to be enabled. Observability is enabled if there is at least one observability
322 callback present.
324 Callers might use this method to determine whether they should compute an
325 expensive representation that is only used under observability, for instance
326 by |alternative backends|.
327 """
328 return bool(_callbacks) or bool(_callbacks_all_threads)
331@contextmanager
332def with_observability_callback(
333 f: Callable[[Observation], None], /, *, all_threads: bool = False
334) -> Generator[None, None, None]:
335 """
336 A simple context manager which calls |add_observability_callback| on ``f``
337 when it enters and |remove_observability_callback| on ``f`` when it exits.
338 """
339 add_observability_callback(f, all_threads=all_threads)
340 try:
341 yield
342 finally:
343 remove_observability_callback(f)
346def deliver_observation(observation: Observation) -> None:
347 thread_id = threading.get_ident()
349 for callback in _callbacks.get(thread_id, []):
350 callback(observation)
352 for callback in _callbacks_all_threads:
353 callback(observation, thread_id)
356class _TestcaseCallbacks:
357 def __bool__(self):
358 self._note_deprecation()
359 return bool(_callbacks)
361 def _note_deprecation(self):
362 note_deprecation(
363 "hypothesis.internal.observability.TESTCASE_CALLBACKS is deprecated. "
364 "Replace TESTCASE_CALLBACKS.append with add_observability_callback, "
365 "TESTCASE_CALLBACKS.remove with remove_observability_callback, and "
366 "bool(TESTCASE_CALLBACKS) with observability_enabled().",
367 since="2025-08-01",
368 has_codemod=False,
369 )
371 def append(self, f):
372 self._note_deprecation()
373 add_observability_callback(f)
375 def remove(self, f):
376 self._note_deprecation()
377 remove_observability_callback(f)
380#: .. warning::
381#:
382#: Deprecated in favor of |add_observability_callback|,
383#: |remove_observability_callback|, and |observability_enabled|.
384#:
385#: |TESTCASE_CALLBACKS| remains a thin compatibility
386#: shim which forwards ``.append``, ``.remove``, and ``bool()`` to those
387#: three methods. It is not an attempt to be fully compatible with the previous
388#: ``TESTCASE_CALLBACKS = []``, so iteration or other usages will not work
389#: anymore. Please update to using the new methods instead.
390#:
391#: |TESTCASE_CALLBACKS| will eventually be removed.
392TESTCASE_CALLBACKS = _TestcaseCallbacks()
395def make_testcase(
396 *,
397 run_start: float,
398 property: str,
399 data: "ConjectureData",
400 how_generated: str,
401 representation: str = "<unknown>",
402 timing: dict[str, float],
403 arguments: dict | None = None,
404 coverage: dict[str, list[int]] | None = None,
405 phase: str | None = None,
406 backend_metadata: dict[str, Any] | None = None,
407 status: (
408 Union[TestCaseStatus, "Status"] | None
409 ) = None, # overrides automatic calculation
410 status_reason: str | None = None, # overrides automatic calculation
411 # added to calculated metadata. If keys overlap, the value from this `metadata`
412 # is used
413 metadata: dict[str, Any] | None = None,
414) -> TestCaseObservation:
415 from hypothesis.core import reproduction_decorator
416 from hypothesis.internal.conjecture.data import Status
418 # We should only be sending observability reports for datas that have finished
419 # being modified.
420 assert data.frozen
422 if status_reason is not None:
423 pass
424 elif data.interesting_origin:
425 status_reason = str(data.interesting_origin)
426 elif phase == "shrink" and data.status == Status.OVERRUN:
427 status_reason = "exceeded size of current best example"
428 else:
429 status_reason = str(data.events.pop("invalid because", ""))
431 status_map: dict[Status, TestCaseStatus] = {
432 Status.OVERRUN: "gave_up",
433 Status.INVALID: "gave_up",
434 Status.VALID: "passed",
435 Status.INTERESTING: "failed",
436 }
438 if status is not None and isinstance(status, Status):
439 status = status_map[status]
440 if status is None:
441 status = status_map[data.status]
443 return TestCaseObservation(
444 type="test_case",
445 status=status,
446 status_reason=status_reason,
447 representation=representation,
448 arguments={
449 k.removeprefix("generate:"): v for k, v in (arguments or {}).items()
450 },
451 how_generated=how_generated, # iid, mutation, etc.
452 features={
453 **{
454 f"target:{k}".strip(":"): v for k, v in data.target_observations.items()
455 },
456 **data.events,
457 },
458 coverage=coverage,
459 timing=timing,
460 metadata=ObservationMetadata(
461 **{
462 "traceback": data.expected_traceback,
463 "reproduction_decorator": (
464 reproduction_decorator(data.choices) if status == "failed" else None
465 ),
466 "notes": list(data.notes),
467 "predicates": dict(data._observability_predicates),
468 "backend": backend_metadata or {},
469 "data_status": data.status,
470 "phase": phase,
471 "interesting_origin": data.interesting_origin,
472 "choice_nodes": data.nodes if OBSERVABILITY_CHOICES else None,
473 "choice_spans": data.spans if OBSERVABILITY_CHOICES else None,
474 **_system_metadata(),
475 # unpack last so it takes precedence for duplicate keys
476 **(metadata or {}),
477 }
478 ),
479 run_start=run_start,
480 property=property,
481 )
484_WROTE_TO: set[Path] = set()
485_deliver_to_file_lock = Lock()
488def _deliver_to_file(
489 observation: Observation, thread_id: int
490) -> None: # pragma: no cover
491 from hypothesis.strategies._internal.utils import to_jsonable
493 kind = "testcases" if observation.type == "test_case" else "info"
494 observed_dir = storage_directory("observed")
495 observed_dir.create_if_missing()
496 observation_p = observed_dir.path / f"{date.today().isoformat()}_{kind}.jsonl"
498 observation_bytes = (
499 json.dumps(to_jsonable(observation, avoid_realization=False)) + "\n"
500 )
501 # only allow one conccurent file write to avoid write races. This is likely to make
502 # HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY quite slow under threading. A queue
503 # would be an improvement, but that requires a background thread, and I
504 # would prefer to avoid a thread in the single-threaded case. We could
505 # switch over to a queue if we detect multithreading, but it's tricky to get
506 # right.
507 with _deliver_to_file_lock:
508 _WROTE_TO.add(observation_p)
509 with observation_p.open(mode="a") as f:
510 f.write(observation_bytes)
513_imported_at = time.time()
516@lru_cache
517def _system_metadata() -> dict[str, Any]:
518 return {
519 "sys_argv": sys.argv,
520 "os_getpid": os.getpid(),
521 "imported_at": _imported_at,
522 }
525#: If ``False``, do not collect coverage information when observability is enabled.
526#:
527#: This is exposed both for performance (as coverage collection can be slow on
528#: Python 3.11 and earlier) and size (if you do not use coverage information,
529#: you may not want to store it in-memory).
530OBSERVABILITY_COLLECT_COVERAGE = (
531 "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY_NOCOVER" not in os.environ
532)
533#: If ``True``, include the ``metadata.choice_nodes`` and ``metadata.spans`` keys
534#: in test case observations.
535#:
536#: ``False`` by default. ``metadata.choice_nodes`` and ``metadata.spans`` can be
537#: a substantial amount of data, and so must be opted-in to, even when
538#: observability is enabled.
539#:
540#: .. warning::
541#:
542#: EXPERIMENTAL AND UNSTABLE. We are actively working towards a better
543#: interface for this as of June 2025, and this attribute may disappear or
544#: be renamed without notice.
545#:
546OBSERVABILITY_CHOICES = "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY_CHOICES" in os.environ
548if OBSERVABILITY_COLLECT_COVERAGE is False and (
549 sys.version_info[:2] >= (3, 12)
550): # pragma: no cover
551 warnings.warn(
552 "Coverage data collection should be quite fast in Python 3.12 or later "
553 "so there should be no need to turn coverage reporting off.",
554 HypothesisWarning,
555 stacklevel=2,
556 )
558if (
559 "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY" in os.environ
560 or OBSERVABILITY_COLLECT_COVERAGE is False
561): # pragma: no cover
562 add_observability_callback(_deliver_to_file, all_threads=True)
564 # Remove files more than a week old, to cap the size on disk
565 max_age = (date.today() - timedelta(days=8)).isoformat()
566 for p in storage_directory("observed", intent_to_write=False).path.glob("*.jsonl"):
567 if p.stem < max_age: # pragma: no branch
568 p.unlink(missing_ok=True)