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