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 math
12import time
13from collections import defaultdict
14from collections.abc import Hashable, Iterable, Iterator, Sequence
15from dataclasses import dataclass, field
16from enum import IntEnum
17from functools import cached_property
18from random import Random
19from typing import (
20 TYPE_CHECKING,
21 Any,
22 Literal,
23 NoReturn,
24 TypeAlias,
25 TypeVar,
26 cast,
27 overload,
28)
30from hypothesis.errors import (
31 CannotProceedScopeT,
32 ChoiceTooLarge,
33 FlakyStrategyDefinition,
34 Frozen,
35 InvalidArgument,
36 StopTest,
37)
38from hypothesis.internal.cache import LRUCache
39from hypothesis.internal.compat import add_note
40from hypothesis.internal.conjecture.choice import (
41 BooleanConstraints,
42 BytesConstraints,
43 ChoiceConstraintsT,
44 ChoiceNode,
45 ChoiceT,
46 ChoiceTemplate,
47 ChoiceTypeT,
48 FloatConstraints,
49 IntegerConstraints,
50 StringConstraints,
51 choice_constraints_key,
52 choice_from_index,
53 choice_permitted,
54 choices_size,
55)
56from hypothesis.internal.conjecture.junkdrawer import IntList, gc_cumulative_time
57from hypothesis.internal.conjecture.providers import (
58 COLLECTION_DEFAULT_MAX_SIZE,
59 HypothesisProvider,
60 PrimitiveProvider,
61)
62from hypothesis.internal.conjecture.utils import calc_label_from_name
63from hypothesis.internal.escalation import InterestingOrigin
64from hypothesis.internal.floats import (
65 SMALLEST_SUBNORMAL,
66 float_to_int,
67 int_to_float,
68 sign_aware_lte,
69)
70from hypothesis.internal.intervalsets import IntervalSet
71from hypothesis.internal.observability import PredicateCounts
72from hypothesis.reporting import debug_report
73from hypothesis.utils.conventions import not_set
74from hypothesis.utils.deprecation import note_deprecation
75from hypothesis.utils.threading import ThreadLocal
77if TYPE_CHECKING:
78 from hypothesis.strategies import SearchStrategy
79 from hypothesis.strategies._internal.core import DataObject
80 from hypothesis.strategies._internal.random import RandomState
81 from hypothesis.strategies._internal.strategies import Ex
84def __getattr__(name: str) -> Any:
85 if name == "AVAILABLE_PROVIDERS":
86 from hypothesis.internal.conjecture.providers import AVAILABLE_PROVIDERS
88 note_deprecation(
89 "hypothesis.internal.conjecture.data.AVAILABLE_PROVIDERS has been moved to "
90 "hypothesis.internal.conjecture.providers.AVAILABLE_PROVIDERS.",
91 since="2025-01-25",
92 has_codemod=False,
93 stacklevel=1,
94 )
95 return AVAILABLE_PROVIDERS
97 raise AttributeError(
98 f"Module 'hypothesis.internal.conjecture.data' has no attribute {name}"
99 )
102T = TypeVar("T")
103TargetObservations = dict[str, int | float]
104# index, choice_type, constraints, forced value
105MisalignedAt: TypeAlias = tuple[int, ChoiceTypeT, ChoiceConstraintsT, ChoiceT | None]
107TOP_LABEL = calc_label_from_name("top")
108MAX_DEPTH = 100
110threadlocal = ThreadLocal(global_test_counter=int)
113class Status(IntEnum):
114 OVERRUN = 0
115 INVALID = 1
116 VALID = 2
117 INTERESTING = 3
119 def __repr__(self) -> str:
120 return f"Status.{self.name}"
123@dataclass(slots=True, frozen=True)
124class StructuralCoverageTag:
125 label: int
128STRUCTURAL_COVERAGE_CACHE: dict[int, StructuralCoverageTag] = {}
131def structural_coverage(label: int) -> StructuralCoverageTag:
132 try:
133 return STRUCTURAL_COVERAGE_CACHE[label]
134 except KeyError:
135 return STRUCTURAL_COVERAGE_CACHE.setdefault(label, StructuralCoverageTag(label))
138# This cache can be quite hot and so we prefer LRUCache over LRUReusedCache for
139# performance. We lose scan resistance, but that's probably fine here.
140POOLED_CONSTRAINTS_CACHE: LRUCache[tuple[Any, ...], ChoiceConstraintsT] = LRUCache(4096)
143class Span:
144 """A span tracks the hierarchical structure of choices within a single test run.
146 Spans are created to mark regions of the choice sequence that are
147 logically related to each other. For instance, Hypothesis tracks:
148 - A single top-level span for the entire choice sequence
149 - A span for the choices made by each strategy
150 - Some strategies define additional spans within their choices. For instance,
151 st.lists() tracks the "should add another element" choice and the "add
152 another element" choices as separate spans.
154 Spans provide useful information to the shrinker, mutator, targeted PBT,
155 and other subsystems of Hypothesis.
157 Rather than store each ``Span`` as a rich object, it is actually
158 just an index into the ``Spans`` class defined below. This has two
159 purposes: Firstly, for most properties of spans we will never need
160 to allocate storage at all, because most properties are not used on
161 most spans. Secondly, by storing the spans as compact lists
162 of integers, we save a considerable amount of space compared to
163 Python's normal object size.
165 This does have the downside that it increases the amount of allocation
166 we do, and slows things down as a result, in some usage patterns because
167 we repeatedly allocate the same Span or int objects, but it will
168 often dramatically reduce our memory usage, so is worth it.
169 """
171 __slots__ = ("index", "owner")
173 def __init__(self, owner: "Spans", index: int) -> None:
174 self.owner = owner
175 self.index = index
177 def __eq__(self, other: object) -> bool:
178 if self is other:
179 return True
180 if not isinstance(other, Span):
181 return NotImplemented
182 return (self.owner is other.owner) and (self.index == other.index)
184 def __ne__(self, other: object) -> bool:
185 if self is other:
186 return False
187 if not isinstance(other, Span):
188 return NotImplemented
189 return (self.owner is not other.owner) or (self.index != other.index)
191 def __repr__(self) -> str:
192 return f"spans[{self.index}]"
194 @property
195 def label(self) -> int:
196 """A label is an opaque value that associates each span with its
197 approximate origin, such as a particular strategy class or a particular
198 kind of draw."""
199 return self.owner.labels[self.owner.label_indices[self.index]]
201 @property
202 def parent(self) -> int | None:
203 """The index of the span that this one is nested directly within."""
204 if self.index == 0:
205 return None
206 return self.owner.parentage[self.index]
208 @property
209 def start(self) -> int:
210 return self.owner.starts[self.index]
212 @property
213 def end(self) -> int:
214 return self.owner.ends[self.index]
216 @property
217 def depth(self) -> int:
218 """
219 Depth of this span in the span tree. The top-level span has a depth of 0.
220 """
221 return self.owner.depths[self.index]
223 @property
224 def discarded(self) -> bool:
225 """True if this is span's ``stop_span`` call had ``discard`` set to
226 ``True``. This means we believe that the shrinker should be able to delete
227 this span completely, without affecting the value produced by its enclosing
228 strategy. Typically set when a rejection sampler decides to reject a
229 generated value and try again."""
230 return self.index in self.owner.discarded
232 @property
233 def choice_count(self) -> int:
234 """The number of choices in this span."""
235 return self.end - self.start
237 @property
238 def children(self) -> "list[Span]":
239 """The list of all spans with this as a parent, in increasing index
240 order."""
241 return [self.owner[i] for i in self.owner.children[self.index]]
244class SpanProperty:
245 """There are many properties of spans that we calculate by
246 essentially rerunning the test case multiple times based on the
247 calls which we record in SpanProperty.
249 This class defines a visitor, subclasses of which can be used
250 to calculate these properties.
251 """
253 def __init__(self, spans: "Spans"):
254 self.span_stack: list[int] = []
255 self.spans = spans
256 self.span_count = 0
257 self.choice_count = 0
259 def run(self) -> Any:
260 """Rerun the test case with this visitor and return the
261 results of ``self.finish()``."""
262 for record in self.spans.trail:
263 if record == TrailType.STOP_SPAN_DISCARD:
264 self.__pop(discarded=True)
265 elif record == TrailType.STOP_SPAN_NO_DISCARD:
266 self.__pop(discarded=False)
267 elif record == TrailType.CHOICE:
268 self.choice_count += 1
269 else:
270 # everything after TrailType.CHOICE is the label of a span start.
271 self.__push(record - TrailType.CHOICE - 1)
273 return self.finish()
275 def __push(self, label_index: int) -> None:
276 i = self.span_count
277 assert i < len(self.spans)
278 self.start_span(i, label_index=label_index)
279 self.span_count += 1
280 self.span_stack.append(i)
282 def __pop(self, *, discarded: bool) -> None:
283 i = self.span_stack.pop()
284 self.stop_span(i, discarded=discarded)
286 def start_span(self, i: int, label_index: int) -> None:
287 """Called at the start of each span, with ``i`` the
288 index of the span and ``label_index`` the index of
289 its label in ``self.spans.labels``."""
291 def stop_span(self, i: int, *, discarded: bool) -> None:
292 """Called at the end of each span, with ``i`` the
293 index of the span and ``discarded`` being ``True`` if ``stop_span``
294 was called with ``discard=True``."""
296 def finish(self) -> Any:
297 raise NotImplementedError
300class TrailType(IntEnum):
301 STOP_SPAN_DISCARD = 1
302 STOP_SPAN_NO_DISCARD = 2
303 CHOICE = 3
304 # every trail element larger than TrailType.CHOICE is the label of a span
305 # start, offset by its index. So the first span label is stored as 4, the
306 # second as 5, etc, regardless of its actual integer label.
309class SpanRecord:
310 """Records the series of ``start_span``, ``stop_span``, and
311 ``draw_bits`` calls so that these may be stored in ``Spans`` and
312 replayed when we need to know about the structure of individual
313 ``Span`` objects.
315 Note that there is significant similarity between this class and
316 ``DataObserver``, and the plan is to eventually unify them, but
317 they currently have slightly different functions and implementations.
318 """
320 def __init__(self) -> None:
321 self.labels: list[int] = []
322 self.__index_of_labels: dict[int, int] | None = {}
323 self.trail = IntList()
324 self.nodes: list[ChoiceNode] = []
325 # The number of spans started so far, which is also the index that the
326 # next span to start will get. Spans are indexed in start order.
327 self.span_count = 0
329 def freeze(self) -> None:
330 self.__index_of_labels = None
332 def record_choice(self) -> None:
333 self.trail.append(TrailType.CHOICE)
335 def start_span(self, label: int) -> None:
336 assert self.__index_of_labels is not None
337 try:
338 i = self.__index_of_labels[label]
339 except KeyError:
340 i = self.__index_of_labels.setdefault(label, len(self.labels))
341 self.labels.append(label)
342 self.trail.append(TrailType.CHOICE + 1 + i)
343 self.span_count += 1
345 def stop_span(self, *, discard: bool) -> None:
346 if discard:
347 self.trail.append(TrailType.STOP_SPAN_DISCARD)
348 else:
349 self.trail.append(TrailType.STOP_SPAN_NO_DISCARD)
352class _starts_and_ends(SpanProperty):
353 def __init__(self, spans: "Spans") -> None:
354 super().__init__(spans)
355 self.starts = IntList.of_length(len(self.spans))
356 self.ends = IntList.of_length(len(self.spans))
358 def start_span(self, i: int, label_index: int) -> None:
359 self.starts[i] = self.choice_count
361 def stop_span(self, i: int, *, discarded: bool) -> None:
362 self.ends[i] = self.choice_count
364 def finish(self) -> tuple[IntList, IntList]:
365 return (self.starts, self.ends)
368class _discarded(SpanProperty):
369 def __init__(self, spans: "Spans") -> None:
370 super().__init__(spans)
371 self.result: set[int] = set()
373 def finish(self) -> frozenset[int]:
374 return frozenset(self.result)
376 def stop_span(self, i: int, *, discarded: bool) -> None:
377 if discarded:
378 self.result.add(i)
381class _parentage(SpanProperty):
382 def __init__(self, spans: "Spans") -> None:
383 super().__init__(spans)
384 self.result = IntList.of_length(len(self.spans))
386 def stop_span(self, i: int, *, discarded: bool) -> None:
387 if i > 0:
388 self.result[i] = self.span_stack[-1]
390 def finish(self) -> IntList:
391 return self.result
394class _depths(SpanProperty):
395 def __init__(self, spans: "Spans") -> None:
396 super().__init__(spans)
397 self.result = IntList.of_length(len(self.spans))
399 def start_span(self, i: int, label_index: int) -> None:
400 self.result[i] = len(self.span_stack)
402 def finish(self) -> IntList:
403 return self.result
406class _label_indices(SpanProperty):
407 def __init__(self, spans: "Spans") -> None:
408 super().__init__(spans)
409 self.result = IntList.of_length(len(self.spans))
411 def start_span(self, i: int, label_index: int) -> None:
412 self.result[i] = label_index
414 def finish(self) -> IntList:
415 return self.result
418class _mutator_groups(SpanProperty):
419 def __init__(self, spans: "Spans") -> None:
420 super().__init__(spans)
421 self.groups: dict[int, set[tuple[int, int]]] = defaultdict(set)
423 def start_span(self, i: int, label_index: int) -> None:
424 # TODO should we discard start == end cases? occurs for eg st.data()
425 # which is conditionally or never drawn from. arguably swapping
426 # nodes with the empty list is a useful mutation enabled by start == end?
427 key = (self.spans[i].start, self.spans[i].end)
428 self.groups[label_index].add(key)
430 def finish(self) -> Iterable[set[tuple[int, int]]]:
431 # Discard groups with only one span, since the mutator can't
432 # do anything useful with them.
433 return [g for g in self.groups.values() if len(g) >= 2]
436class Spans:
437 """A lazy collection of ``Span`` objects, derived from
438 the record of recorded behaviour in ``SpanRecord``.
440 Behaves logically as if it were a list of ``Span`` objects,
441 but actually mostly exists as a compact store of information
442 for them to reference into. All properties on here are best
443 understood as the backing storage for ``Span`` and are
444 described there.
445 """
447 def __init__(self, record: SpanRecord) -> None:
448 self.trail = record.trail
449 self.labels = record.labels
450 self.__length = self.trail.count(
451 TrailType.STOP_SPAN_DISCARD
452 ) + record.trail.count(TrailType.STOP_SPAN_NO_DISCARD)
453 self.__children: list[Sequence[int]] | None = None
455 @cached_property
456 def starts_and_ends(self) -> tuple[IntList, IntList]:
457 return _starts_and_ends(self).run()
459 @property
460 def starts(self) -> IntList:
461 return self.starts_and_ends[0]
463 @property
464 def ends(self) -> IntList:
465 return self.starts_and_ends[1]
467 @cached_property
468 def discarded(self) -> frozenset[int]:
469 return _discarded(self).run()
471 @cached_property
472 def parentage(self) -> IntList:
473 return _parentage(self).run()
475 @cached_property
476 def depths(self) -> IntList:
477 return _depths(self).run()
479 @cached_property
480 def label_indices(self) -> IntList:
481 return _label_indices(self).run()
483 @cached_property
484 def mutator_groups(self) -> list[set[tuple[int, int]]]:
485 return _mutator_groups(self).run()
487 @property
488 def children(self) -> list[Sequence[int]]:
489 if self.__children is None:
490 children = [IntList() for _ in range(len(self))]
491 for i, p in enumerate(self.parentage):
492 if i > 0:
493 children[p].append(i)
494 # Replace empty children lists with a tuple to reduce
495 # memory usage.
496 for i, c in enumerate(children):
497 if not c:
498 children[i] = () # type: ignore
499 self.__children = children # type: ignore
500 return self.__children # type: ignore
502 def __len__(self) -> int:
503 return self.__length
505 def __getitem__(self, i: int) -> Span:
506 n = self.__length
507 if i < -n or i >= n:
508 raise IndexError(f"Index {i} out of range [-{n}, {n})")
509 if i < 0:
510 i += n
511 return Span(self, i)
513 # not strictly necessary as we have len/getitem, but required for mypy.
514 # https://github.com/python/mypy/issues/9737
515 def __iter__(self) -> Iterator[Span]:
516 for i in range(len(self)):
517 yield self[i]
520class _Overrun:
521 status: Status = Status.OVERRUN
523 def __repr__(self) -> str:
524 return "Overrun"
527Overrun = _Overrun()
530class DataObserver:
531 """Observer class for recording the behaviour of a
532 ConjectureData object, primarily used for tracking
533 the behaviour in the tree cache."""
535 def conclude_test(
536 self,
537 status: Status,
538 interesting_origin: InterestingOrigin | None,
539 ) -> None:
540 """Called when ``conclude_test`` is called on the
541 observed ``ConjectureData``, with the same arguments.
543 Note that this is called after ``freeze`` has completed.
544 """
546 def kill_branch(self) -> None:
547 """Mark this part of the tree as not worth re-exploring."""
549 def draw_integer(
550 self, value: int, *, constraints: IntegerConstraints, was_forced: bool
551 ) -> None:
552 pass
554 def draw_float(
555 self, value: float, *, constraints: FloatConstraints, was_forced: bool
556 ) -> None:
557 pass
559 def draw_string(
560 self, value: str, *, constraints: StringConstraints, was_forced: bool
561 ) -> None:
562 pass
564 def draw_bytes(
565 self, value: bytes, *, constraints: BytesConstraints, was_forced: bool
566 ) -> None:
567 pass
569 def draw_boolean(
570 self, value: bool, *, constraints: BooleanConstraints, was_forced: bool
571 ) -> None:
572 pass
575@dataclass(slots=True, frozen=True)
576class ConjectureResult:
577 """Result class storing the parts of ConjectureData that we
578 will care about after the original ConjectureData has outlived its
579 usefulness."""
581 status: Status
582 interesting_origin: InterestingOrigin | None
583 nodes: tuple[ChoiceNode, ...] = field(repr=False, compare=False)
584 length: int
585 notes: list[str]
586 expected_exception: BaseException | None
587 expected_traceback: str | None
588 has_discards: bool
589 target_observations: TargetObservations
590 tags: frozenset[StructuralCoverageTag]
591 spans: Spans = field(repr=False, compare=False)
592 arg_spans: set[int] = field(repr=False)
593 # Comments for the explain phase, keyed by span index. The ``None`` key
594 # holds the whole-test comment about varying all commented parts together.
595 span_comments: dict[int | None, str] = field(repr=False)
596 misaligned_at: MisalignedAt | None = field(repr=False)
597 cannot_proceed_scope: CannotProceedScopeT | None = field(repr=False)
599 def as_result(self) -> "ConjectureResult":
600 return self
602 @property
603 def choices(self) -> tuple[ChoiceT, ...]:
604 return tuple(node.value for node in self.nodes)
607class ConjectureData:
608 @classmethod
609 def for_choices(
610 cls,
611 choices: Sequence[ChoiceTemplate | ChoiceT],
612 *,
613 observer: DataObserver | None = None,
614 provider: PrimitiveProvider | type[PrimitiveProvider] = HypothesisProvider,
615 random: Random | None = None,
616 ) -> "ConjectureData":
617 from hypothesis.internal.conjecture.engine import choice_count
619 return cls(
620 max_choices=choice_count(choices),
621 random=random,
622 prefix=choices,
623 observer=observer,
624 provider=provider,
625 )
627 def __init__(
628 self,
629 *,
630 random: Random | None,
631 observer: DataObserver | None = None,
632 provider: PrimitiveProvider | type[PrimitiveProvider] = HypothesisProvider,
633 prefix: Sequence[ChoiceTemplate | ChoiceT] | None = None,
634 max_choices: int | None = None,
635 provider_kw: dict[str, Any] | None = None,
636 ) -> None:
637 from hypothesis.internal.conjecture.engine import BUFFER_SIZE
639 if observer is None:
640 observer = DataObserver()
641 if provider_kw is None:
642 provider_kw = {}
643 elif not isinstance(provider, type):
644 raise InvalidArgument(
645 f"Expected {provider=} to be a class since {provider_kw=} was "
646 "passed, but got an instance instead."
647 )
649 assert isinstance(observer, DataObserver)
650 self.observer = observer
651 self.max_choices = max_choices
652 self.max_length = BUFFER_SIZE
653 self.overdraw = 0
654 self._random = random
656 self.length: int = 0
657 self.index: int = 0
658 self.notes: list[str] = []
659 self.status: Status = Status.VALID
660 self.frozen: bool = False
661 self.testcounter: int = threadlocal.global_test_counter
662 threadlocal.global_test_counter += 1
663 self.start_time = time.perf_counter()
664 self.gc_start_time = gc_cumulative_time()
665 self.events: dict[str, str | int | float] = {}
666 self.interesting_origin: InterestingOrigin | None = None
667 self.draw_times: dict[str, float] = {}
668 self._stateful_run_times: dict[str, float] = defaultdict(float)
669 self.max_depth: int = 0
670 self.has_discards: bool = False
672 self.provider: PrimitiveProvider = (
673 provider(self, **provider_kw) if isinstance(provider, type) else provider
674 )
675 assert isinstance(self.provider, PrimitiveProvider)
677 self.__result: ConjectureResult | None = None
679 # Observations used for targeted search. They'll be aggregated in
680 # ConjectureRunner.generate_new_examples and fed to TargetSelector.
681 self.target_observations: TargetObservations = {}
683 # Tags which indicate something about which part of the search space
684 # this example is in. These are used to guide generation.
685 self.tags: set[StructuralCoverageTag] = set()
686 self.labels_for_structure_stack: list[set[int]] = []
688 # Normally unpopulated but we need this in the niche case
689 # that self.as_result() is Overrun but we still want the
690 # examples for reporting purposes.
691 self.__spans: Spans | None = None
693 # We want the top level span to have depth 0, so we start at -1.
694 self.depth: int = -1
695 self.__span_record = SpanRecord()
697 # Span indices for discrete reportable parts that which-parts-matter can
698 # try varying, to report if the minimal example always fails anyway.
699 self.arg_spans: set[int] = set()
700 self.span_comments: dict[int | None, str] = {}
701 self._observability_args: dict[str, Any] = {}
702 self._observability_predicates: defaultdict[str, PredicateCounts] = defaultdict(
703 PredicateCounts
704 )
706 self._sampled_from_all_strategies_elements_message: (
707 tuple[str, object] | None
708 ) = None
709 self._shared_strategy_draws: dict[Hashable, tuple[Any, SearchStrategy]] = {}
710 self._shared_data_strategy: DataObject | None = None
711 self._stateful_repr_parts: list[Any] | None = None
712 self.states_for_ids: dict[int, RandomState] | None = None
713 self.seeds_to_states: dict[Any, RandomState] | None = None
714 self.hypothesis_runner: Any = not_set
716 self.expected_exception: BaseException | None = None
717 self.expected_traceback: str | None = None
719 self.prefix = prefix
720 self.nodes: tuple[ChoiceNode, ...] = ()
721 self.misaligned_at: MisalignedAt | None = None
722 self.cannot_proceed_scope: CannotProceedScopeT | None = None
723 self.start_span(TOP_LABEL)
725 def __repr__(self) -> str:
726 return (
727 f"ConjectureData({self.status.name}, {len(self.nodes)} "
728 f"choices{', frozen' if self.frozen else ''})"
729 )
731 @property
732 def choices(self) -> tuple[ChoiceT, ...]:
733 return tuple(node.value for node in self.nodes)
735 # draw_* functions might be called in one of two contexts: either "above" or
736 # "below" the choice sequence. For instance, draw_string calls draw_boolean
737 # from ``many`` when calculating the number of characters to return. We do
738 # not want these choices to get written to the choice sequence, because they
739 # are not true choices themselves.
740 #
741 # `observe` formalizes this. The choice will only be written to the choice
742 # sequence if observe is True.
744 @overload
745 def _draw(
746 self,
747 choice_type: Literal["integer"],
748 constraints: IntegerConstraints,
749 *,
750 observe: bool,
751 forced: int | None,
752 ) -> int: ...
754 @overload
755 def _draw(
756 self,
757 choice_type: Literal["float"],
758 constraints: FloatConstraints,
759 *,
760 observe: bool,
761 forced: float | None,
762 ) -> float: ...
764 @overload
765 def _draw(
766 self,
767 choice_type: Literal["string"],
768 constraints: StringConstraints,
769 *,
770 observe: bool,
771 forced: str | None,
772 ) -> str: ...
774 @overload
775 def _draw(
776 self,
777 choice_type: Literal["bytes"],
778 constraints: BytesConstraints,
779 *,
780 observe: bool,
781 forced: bytes | None,
782 ) -> bytes: ...
784 @overload
785 def _draw(
786 self,
787 choice_type: Literal["boolean"],
788 constraints: BooleanConstraints,
789 *,
790 observe: bool,
791 forced: bool | None,
792 ) -> bool: ...
794 def _draw(
795 self,
796 choice_type: ChoiceTypeT,
797 constraints: ChoiceConstraintsT,
798 *,
799 observe: bool,
800 forced: ChoiceT | None,
801 ) -> ChoiceT:
802 # this is somewhat redundant with the length > max_length check at the
803 # end of the function, but avoids trying to use a null self.random when
804 # drawing past the node of a ConjectureData.for_choices data.
805 if self.length == self.max_length:
806 debug_report(f"overrun because hit {self.max_length=}")
807 self.mark_overrun()
808 if len(self.nodes) == self.max_choices:
809 debug_report(f"overrun because hit {self.max_choices=}")
810 self.mark_overrun()
812 if observe and self.prefix is not None and self.index < len(self.prefix):
813 value = self._pop_choice(choice_type, constraints, forced=forced)
814 elif forced is None:
815 value = getattr(self.provider, f"draw_{choice_type}")(**constraints)
817 if forced is not None:
818 value = forced
820 # nan values generated via int_to_float break list membership:
821 #
822 # >>> n = 18444492273895866368
823 # >>> assert math.isnan(int_to_float(n))
824 # >>> assert int_to_float(n) not in [int_to_float(n)]
825 #
826 # because int_to_float nans are not equal in the sense of either
827 # `a == b` or `a is b`.
828 #
829 # This can lead to flaky errors when collections require unique
830 # floats. What was happening is that in some places we provided math.nan
831 # provide math.nan, and in others we provided
832 # int_to_float(float_to_int(math.nan)), and which one gets used
833 # was not deterministic across test iterations.
834 #
835 # To fix this, *never* provide a nan value which is equal (via `is`) to
836 # another provided nan value. This sacrifices some test power; we should
837 # bring that back (ABOVE the choice sequence layer) in the future.
838 #
839 # See https://github.com/HypothesisWorks/hypothesis/issues/3926.
840 if choice_type == "float":
841 assert isinstance(value, float)
842 if math.isnan(value):
843 value = int_to_float(float_to_int(value))
845 if observe:
846 was_forced = forced is not None
847 getattr(self.observer, f"draw_{choice_type}")(
848 value, constraints=constraints, was_forced=was_forced
849 )
850 size = 0 if self.provider.avoid_realization else choices_size([value])
851 if self.length + size > self.max_length:
852 debug_report(
853 f"overrun because {self.length=} + {size=} > {self.max_length=}"
854 )
855 self.mark_overrun()
857 node = ChoiceNode(
858 type=choice_type,
859 value=value,
860 constraints=constraints,
861 was_forced=was_forced,
862 index=len(self.nodes),
863 )
864 self.__span_record.record_choice()
865 self.nodes += (node,)
866 self.length += size
868 return value
870 def draw_integer(
871 self,
872 min_value: int | None = None,
873 max_value: int | None = None,
874 *,
875 weights: dict[int, float] | None = None,
876 shrink_towards: int = 0,
877 forced: int | None = None,
878 observe: bool = True,
879 ) -> int:
880 # Validate arguments
881 if weights is not None:
882 assert min_value is not None
883 assert max_value is not None
884 assert len(weights) <= 255 # arbitrary practical limit
885 # We can and should eventually support total weights. But this
886 # complicates shrinking as we can no longer assume we can force
887 # a value to the unmapped probability mass if that mass might be 0.
888 assert sum(weights.values()) < 1
889 # similarly, things get simpler if we assume every value is possible.
890 # we'll want to drop this restriction eventually.
891 assert all(w != 0 for w in weights.values())
893 if forced is not None and min_value is not None:
894 assert min_value <= forced
895 if forced is not None and max_value is not None:
896 assert forced <= max_value
898 constraints: IntegerConstraints = self._pooled_constraints(
899 "integer",
900 {
901 "min_value": min_value,
902 "max_value": max_value,
903 "weights": weights,
904 "shrink_towards": shrink_towards,
905 },
906 )
907 return self._draw("integer", constraints, observe=observe, forced=forced)
909 def draw_float(
910 self,
911 min_value: float = -math.inf,
912 max_value: float = math.inf,
913 *,
914 allow_nan: bool = True,
915 smallest_nonzero_magnitude: float = SMALLEST_SUBNORMAL,
916 # TODO: consider supporting these float widths at the choice sequence
917 # level in the future.
918 # width: Literal[16, 32, 64] = 64,
919 forced: float | None = None,
920 observe: bool = True,
921 ) -> float:
922 assert smallest_nonzero_magnitude > 0
923 assert not math.isnan(min_value)
924 assert not math.isnan(max_value)
926 if smallest_nonzero_magnitude == 0.0: # pragma: no cover
927 raise FloatingPointError(
928 "Got allow_subnormal=True, but we can't represent subnormal floats "
929 "right now, in violation of the IEEE-754 floating-point "
930 "specification. This is usually because something was compiled with "
931 "-ffast-math or a similar option, which sets global processor state. "
932 "See https://simonbyrne.github.io/notes/fastmath/ for a more detailed "
933 "writeup - and good luck!"
934 )
936 if forced is not None:
937 assert allow_nan or not math.isnan(forced)
938 assert math.isnan(forced) or (
939 sign_aware_lte(min_value, forced) and sign_aware_lte(forced, max_value)
940 )
942 constraints: FloatConstraints = self._pooled_constraints(
943 "float",
944 {
945 "min_value": min_value,
946 "max_value": max_value,
947 "allow_nan": allow_nan,
948 "smallest_nonzero_magnitude": smallest_nonzero_magnitude,
949 },
950 )
951 return self._draw("float", constraints, observe=observe, forced=forced)
953 def draw_string(
954 self,
955 intervals: IntervalSet,
956 *,
957 min_size: int = 0,
958 max_size: int = COLLECTION_DEFAULT_MAX_SIZE,
959 forced: str | None = None,
960 observe: bool = True,
961 ) -> str:
962 assert forced is None or min_size <= len(forced) <= max_size
963 assert min_size >= 0
964 if len(intervals) == 0:
965 assert min_size == 0
967 constraints: StringConstraints = self._pooled_constraints(
968 "string",
969 {
970 "intervals": intervals,
971 "min_size": min_size,
972 "max_size": max_size,
973 },
974 )
975 return self._draw("string", constraints, observe=observe, forced=forced)
977 def draw_bytes(
978 self,
979 min_size: int = 0,
980 max_size: int = COLLECTION_DEFAULT_MAX_SIZE,
981 *,
982 forced: bytes | None = None,
983 observe: bool = True,
984 ) -> bytes:
985 assert forced is None or min_size <= len(forced) <= max_size
986 assert min_size >= 0
988 constraints: BytesConstraints = self._pooled_constraints(
989 "bytes", {"min_size": min_size, "max_size": max_size}
990 )
991 return self._draw("bytes", constraints, observe=observe, forced=forced)
993 def draw_boolean(
994 self,
995 p: float = 0.5,
996 *,
997 forced: bool | None = None,
998 observe: bool = True,
999 ) -> bool:
1000 assert (forced is not True) or p > 0
1001 assert (forced is not False) or p < 1
1003 constraints: BooleanConstraints = self._pooled_constraints("boolean", {"p": p})
1004 return self._draw("boolean", constraints, observe=observe, forced=forced)
1006 @overload
1007 def _pooled_constraints(
1008 self, choice_type: Literal["integer"], constraints: IntegerConstraints
1009 ) -> IntegerConstraints: ...
1011 @overload
1012 def _pooled_constraints(
1013 self, choice_type: Literal["float"], constraints: FloatConstraints
1014 ) -> FloatConstraints: ...
1016 @overload
1017 def _pooled_constraints(
1018 self, choice_type: Literal["string"], constraints: StringConstraints
1019 ) -> StringConstraints: ...
1021 @overload
1022 def _pooled_constraints(
1023 self, choice_type: Literal["bytes"], constraints: BytesConstraints
1024 ) -> BytesConstraints: ...
1026 @overload
1027 def _pooled_constraints(
1028 self, choice_type: Literal["boolean"], constraints: BooleanConstraints
1029 ) -> BooleanConstraints: ...
1031 def _pooled_constraints(
1032 self, choice_type: ChoiceTypeT, constraints: ChoiceConstraintsT
1033 ) -> ChoiceConstraintsT:
1034 """Memoize common dictionary objects to reduce memory pressure."""
1035 # caching runs afoul of nondeterminism checks
1036 if self.provider.avoid_realization:
1037 return constraints
1039 key = (choice_type, *choice_constraints_key(choice_type, constraints))
1040 try:
1041 return POOLED_CONSTRAINTS_CACHE[key]
1042 except KeyError:
1043 POOLED_CONSTRAINTS_CACHE[key] = constraints
1044 return constraints
1046 def _pop_choice(
1047 self,
1048 choice_type: ChoiceTypeT,
1049 constraints: ChoiceConstraintsT,
1050 *,
1051 forced: ChoiceT | None,
1052 ) -> ChoiceT:
1053 assert self.prefix is not None
1054 # checked in _draw
1055 assert self.index < len(self.prefix)
1057 value = self.prefix[self.index]
1058 if isinstance(value, ChoiceTemplate):
1059 node: ChoiceTemplate = value
1060 if node.count is not None:
1061 assert node.count >= 0
1062 # node templates have to be at the end for now, since it's not immediately
1063 # apparent how to handle overruning a node template while generating a single
1064 # node if the alternative is not "the entire data is an overrun".
1065 assert self.index == len(self.prefix) - 1
1066 if node.type == "simplest":
1067 if forced is not None:
1068 choice = forced
1069 try:
1070 choice = choice_from_index(0, choice_type, constraints)
1071 except ChoiceTooLarge:
1072 self.mark_overrun()
1073 else:
1074 raise NotImplementedError
1076 if node.count is not None:
1077 node.count -= 1
1078 if node.count < 0:
1079 self.mark_overrun()
1080 return choice
1082 choice = value
1083 node_choice_type = {
1084 str: "string",
1085 float: "float",
1086 int: "integer",
1087 bool: "boolean",
1088 bytes: "bytes",
1089 }[type(choice)]
1090 # If we're trying to:
1091 # * draw a different choice type at the same location
1092 # * draw the same choice type with a different constraints, which does not permit
1093 # the current value
1094 #
1095 # then we call this a misalignment, because the choice sequence has
1096 # changed from what we expected at some point. An easy misalignment is
1097 #
1098 # one_of(integers(0, 100), integers(101, 200))
1099 #
1100 # where the choice sequence [0, 100] has constraints {min_value: 0, max_value: 100}
1101 # at index 1, but [0, 101] has constraints {min_value: 101, max_value: 200} at
1102 # index 1 (which does not permit any of the values 0-100).
1103 #
1104 # When the choice sequence becomes misaligned, we generate a new value of the
1105 # type and constraints the strategy expects.
1106 if node_choice_type != choice_type or not choice_permitted(choice, constraints):
1107 # only track first misalignment for now.
1108 if self.misaligned_at is None:
1109 self.misaligned_at = (self.index, choice_type, constraints, forced)
1110 try:
1111 # Fill in any misalignments with index 0 choices. An alternative to
1112 # this is using the index of the misaligned choice instead
1113 # of index 0, which may be useful for maintaining
1114 # "similarly-complex choices" in the shrinker. This requires
1115 # attaching an index to every choice in ConjectureData.for_choices,
1116 # which we don't always have (e.g. when reading from db).
1117 #
1118 # If we really wanted this in the future we could make this complexity
1119 # optional, use it if present, and default to index 0 otherwise.
1120 # This complicates our internal api and so I'd like to avoid it
1121 # if possible.
1122 #
1123 # Additionally, I don't think slips which require
1124 # slipping to high-complexity values are common. Though arguably
1125 # we may want to expand a bit beyond *just* the simplest choice.
1126 # (we could for example consider sampling choices from index 0-10).
1127 choice = choice_from_index(0, choice_type, constraints)
1128 except ChoiceTooLarge:
1129 # should really never happen with a 0-index choice, but let's be safe.
1130 self.mark_overrun()
1132 self.index += 1
1133 return choice
1135 def as_result(self) -> ConjectureResult | _Overrun:
1136 """Convert the result of running this test into
1137 either an Overrun object or a ConjectureResult."""
1139 assert self.frozen
1140 if self.status == Status.OVERRUN:
1141 return Overrun
1142 if self.__result is None:
1143 self.__result = ConjectureResult(
1144 status=self.status,
1145 interesting_origin=self.interesting_origin,
1146 spans=self.spans,
1147 nodes=self.nodes,
1148 length=self.length,
1149 notes=self.notes,
1150 expected_traceback=self.expected_traceback,
1151 expected_exception=self.expected_exception,
1152 has_discards=self.has_discards,
1153 target_observations=self.target_observations,
1154 tags=frozenset(self.tags),
1155 arg_spans=self.arg_spans,
1156 span_comments=self.span_comments,
1157 misaligned_at=self.misaligned_at,
1158 cannot_proceed_scope=self.cannot_proceed_scope,
1159 )
1160 assert self.__result is not None
1161 return self.__result
1163 def __assert_not_frozen(self, name: str) -> None:
1164 if self.frozen:
1165 raise Frozen(f"Cannot call {name} on frozen ConjectureData")
1167 def note(self, value: str) -> None:
1168 self.__assert_not_frozen("note")
1169 self.notes.append(value)
1171 def draw(
1172 self,
1173 strategy: "SearchStrategy[Ex]",
1174 label: int | None = None,
1175 observe_as: str | None = None,
1176 ) -> "Ex":
1177 from hypothesis.internal.observability import observability_enabled
1178 from hypothesis.strategies._internal.lazy import unwrap_strategies
1179 from hypothesis.strategies._internal.utils import to_jsonable
1181 at_top_level = self.depth == 0
1182 start_time = None
1183 if at_top_level:
1184 # We start this timer early, because accessing attributes on a LazyStrategy
1185 # can be almost arbitrarily slow. In cases like characters() and text()
1186 # where we cache something expensive, this led to Flaky deadline errors!
1187 # See https://github.com/HypothesisWorks/hypothesis/issues/2108
1188 start_time = time.perf_counter()
1189 gc_start_time = gc_cumulative_time()
1191 strategy.validate()
1193 if strategy.is_empty:
1194 self.mark_invalid(f"empty strategy {self!r}")
1196 if self.depth >= MAX_DEPTH:
1197 self.mark_invalid("max depth exceeded")
1199 # Jump directly to the unwrapped strategy for the label and for do_draw.
1200 # This avoids adding an extra span to all lazy strategies.
1201 unwrapped = unwrap_strategies(strategy)
1202 if label is None:
1203 label = unwrapped.label
1204 assert isinstance(label, int)
1206 self.start_span(label=label)
1207 try:
1208 if not at_top_level:
1209 try:
1210 return unwrapped.do_draw(self)
1211 except FlakyStrategyDefinition as err:
1212 # Record the strategy stack as the error unwinds, so that an
1213 # inconsistent-generation failure is explained in terms of the
1214 # strategies being drawn from, not just the choice sequence.
1215 # The top-level draw adds its own "while generating ..." note.
1216 add_note(err, f"while drawing from {strategy!r}")
1217 raise
1218 assert start_time is not None
1219 key = observe_as or f"generate:unlabeled_{len(self.draw_times)}"
1220 try:
1221 try:
1222 v = unwrapped.do_draw(self)
1223 finally:
1224 # Subtract the time spent in GC to avoid overcounting, as it is
1225 # accounted for at the overall example level.
1226 in_gctime = gc_cumulative_time() - gc_start_time
1227 self.draw_times[key] = time.perf_counter() - start_time - in_gctime
1228 except Exception as err:
1229 add_note(
1230 err,
1231 f"while generating {key.removeprefix('generate:')!r} from {strategy!r}",
1232 )
1233 raise
1234 if observability_enabled():
1235 avoid = self.provider.avoid_realization
1236 self._observability_args[key] = to_jsonable(v, avoid_realization=avoid)
1237 return v
1238 finally:
1239 self.stop_span()
1241 @property
1242 def next_span_index(self) -> int:
1243 """The index that the next span to start will get. Spans are indexed
1244 in start order, so this also counts the spans started so far."""
1245 return self.__span_record.span_count
1247 def start_span(self, label: int) -> None:
1248 self.provider.span_start(label)
1249 self.__assert_not_frozen("start_span")
1250 self.depth += 1
1251 # Logically it would make sense for this to just be
1252 # ``self.depth = max(self.depth, self.max_depth)``, which is what it used to
1253 # be until we ran the code under tracemalloc and found a rather significant
1254 # chunk of allocation was happening here. This was presumably due to varargs
1255 # or the like, but we didn't investigate further given that it was easy
1256 # to fix with this check.
1257 if self.depth > self.max_depth:
1258 self.max_depth = self.depth
1259 self.__span_record.start_span(label)
1260 self.labels_for_structure_stack.append({label})
1262 def stop_span(self, *, discard: bool = False) -> None:
1263 self.provider.span_end(discard)
1264 if self.frozen:
1265 return
1266 if discard:
1267 self.has_discards = True
1268 self.depth -= 1
1269 assert self.depth >= -1
1270 self.__span_record.stop_span(discard=discard)
1272 labels_for_structure = self.labels_for_structure_stack.pop()
1274 if not discard:
1275 if self.labels_for_structure_stack:
1276 self.labels_for_structure_stack[-1].update(labels_for_structure)
1277 else:
1278 self.tags.update([structural_coverage(l) for l in labels_for_structure])
1280 if discard:
1281 # Once we've discarded a span, every test case starting with
1282 # this prefix contains discards. We prune the tree at that point so
1283 # as to avoid future test cases bothering with this region, on the
1284 # assumption that some span that you could have used instead
1285 # there would *not* trigger the discard. This greatly speeds up
1286 # test case generation in some cases, because it allows us to
1287 # ignore large swathes of the search space that are effectively
1288 # redundant.
1289 #
1290 # A scenario that can cause us problems but which we deliberately
1291 # have decided not to support is that if there are side effects
1292 # during data generation then you may end up with a scenario where
1293 # every good test case generates a discard because the discarded
1294 # section sets up important things for later. This is not terribly
1295 # likely and all that you see in this case is some degradation in
1296 # quality of testing, so we don't worry about it.
1297 #
1298 # Note that killing the branch does *not* mean we will never
1299 # explore below this point, and in particular we may do so during
1300 # shrinking. Any explicit request for a data object that starts
1301 # with the branch here will work just fine, but novel prefix
1302 # generation will avoid it, and we can use it to detect when we
1303 # have explored the entire tree (up to redundancy).
1305 self.observer.kill_branch()
1307 @property
1308 def spans(self) -> Spans:
1309 assert self.frozen
1310 if self.__spans is None:
1311 self.__spans = Spans(record=self.__span_record)
1312 return self.__spans
1314 def freeze(self) -> None:
1315 if self.frozen:
1316 return
1317 self.finish_time = time.perf_counter()
1318 self.gc_finish_time = gc_cumulative_time()
1320 # Always finish by closing all remaining spans so that we have a valid tree.
1321 while self.depth >= 0:
1322 self.stop_span()
1324 self.__span_record.freeze()
1325 self.frozen = True
1326 self.observer.conclude_test(self.status, self.interesting_origin)
1328 def choice(
1329 self,
1330 values: Sequence[T],
1331 *,
1332 forced: T | None = None,
1333 observe: bool = True,
1334 ) -> T:
1335 forced_i = None if forced is None else values.index(forced)
1336 i = self.draw_integer(
1337 0,
1338 len(values) - 1,
1339 forced=forced_i,
1340 observe=observe,
1341 )
1342 return values[i]
1344 def conclude_test(
1345 self,
1346 status: Status,
1347 interesting_origin: InterestingOrigin | None = None,
1348 ) -> NoReturn:
1349 assert (interesting_origin is None) or (status == Status.INTERESTING)
1350 self.__assert_not_frozen("conclude_test")
1351 self.interesting_origin = interesting_origin
1352 self.status = status
1353 self.freeze()
1354 raise StopTest(self.testcounter)
1356 def mark_interesting(self, interesting_origin: InterestingOrigin) -> NoReturn:
1357 self.conclude_test(Status.INTERESTING, interesting_origin)
1359 def mark_invalid(self, why: str | None = None) -> NoReturn:
1360 if why is not None:
1361 self.events["invalid because"] = why
1362 self.conclude_test(Status.INVALID)
1364 def mark_overrun(self) -> NoReturn:
1365 self.conclude_test(Status.OVERRUN)
1368def draw_choice(
1369 choice_type: ChoiceTypeT, constraints: ChoiceConstraintsT, *, random: Random
1370) -> ChoiceT:
1371 cd = ConjectureData(random=random)
1372 return cast(ChoiceT, getattr(cd.provider, f"draw_{choice_type}")(**constraints))