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 datetime
12import math
13import time
14import types
15import weakref
16from collections import defaultdict
17from collections.abc import Callable, Generator, Hashable, Iterable, Iterator, Sequence
18from contextlib import contextmanager
19from dataclasses import dataclass, field
20from enum import IntEnum
21from functools import cached_property
22from random import Random
23from typing import (
24 TYPE_CHECKING,
25 Any,
26 Literal,
27 NoReturn,
28 TypeAlias,
29 TypeVar,
30 cast,
31 overload,
32)
34from hypothesis.errors import (
35 CannotInvert,
36 CannotProceedScopeT,
37 ChoiceTooLarge,
38 FlakyStrategyDefinition,
39 Frozen,
40 InvalidArgument,
41 StopTest,
42)
43from hypothesis.internal.cache import LRUCache
44from hypothesis.internal.compat import add_note
45from hypothesis.internal.conjecture.choice import (
46 BooleanConstraints,
47 BytesConstraints,
48 ChoiceConstraintsT,
49 ChoiceNode,
50 ChoiceT,
51 ChoiceTemplate,
52 ChoiceTypeT,
53 FloatConstraints,
54 IntegerConstraints,
55 StringConstraints,
56 ValueHole,
57 choice_constraints_key,
58 choice_from_index,
59 choice_permitted,
60 choice_size,
61)
62from hypothesis.internal.conjecture.junkdrawer import IntList, gc_cumulative_time
63from hypothesis.internal.conjecture.providers import (
64 COLLECTION_DEFAULT_MAX_SIZE,
65 HypothesisProvider,
66 PrimitiveProvider,
67)
68from hypothesis.internal.conjecture.utils import calc_label_from_name
69from hypothesis.internal.escalation import InterestingOrigin
70from hypothesis.internal.floats import (
71 SMALLEST_SUBNORMAL,
72 float_to_int,
73 int_to_float,
74 sign_aware_lte,
75)
76from hypothesis.internal.intervalsets import IntervalSet
77from hypothesis.internal.observability import PredicateCounts
78from hypothesis.internal.reflection import function_location
79from hypothesis.reporting import debug_report
80from hypothesis.utils.conventions import UniqueIdentifier, not_set
81from hypothesis.utils.deprecation import note_deprecation
82from hypothesis.utils.threading import ThreadLocal
83from hypothesis.vendor.pretty import ArgLabelsT
85if TYPE_CHECKING:
86 from hypothesis.strategies import SearchStrategy
87 from hypothesis.strategies._internal.core import DataObject
88 from hypothesis.strategies._internal.random import RandomState
89 from hypothesis.strategies._internal.strategies import Ex
92def __getattr__(name: str) -> Any:
93 if name == "AVAILABLE_PROVIDERS":
94 from hypothesis.internal.conjecture.providers import AVAILABLE_PROVIDERS
96 note_deprecation(
97 "hypothesis.internal.conjecture.data.AVAILABLE_PROVIDERS has been moved to "
98 "hypothesis.internal.conjecture.providers.AVAILABLE_PROVIDERS.",
99 since="2025-01-25",
100 has_codemod=False,
101 stacklevel=1,
102 )
103 return AVAILABLE_PROVIDERS
105 raise AttributeError(
106 f"Module 'hypothesis.internal.conjecture.data' has no attribute {name}"
107 )
110T = TypeVar("T")
111TargetObservations = dict[str, int | float]
112# index, choice_type, constraints, forced value
113MisalignedAt: TypeAlias = tuple[int, ChoiceTypeT, ChoiceConstraintsT, ChoiceT | None]
115TOP_LABEL = calc_label_from_name("top")
116UNIQUE_COLLECTION_LABEL = calc_label_from_name(
117 "a collection with a uniqueness constraint"
118)
119MAX_DEPTH = 100
121threadlocal = ThreadLocal(global_test_counter=int)
124class Status(IntEnum):
125 OVERRUN = 0
126 INVALID = 1
127 VALID = 2
128 INTERESTING = 3
130 def __repr__(self) -> str:
131 return f"Status.{self.name}"
134@dataclass(slots=True, frozen=True)
135class StructuralCoverageTag:
136 label: int
139STRUCTURAL_COVERAGE_CACHE: dict[int, StructuralCoverageTag] = {}
142def structural_coverage(label: int) -> StructuralCoverageTag:
143 try:
144 return STRUCTURAL_COVERAGE_CACHE[label]
145 except KeyError:
146 return STRUCTURAL_COVERAGE_CACHE.setdefault(label, StructuralCoverageTag(label))
149# This cache can be quite hot and so we prefer LRUCache over LRUReusedCache for
150# performance. We lose scan resistance, but that's probably fine here.
151POOLED_CONSTRAINTS_CACHE: LRUCache[tuple[Any, ...], ChoiceConstraintsT] = LRUCache(4096)
153# The exact types whose values record_value_for_span holds directly: small
154# immutable stdlib types, none of which support weak references. Values of
155# any other type are held via weakref where possible, so that recording never
156# extends an object's lifetime. A tuple rather than a set so that ``in``
157# works even if a value's class has an unhashable metaclass. Exact type
158# checks also exclude symbolic backends.
159_RECORDABLE_VALUE_TYPES: tuple[type, ...] = (
160 int,
161 bool,
162 str,
163 bytes,
164 float,
165 types.NoneType,
166 datetime.date,
167 datetime.time,
168 datetime.datetime,
169 datetime.timedelta,
170)
172no_recorded_value = UniqueIdentifier("no_recorded_value")
175class Span:
176 """A span tracks the hierarchical structure of choices within a single test run.
178 Spans are created to mark regions of the choice sequence that are
179 logically related to each other. For instance, Hypothesis tracks:
180 - A single top-level span for the entire choice sequence
181 - A span for the choices made by each strategy
182 - Some strategies define additional spans within their choices. For instance,
183 st.lists() tracks the "should add another element" choice and the "add
184 another element" choices as separate spans.
186 Spans provide useful information to the shrinker, mutator, targeted PBT,
187 and other subsystems of Hypothesis.
189 Rather than store each ``Span`` as a rich object, it is actually
190 just an index into the ``Spans`` class defined below. This has two
191 purposes: Firstly, for most properties of spans we will never need
192 to allocate storage at all, because most properties are not used on
193 most spans. Secondly, by storing the spans as compact lists
194 of integers, we save a considerable amount of space compared to
195 Python's normal object size.
197 This does have the downside that it increases the amount of allocation
198 we do, and slows things down as a result, in some usage patterns because
199 we repeatedly allocate the same Span or int objects, but it will
200 often dramatically reduce our memory usage, so is worth it.
201 """
203 __slots__ = ("index", "owner")
205 def __init__(self, owner: "Spans", index: int) -> None:
206 self.owner = owner
207 self.index = index
209 def __eq__(self, other: object) -> bool:
210 if self is other:
211 return True
212 if not isinstance(other, Span):
213 return NotImplemented
214 return (self.owner is other.owner) and (self.index == other.index)
216 def __ne__(self, other: object) -> bool:
217 if self is other:
218 return False
219 if not isinstance(other, Span):
220 return NotImplemented
221 return (self.owner is not other.owner) or (self.index != other.index)
223 def __repr__(self) -> str:
224 return f"spans[{self.index}]"
226 @property
227 def label(self) -> int:
228 """A label is an opaque value that associates each span with its
229 approximate origin, such as a particular strategy class or a particular
230 kind of draw."""
231 return self.owner.labels[self.owner.label_indices[self.index]]
233 @property
234 def parent(self) -> int | None:
235 """The index of the span that this one is nested directly within."""
236 if self.index == 0:
237 return None
238 return self.owner.parentage[self.index]
240 @property
241 def start(self) -> int:
242 return self.owner.starts[self.index]
244 @property
245 def end(self) -> int:
246 return self.owner.ends[self.index]
248 @property
249 def depth(self) -> int:
250 """
251 Depth of this span in the span tree. The top-level span has a depth of 0.
252 """
253 return self.owner.depths[self.index]
255 @property
256 def discarded(self) -> bool:
257 """True if this is span's ``stop_span`` call had ``discard`` set to
258 ``True``. This means we believe that the shrinker should be able to delete
259 this span completely, without affecting the value produced by its enclosing
260 strategy. Typically set when a rejection sampler decides to reject a
261 generated value and try again."""
262 return self.index in self.owner.discarded
264 @property
265 def choice_count(self) -> int:
266 """The number of choices in this span."""
267 return self.end - self.start
269 @property
270 def children(self) -> "list[Span]":
271 """The list of all spans with this as a parent, in increasing index
272 order."""
273 return [self.owner[i] for i in self.owner.children[self.index]]
275 @property
276 def recorded_value(self) -> Any:
277 """The value produced by the strategy draw corresponding to this span,
278 or the ``no_recorded_value`` sentinel if no value was recorded, or if
279 a weakly-referenced value has since been collected."""
280 value = self.owner.span_values.get(self.index, no_recorded_value)
281 if isinstance(value, weakref.ReferenceType):
282 referent = value()
283 return no_recorded_value if referent is None else referent
284 return value
287class SpanProperty:
288 """There are many properties of spans that we calculate by
289 essentially rerunning the test case multiple times based on the
290 calls which we record in SpanProperty.
292 This class defines a visitor, subclasses of which can be used
293 to calculate these properties.
294 """
296 def __init__(self, spans: "Spans"):
297 self.span_stack: list[int] = []
298 self.spans = spans
299 self.span_count = 0
300 self.choice_count = 0
302 def run(self) -> Any:
303 """Rerun the test case with this visitor and return the
304 results of ``self.finish()``."""
305 for record in self.spans.trail:
306 if record == TrailType.STOP_SPAN_DISCARD:
307 self.__pop(discarded=True)
308 elif record == TrailType.STOP_SPAN_NO_DISCARD:
309 self.__pop(discarded=False)
310 elif record == TrailType.CHOICE:
311 self.choice_count += 1
312 else:
313 # everything after TrailType.CHOICE is the label of a span start.
314 self.__push(record - TrailType.CHOICE - 1)
316 return self.finish()
318 def __push(self, label_index: int) -> None:
319 i = self.span_count
320 assert i < len(self.spans)
321 self.start_span(i, label_index=label_index)
322 self.span_count += 1
323 self.span_stack.append(i)
325 def __pop(self, *, discarded: bool) -> None:
326 i = self.span_stack.pop()
327 self.stop_span(i, discarded=discarded)
329 def start_span(self, i: int, label_index: int) -> None:
330 """Called at the start of each span, with ``i`` the
331 index of the span and ``label_index`` the index of
332 its label in ``self.spans.labels``."""
334 def stop_span(self, i: int, *, discarded: bool) -> None:
335 """Called at the end of each span, with ``i`` the
336 index of the span and ``discarded`` being ``True`` if ``stop_span``
337 was called with ``discard=True``."""
339 def finish(self) -> Any:
340 raise NotImplementedError
343class TrailType(IntEnum):
344 STOP_SPAN_DISCARD = 1
345 STOP_SPAN_NO_DISCARD = 2
346 CHOICE = 3
347 # every trail element larger than TrailType.CHOICE is the label of a span
348 # start, offset by its index. So the first span label is stored as 4, the
349 # second as 5, etc, regardless of its actual integer label.
352class SpanRecord:
353 """Records the series of ``start_span``, ``stop_span``, and
354 ``draw_bits`` calls so that these may be stored in ``Spans`` and
355 replayed when we need to know about the structure of individual
356 ``Span`` objects.
358 Note that there is significant similarity between this class and
359 ``DataObserver``, and the plan is to eventually unify them, but
360 they currently have slightly different functions and implementations.
361 """
363 def __init__(self) -> None:
364 self.labels: list[int] = []
365 self.__index_of_labels: dict[int, int] | None = {}
366 self.trail = IntList()
367 self.nodes: list[ChoiceNode] = []
368 # The number of spans started so far, which is also the index that the
369 # next span to start will get. Spans are indexed in start order.
370 self.span_count = 0
371 self.span_values: dict[int, Any] = {}
373 def freeze(self) -> None:
374 self.__index_of_labels = None
376 def record_choice(self) -> None:
377 self.trail.append(TrailType.CHOICE)
379 def start_span(self, label: int) -> None:
380 assert self.__index_of_labels is not None
381 try:
382 i = self.__index_of_labels[label]
383 except KeyError:
384 i = self.__index_of_labels.setdefault(label, len(self.labels))
385 self.labels.append(label)
386 self.trail.append(TrailType.CHOICE + 1 + i)
387 self.span_count += 1
389 def stop_span(self, *, discard: bool) -> None:
390 if discard:
391 self.trail.append(TrailType.STOP_SPAN_DISCARD)
392 else:
393 self.trail.append(TrailType.STOP_SPAN_NO_DISCARD)
395 def record_value_for_span(self, span_index: int, value: Any) -> None:
396 # Record ``value`` against the span at ``span_index``. Called by
397 # ConjectureData.draw with the value each strategy's do_draw returned.
398 # Values which can be neither held directly nor weakly referenced
399 # (e.g. stdlib containers) are not recorded, and so cannot widen.
400 if type(value) in _RECORDABLE_VALUE_TYPES:
401 self.span_values[span_index] = value
402 else:
403 try:
404 self.span_values[span_index] = weakref.ref(value)
405 except TypeError:
406 pass
409class _starts_and_ends(SpanProperty):
410 def __init__(self, spans: "Spans") -> None:
411 super().__init__(spans)
412 self.starts = IntList.of_length(len(self.spans))
413 self.ends = IntList.of_length(len(self.spans))
415 def start_span(self, i: int, label_index: int) -> None:
416 self.starts[i] = self.choice_count
418 def stop_span(self, i: int, *, discarded: bool) -> None:
419 self.ends[i] = self.choice_count
421 def finish(self) -> tuple[IntList, IntList]:
422 return (self.starts, self.ends)
425class _discarded(SpanProperty):
426 def __init__(self, spans: "Spans") -> None:
427 super().__init__(spans)
428 self.result: set[int] = set()
430 def finish(self) -> frozenset[int]:
431 return frozenset(self.result)
433 def stop_span(self, i: int, *, discarded: bool) -> None:
434 if discarded:
435 self.result.add(i)
438class _parentage(SpanProperty):
439 def __init__(self, spans: "Spans") -> None:
440 super().__init__(spans)
441 self.result = IntList.of_length(len(self.spans))
443 def stop_span(self, i: int, *, discarded: bool) -> None:
444 if i > 0:
445 self.result[i] = self.span_stack[-1]
447 def finish(self) -> IntList:
448 return self.result
451class _depths(SpanProperty):
452 def __init__(self, spans: "Spans") -> None:
453 super().__init__(spans)
454 self.result = IntList.of_length(len(self.spans))
456 def start_span(self, i: int, label_index: int) -> None:
457 self.result[i] = len(self.span_stack)
459 def finish(self) -> IntList:
460 return self.result
463class _label_indices(SpanProperty):
464 def __init__(self, spans: "Spans") -> None:
465 super().__init__(spans)
466 self.result = IntList.of_length(len(self.spans))
468 def start_span(self, i: int, label_index: int) -> None:
469 self.result[i] = label_index
471 def finish(self) -> IntList:
472 return self.result
475class _mutator_groups(SpanProperty):
476 def __init__(self, spans: "Spans") -> None:
477 super().__init__(spans)
478 self.groups: dict[int, set[tuple[int, int]]] = defaultdict(set)
479 self._unique_stack: list[int] = []
481 def start_span(self, i: int, label_index: int) -> None:
482 # TODO should we discard start == end cases? occurs for eg st.data()
483 # which is conditionally or never drawn from. arguably swapping
484 # nodes with the empty list is a useful mutation enabled by start == end?
485 #
486 # Skip spans inside a unique collection: duplicating one element of a
487 # unique collection over another is certain to fail the uniqueness
488 # filter and be discarded. The collection's own span is still added,
489 # so duplicating a collection as a whole remains possible.
490 if not self._unique_stack:
491 key = (self.spans[i].start, self.spans[i].end)
492 self.groups[label_index].add(key)
493 if self.spans.labels[label_index] == UNIQUE_COLLECTION_LABEL:
494 self._unique_stack.append(i)
496 def stop_span(self, i: int, *, discarded: bool) -> None:
497 if self._unique_stack and self._unique_stack[-1] == i:
498 self._unique_stack.pop()
500 def finish(self) -> Iterable[set[tuple[int, int]]]:
501 # Discard groups with only one span, since the mutator can't
502 # do anything useful with them.
503 return [g for g in self.groups.values() if len(g) >= 2]
506class Spans:
507 """A lazy collection of ``Span`` objects, derived from
508 the record of recorded behaviour in ``SpanRecord``.
510 Behaves logically as if it were a list of ``Span`` objects,
511 but actually mostly exists as a compact store of information
512 for them to reference into. All properties on here are best
513 understood as the backing storage for ``Span`` and are
514 described there.
515 """
517 def __init__(self, record: SpanRecord) -> None:
518 self.trail = record.trail
519 self.labels = record.labels
520 self.span_values = record.span_values
521 self.__length = self.trail.count(
522 TrailType.STOP_SPAN_DISCARD
523 ) + record.trail.count(TrailType.STOP_SPAN_NO_DISCARD)
524 self.__children: list[Sequence[int]] | None = None
526 @cached_property
527 def starts_and_ends(self) -> tuple[IntList, IntList]:
528 return _starts_and_ends(self).run()
530 @property
531 def starts(self) -> IntList:
532 return self.starts_and_ends[0]
534 @property
535 def ends(self) -> IntList:
536 return self.starts_and_ends[1]
538 @cached_property
539 def discarded(self) -> frozenset[int]:
540 return _discarded(self).run()
542 @cached_property
543 def parentage(self) -> IntList:
544 return _parentage(self).run()
546 @cached_property
547 def depths(self) -> IntList:
548 return _depths(self).run()
550 @cached_property
551 def label_indices(self) -> IntList:
552 return _label_indices(self).run()
554 @cached_property
555 def mutator_groups(self) -> list[set[tuple[int, int]]]:
556 return _mutator_groups(self).run()
558 @property
559 def children(self) -> list[Sequence[int]]:
560 if self.__children is None:
561 children = [IntList() for _ in range(len(self))]
562 for i, p in enumerate(self.parentage):
563 if i > 0:
564 children[p].append(i)
565 # Replace empty children lists with a tuple to reduce
566 # memory usage.
567 for i, c in enumerate(children):
568 if not c:
569 children[i] = () # type: ignore
570 self.__children = children # type: ignore
571 return self.__children # type: ignore
573 def __len__(self) -> int:
574 return self.__length
576 def __getitem__(self, i: int) -> Span:
577 n = self.__length
578 if i < -n or i >= n:
579 raise IndexError(f"Index {i} out of range [-{n}, {n})")
580 if i < 0:
581 i += n
582 return Span(self, i)
584 # not strictly necessary as we have len/getitem, but required for mypy.
585 # https://github.com/python/mypy/issues/9737
586 def __iter__(self) -> Iterator[Span]:
587 for i in range(len(self)):
588 yield self[i]
591class _Overrun:
592 status: Status = Status.OVERRUN
594 def __repr__(self) -> str:
595 return "Overrun"
598Overrun = _Overrun()
601class DataObserver:
602 """Observer class for recording the behaviour of a
603 ConjectureData object, primarily used for tracking
604 the behaviour in the tree cache."""
606 def conclude_test(
607 self,
608 status: Status,
609 interesting_origin: InterestingOrigin | None,
610 ) -> None:
611 """Called when ``conclude_test`` is called on the
612 observed ``ConjectureData``, with the same arguments.
614 Note that this is called after ``freeze`` has completed.
615 """
617 def kill_branch(self) -> None:
618 """Mark this part of the tree as not worth re-exploring."""
620 def draw_integer(
621 self, value: int, *, constraints: IntegerConstraints, was_forced: bool
622 ) -> None:
623 pass
625 def draw_float(
626 self, value: float, *, constraints: FloatConstraints, was_forced: bool
627 ) -> None:
628 pass
630 def draw_string(
631 self, value: str, *, constraints: StringConstraints, was_forced: bool
632 ) -> None:
633 pass
635 def draw_bytes(
636 self, value: bytes, *, constraints: BytesConstraints, was_forced: bool
637 ) -> None:
638 pass
640 def draw_boolean(
641 self, value: bool, *, constraints: BooleanConstraints, was_forced: bool
642 ) -> None:
643 pass
646@dataclass(slots=True, frozen=True)
647class ConjectureResult:
648 """Result class storing the parts of ConjectureData that we
649 will care about after the original ConjectureData has outlived its
650 usefulness."""
652 status: Status
653 interesting_origin: InterestingOrigin | None
654 nodes: tuple[ChoiceNode, ...] = field(repr=False, compare=False)
655 length: int
656 notes: list[str]
657 expected_exception: BaseException | None
658 expected_traceback: str | None
659 has_discards: bool
660 target_observations: TargetObservations
661 tags: frozenset[StructuralCoverageTag]
662 spans: Spans = field(repr=False, compare=False)
663 arg_spans: set[int] = field(repr=False)
664 # Comments for the explain phase, keyed by span index. The ``None`` key
665 # holds the whole-test comment about varying all commented parts together.
666 span_comments: dict[int | None, str] = field(repr=False)
667 misaligned_at: MisalignedAt | None = field(repr=False)
668 cannot_proceed_scope: CannotProceedScopeT | None = field(repr=False)
670 def as_result(self) -> "ConjectureResult":
671 return self
673 @property
674 def choices(self) -> tuple[ChoiceT, ...]:
675 return tuple(node.value for node in self.nodes)
678class ConjectureData:
679 @classmethod
680 def for_choices(
681 cls,
682 choices: Sequence[ChoiceTemplate | ValueHole | ChoiceT],
683 *,
684 observer: DataObserver | None = None,
685 provider: PrimitiveProvider | type[PrimitiveProvider] = HypothesisProvider,
686 random: Random | None = None,
687 ) -> "ConjectureData":
688 from hypothesis.internal.conjecture.engine import choice_count
690 return cls(
691 max_choices=choice_count(choices),
692 random=random,
693 prefix=choices,
694 observer=observer,
695 provider=provider,
696 )
698 def __init__(
699 self,
700 *,
701 random: Random | None,
702 observer: DataObserver | None = None,
703 provider: PrimitiveProvider | type[PrimitiveProvider] = HypothesisProvider,
704 prefix: Sequence[ChoiceTemplate | ValueHole | ChoiceT] | None = None,
705 max_choices: int | None = None,
706 provider_kw: dict[str, Any] | None = None,
707 ) -> None:
708 from hypothesis.internal.conjecture.engine import BUFFER_SIZE
710 if observer is None:
711 observer = DataObserver()
712 if provider_kw is None:
713 provider_kw = {}
714 elif not isinstance(provider, type):
715 raise InvalidArgument(
716 f"Expected {provider=} to be a class since {provider_kw=} was "
717 "passed, but got an instance instead."
718 )
720 assert isinstance(observer, DataObserver)
721 self.observer = observer
722 self.max_choices = max_choices
723 self.max_length = BUFFER_SIZE
724 self.overdraw = 0
725 self._random = random
727 self.length: int = 0
728 self.index: int = 0
729 self.notes: list[str] = []
730 self.status: Status = Status.VALID
731 self.frozen: bool = False
732 self.testcounter: int = threadlocal.global_test_counter
733 threadlocal.global_test_counter += 1
734 self.start_time = time.perf_counter()
735 self.gc_start_time = gc_cumulative_time()
736 self.events: dict[str, str | int | float] = {}
737 self.interesting_origin: InterestingOrigin | None = None
738 self.draw_times: dict[str, float] = {}
739 self._stateful_run_times: dict[str, float] = defaultdict(float)
740 self.max_depth: int = 0
741 self.has_discards: bool = False
742 # While positive, we are drawing an element of a unique collection, so
743 # a "simplest" ChoiceTemplate enumerates progressively larger choices
744 # (via the shared counter) instead of repeating guaranteed-duplicates.
745 self.unique_element_depth: int = 0
746 self.simplest_index: int = 0
748 self.provider: PrimitiveProvider = (
749 provider(self, **provider_kw) if isinstance(provider, type) else provider
750 )
751 assert isinstance(self.provider, PrimitiveProvider)
753 self.__result: ConjectureResult | None = None
755 # Observations used for targeted search. They'll be aggregated in
756 # ConjectureRunner.generate_new_test_cases and fed to TargetSelector.
757 self.target_observations: TargetObservations = {}
759 # Tags which indicate something about which part of the search space
760 # this example is in. These are used to guide generation.
761 self.tags: set[StructuralCoverageTag] = set()
762 self.labels_for_structure_stack: list[set[int]] = []
764 # Normally unpopulated but we need this in the niche case
765 # that self.as_result() is Overrun but we still want the
766 # examples for reporting purposes.
767 self.__spans: Spans | None = None
769 # We want the top level span to have depth 0, so we start at -1.
770 self.depth: int = -1
771 self.__span_record = SpanRecord()
773 # Span indices for discrete reportable parts that which-parts-matter can
774 # try varying, to report if the minimal test case always fails anyway.
775 self.arg_spans: set[int] = set()
776 self.span_comments: dict[int | None, str] = {}
777 self._observability_args: dict[str, Any] = {}
778 self._observability_predicates: defaultdict[str, PredicateCounts] = defaultdict(
779 PredicateCounts
780 )
781 self.invalid_location: str | None = None
782 # (predicate, location) of the most recent filter rejection
783 self._last_rejected_filter: tuple[Callable[[Any], Any], str | None] | None = (
784 None
785 )
787 self._sampled_from_all_strategies_elements_message: (
788 tuple[str, object] | None
789 ) = None
790 self._shared_strategy_draws: dict[Hashable, tuple[Any, SearchStrategy]] = {}
791 self._shared_data_strategy: DataObject | None = None
792 self._stateful_repr_parts: list[Any] | None = None
793 self.states_for_ids: dict[int, RandomState] | None = None
794 self.seeds_to_states: dict[Any, RandomState] | None = None
795 self.hypothesis_runner: Any = not_set
797 self.expected_exception: BaseException | None = None
798 self.expected_traceback: str | None = None
800 self.prefix = prefix
801 self._inverting = False
802 # Appending to a tuple is O(n), so we accumulate nodes in a list and
803 # expose them as a (cached) tuple.
804 self._nodes: list[ChoiceNode] = []
805 self._nodes_tuple: tuple[ChoiceNode, ...] = ()
806 self.misaligned_at: MisalignedAt | None = None
807 self.cannot_proceed_scope: CannotProceedScopeT | None = None
808 self.start_span(TOP_LABEL)
810 def __repr__(self) -> str:
811 return (
812 f"ConjectureData({self.status.name}, {len(self.nodes)} "
813 f"choices{', frozen' if self.frozen else ''})"
814 )
816 @property
817 def nodes(self) -> tuple[ChoiceNode, ...]:
818 if len(self._nodes_tuple) != len(self._nodes):
819 self._nodes_tuple = tuple(self._nodes)
820 return self._nodes_tuple
822 @property
823 def choices(self) -> tuple[ChoiceT, ...]:
824 return tuple(node.value for node in self._nodes)
826 # draw_* functions might be called in one of two contexts: either "above" or
827 # "below" the choice sequence. For instance, draw_string calls draw_boolean
828 # from ``many`` when calculating the number of characters to return. We do
829 # not want these choices to get written to the choice sequence, because they
830 # are not true choices themselves.
831 #
832 # `observe` formalizes this. The choice will only be written to the choice
833 # sequence if observe is True.
835 @overload
836 def _draw(
837 self,
838 choice_type: Literal["integer"],
839 constraints: IntegerConstraints,
840 *,
841 observe: bool,
842 forced: int | None,
843 ) -> int: ...
845 @overload
846 def _draw(
847 self,
848 choice_type: Literal["float"],
849 constraints: FloatConstraints,
850 *,
851 observe: bool,
852 forced: float | None,
853 ) -> float: ...
855 @overload
856 def _draw(
857 self,
858 choice_type: Literal["string"],
859 constraints: StringConstraints,
860 *,
861 observe: bool,
862 forced: str | None,
863 ) -> str: ...
865 @overload
866 def _draw(
867 self,
868 choice_type: Literal["bytes"],
869 constraints: BytesConstraints,
870 *,
871 observe: bool,
872 forced: bytes | None,
873 ) -> bytes: ...
875 @overload
876 def _draw(
877 self,
878 choice_type: Literal["boolean"],
879 constraints: BooleanConstraints,
880 *,
881 observe: bool,
882 forced: bool | None,
883 ) -> bool: ...
885 def _draw(
886 self,
887 choice_type: ChoiceTypeT,
888 constraints: ChoiceConstraintsT,
889 *,
890 observe: bool,
891 forced: ChoiceT | None,
892 ) -> ChoiceT:
893 if self._inverting:
894 # A strategy tried to draw while re-encoding a ValueHole - e.g. a
895 # filter predicate which draws, like stateful's rule filters.
896 # Inversions must be pure, so treat this as unencodable.
897 raise CannotInvert("cannot draw during _invert")
898 # this is somewhat redundant with the length > max_length check at the
899 # end of the function, but avoids trying to use a null self.random when
900 # drawing past the node of a ConjectureData.for_choices data.
901 if self.length == self.max_length:
902 debug_report(f"overrun because hit {self.max_length=}")
903 self.mark_overrun()
904 if len(self._nodes) == self.max_choices:
905 debug_report(f"overrun because hit {self.max_choices=}")
906 self.mark_overrun()
908 if observe and self.prefix is not None and self.index < len(self.prefix):
909 value = self._pop_choice(choice_type, constraints, forced=forced)
910 if forced is not None:
911 value = forced
912 elif forced is None:
913 value = getattr(self.provider, f"draw_{choice_type}")(**constraints)
914 else:
915 value = forced
917 # nan values generated via int_to_float break list membership:
918 #
919 # >>> n = 18444492273895866368
920 # >>> assert math.isnan(int_to_float(n))
921 # >>> assert int_to_float(n) not in [int_to_float(n)]
922 #
923 # because int_to_float nans are not equal in the sense of either
924 # `a == b` or `a is b`.
925 #
926 # This can lead to flaky errors when collections require unique
927 # floats. What was happening is that in some places we provided math.nan
928 # provide math.nan, and in others we provided
929 # int_to_float(float_to_int(math.nan)), and which one gets used
930 # was not deterministic across test iterations.
931 #
932 # To fix this, *never* provide a nan value which is equal (via `is`) to
933 # another provided nan value. This sacrifices some test power; we should
934 # bring that back (ABOVE the choice sequence layer) in the future.
935 #
936 # See https://github.com/HypothesisWorks/hypothesis/issues/3926.
937 if choice_type == "float":
938 assert isinstance(value, float)
939 if math.isnan(value):
940 value = int_to_float(float_to_int(value))
942 if observe:
943 was_forced = forced is not None
944 getattr(self.observer, f"draw_{choice_type}")(
945 value, constraints=constraints, was_forced=was_forced
946 )
947 size = 0 if self.provider.avoid_realization else choice_size(value)
948 if self.length + size > self.max_length:
949 debug_report(
950 f"overrun because {self.length=} + {size=} > {self.max_length=}"
951 )
952 self.mark_overrun()
954 node = ChoiceNode(
955 type=choice_type,
956 value=value,
957 constraints=constraints,
958 was_forced=was_forced,
959 index=len(self._nodes),
960 )
961 self.__span_record.record_choice()
962 self._nodes.append(node)
963 self.length += size
965 return value
967 def draw_integer(
968 self,
969 min_value: int | None = None,
970 max_value: int | None = None,
971 *,
972 weights: dict[int, float] | None = None,
973 shrink_towards: int = 0,
974 forced: int | None = None,
975 observe: bool = True,
976 ) -> int:
977 # Validate arguments
978 if weights is not None:
979 assert min_value is not None
980 assert max_value is not None
981 assert len(weights) <= 255 # arbitrary practical limit
982 # We can and should eventually support total weights. But this
983 # complicates shrinking as we can no longer assume we can force
984 # a value to the unmapped probability mass if that mass might be 0.
985 assert sum(weights.values()) < 1
986 # similarly, things get simpler if we assume every value is possible.
987 # we'll want to drop this restriction eventually.
988 assert all(w != 0 for w in weights.values())
990 if forced is not None and min_value is not None:
991 assert min_value <= forced
992 if forced is not None and max_value is not None:
993 assert forced <= max_value
995 constraints: IntegerConstraints = self._pooled_constraints(
996 "integer",
997 {
998 "min_value": min_value,
999 "max_value": max_value,
1000 "weights": weights,
1001 "shrink_towards": shrink_towards,
1002 },
1003 )
1004 return self._draw("integer", constraints, observe=observe, forced=forced)
1006 def draw_float(
1007 self,
1008 min_value: float = -math.inf,
1009 max_value: float = math.inf,
1010 *,
1011 allow_nan: bool = True,
1012 smallest_nonzero_magnitude: float = SMALLEST_SUBNORMAL,
1013 # TODO: consider supporting these float widths at the choice sequence
1014 # level in the future.
1015 # width: Literal[16, 32, 64] = 64,
1016 forced: float | None = None,
1017 observe: bool = True,
1018 ) -> float:
1019 assert smallest_nonzero_magnitude > 0
1020 assert not math.isnan(min_value)
1021 assert not math.isnan(max_value)
1023 if smallest_nonzero_magnitude == 0.0: # pragma: no cover
1024 raise FloatingPointError(
1025 "Got allow_subnormal=True, but we can't represent subnormal floats "
1026 "right now, in violation of the IEEE-754 floating-point "
1027 "specification. This is usually because something was compiled with "
1028 "-ffast-math or a similar option, which sets global processor state. "
1029 "See https://simonbyrne.github.io/notes/fastmath/ for a more detailed "
1030 "writeup - and good luck!"
1031 )
1033 if forced is not None:
1034 assert allow_nan or not math.isnan(forced)
1035 assert math.isnan(forced) or (
1036 sign_aware_lte(min_value, forced) and sign_aware_lte(forced, max_value)
1037 )
1039 constraints: FloatConstraints = self._pooled_constraints(
1040 "float",
1041 {
1042 "min_value": min_value,
1043 "max_value": max_value,
1044 "allow_nan": allow_nan,
1045 "smallest_nonzero_magnitude": smallest_nonzero_magnitude,
1046 },
1047 )
1048 return self._draw("float", constraints, observe=observe, forced=forced)
1050 def draw_string(
1051 self,
1052 intervals: IntervalSet,
1053 *,
1054 min_size: int = 0,
1055 max_size: int = COLLECTION_DEFAULT_MAX_SIZE,
1056 forced: str | None = None,
1057 observe: bool = True,
1058 ) -> str:
1059 assert forced is None or min_size <= len(forced) <= max_size
1060 assert min_size >= 0
1061 if len(intervals) == 0:
1062 assert min_size == 0
1064 constraints: StringConstraints = self._pooled_constraints(
1065 "string",
1066 {
1067 "intervals": intervals,
1068 "min_size": min_size,
1069 "max_size": max_size,
1070 },
1071 )
1072 return self._draw("string", constraints, observe=observe, forced=forced)
1074 def draw_bytes(
1075 self,
1076 min_size: int = 0,
1077 max_size: int = COLLECTION_DEFAULT_MAX_SIZE,
1078 *,
1079 forced: bytes | None = None,
1080 observe: bool = True,
1081 ) -> bytes:
1082 assert forced is None or min_size <= len(forced) <= max_size
1083 assert min_size >= 0
1085 constraints: BytesConstraints = self._pooled_constraints(
1086 "bytes", {"min_size": min_size, "max_size": max_size}
1087 )
1088 return self._draw("bytes", constraints, observe=observe, forced=forced)
1090 def draw_boolean(
1091 self,
1092 p: float = 0.5,
1093 *,
1094 forced: bool | None = None,
1095 observe: bool = True,
1096 ) -> bool:
1097 assert (forced is not True) or p > 0
1098 assert (forced is not False) or p < 1
1100 constraints: BooleanConstraints = self._pooled_constraints("boolean", {"p": p})
1101 return self._draw("boolean", constraints, observe=observe, forced=forced)
1103 @overload
1104 def _pooled_constraints(
1105 self, choice_type: Literal["integer"], constraints: IntegerConstraints
1106 ) -> IntegerConstraints: ...
1108 @overload
1109 def _pooled_constraints(
1110 self, choice_type: Literal["float"], constraints: FloatConstraints
1111 ) -> FloatConstraints: ...
1113 @overload
1114 def _pooled_constraints(
1115 self, choice_type: Literal["string"], constraints: StringConstraints
1116 ) -> StringConstraints: ...
1118 @overload
1119 def _pooled_constraints(
1120 self, choice_type: Literal["bytes"], constraints: BytesConstraints
1121 ) -> BytesConstraints: ...
1123 @overload
1124 def _pooled_constraints(
1125 self, choice_type: Literal["boolean"], constraints: BooleanConstraints
1126 ) -> BooleanConstraints: ...
1128 def _pooled_constraints(
1129 self, choice_type: ChoiceTypeT, constraints: ChoiceConstraintsT
1130 ) -> ChoiceConstraintsT:
1131 """Memoize common dictionary objects to reduce memory pressure."""
1132 # caching runs afoul of nondeterminism checks
1133 if self.provider.avoid_realization:
1134 return constraints
1136 key = (choice_type, *choice_constraints_key(choice_type, constraints))
1137 try:
1138 return POOLED_CONSTRAINTS_CACHE[key]
1139 except KeyError:
1140 POOLED_CONSTRAINTS_CACHE[key] = constraints
1141 return constraints
1143 def _pop_choice(
1144 self,
1145 choice_type: ChoiceTypeT,
1146 constraints: ChoiceConstraintsT,
1147 *,
1148 forced: ChoiceT | None,
1149 ) -> ChoiceT:
1150 assert self.prefix is not None
1151 # checked in _draw
1152 assert self.index < len(self.prefix)
1154 value = self.prefix[self.index]
1155 if isinstance(value, ChoiceTemplate):
1156 node: ChoiceTemplate = value
1157 if node.count is not None:
1158 assert node.count >= 0
1159 # node templates have to be at the end for now, since it's not immediately
1160 # apparent how to handle overruning a node template while generating a single
1161 # node if the alternative is not "the entire data is an overrun".
1162 assert self.index == len(self.prefix) - 1
1163 if node.type == "simplest":
1164 index = 0
1165 if forced is None and self.unique_element_depth:
1166 # imported here to break an import cycle
1167 from hypothesis.internal.conjecture.datatree import (
1168 compute_max_children,
1169 )
1171 index = self.simplest_index % compute_max_children(
1172 choice_type, constraints
1173 )
1174 self.simplest_index += 1
1175 try:
1176 choice = choice_from_index(index, choice_type, constraints)
1177 except ChoiceTooLarge:
1178 self.mark_overrun()
1179 else:
1180 raise NotImplementedError
1182 if node.count is not None:
1183 node.count -= 1
1184 if node.count < 0:
1185 self.mark_overrun()
1186 return choice
1188 if isinstance(value, ValueHole):
1189 # A hole that no strategy claimed: either its value could not be
1190 # inverted, or it fell out of alignment with strategy draw
1191 # boundaries. Treat it as a misalignment.
1192 if self.misaligned_at is None:
1193 self.misaligned_at = (self.index, choice_type, constraints, forced)
1194 try:
1195 choice = choice_from_index(0, choice_type, constraints)
1196 except ChoiceTooLarge:
1197 self.mark_overrun()
1198 self.index += 1
1199 return choice
1201 choice = value
1202 node_choice_type = {
1203 str: "string",
1204 float: "float",
1205 int: "integer",
1206 bool: "boolean",
1207 bytes: "bytes",
1208 }[type(choice)]
1209 # If we're trying to:
1210 # * draw a different choice type at the same location
1211 # * draw the same choice type with a different constraints, which does not permit
1212 # the current value
1213 #
1214 # then we call this a misalignment, because the choice sequence has
1215 # changed from what we expected at some point. An easy misalignment is
1216 #
1217 # one_of(integers(0, 100), integers(101, 200))
1218 #
1219 # where the choice sequence [0, 100] has constraints {min_value: 0, max_value: 100}
1220 # at index 1, but [0, 101] has constraints {min_value: 101, max_value: 200} at
1221 # index 1 (which does not permit any of the values 0-100).
1222 #
1223 # When the choice sequence becomes misaligned, we generate a new value of the
1224 # type and constraints the strategy expects.
1225 if node_choice_type != choice_type or not choice_permitted(choice, constraints):
1226 # only track first misalignment for now.
1227 if self.misaligned_at is None:
1228 self.misaligned_at = (self.index, choice_type, constraints, forced)
1229 try:
1230 # Fill in any misalignments with index 0 choices. An alternative to
1231 # this is using the index of the misaligned choice instead
1232 # of index 0, which may be useful for maintaining
1233 # "similarly-complex choices" in the shrinker. This requires
1234 # attaching an index to every choice in ConjectureData.for_choices,
1235 # which we don't always have (e.g. when reading from db).
1236 #
1237 # If we really wanted this in the future we could make this complexity
1238 # optional, use it if present, and default to index 0 otherwise.
1239 # This complicates our internal api and so I'd like to avoid it
1240 # if possible.
1241 #
1242 # Additionally, I don't think slips which require
1243 # slipping to high-complexity values are common. Though arguably
1244 # we may want to expand a bit beyond *just* the simplest choice.
1245 # (we could for example consider sampling choices from index 0-10).
1246 choice = choice_from_index(0, choice_type, constraints)
1247 except ChoiceTooLarge:
1248 # should really never happen with a 0-index choice, but let's be safe.
1249 self.mark_overrun()
1251 self.index += 1
1252 return choice
1254 def as_result(self) -> ConjectureResult | _Overrun:
1255 """Convert the result of running this test into
1256 either an Overrun object or a ConjectureResult."""
1258 assert self.frozen
1259 if self.status == Status.OVERRUN:
1260 return Overrun
1261 if self.__result is None:
1262 self.__result = ConjectureResult(
1263 status=self.status,
1264 interesting_origin=self.interesting_origin,
1265 spans=self.spans,
1266 nodes=self.nodes,
1267 length=self.length,
1268 notes=self.notes,
1269 expected_traceback=self.expected_traceback,
1270 expected_exception=self.expected_exception,
1271 has_discards=self.has_discards,
1272 target_observations=self.target_observations,
1273 tags=frozenset(self.tags),
1274 arg_spans=self.arg_spans,
1275 span_comments=self.span_comments,
1276 misaligned_at=self.misaligned_at,
1277 cannot_proceed_scope=self.cannot_proceed_scope,
1278 )
1279 assert self.__result is not None
1280 return self.__result
1282 def __assert_not_frozen(self, name: str) -> None:
1283 if self.frozen:
1284 raise Frozen(f"Cannot call {name} on frozen ConjectureData")
1286 def note(self, value: str) -> None:
1287 self.__assert_not_frozen("note")
1288 self.notes.append(value)
1290 def draw(
1291 self,
1292 strategy: "SearchStrategy[Ex]",
1293 label: int | None = None,
1294 observe_as: str | None = None,
1295 ) -> "Ex":
1296 from hypothesis.internal.observability import observability_enabled
1297 from hypothesis.strategies._internal.lazy import unwrap_strategies
1298 from hypothesis.strategies._internal.utils import to_jsonable
1300 at_top_level = self.depth == 0
1301 start_time = None
1302 gc_start_time = None
1303 if at_top_level:
1304 # We start this timer early, because accessing attributes on a LazyStrategy
1305 # can be almost arbitrarily slow. In cases like characters() and text()
1306 # where we cache something expensive, this led to Flaky deadline errors!
1307 # See https://github.com/HypothesisWorks/hypothesis/issues/2108
1308 start_time = time.perf_counter()
1309 gc_start_time = gc_cumulative_time()
1311 strategy.validate()
1313 if strategy.is_empty:
1314 self.mark_invalid(f"empty strategy {strategy!r}")
1316 if self.depth >= MAX_DEPTH:
1317 self.mark_invalid("max depth exceeded")
1319 # Jump directly to the unwrapped strategy for the label and for do_draw.
1320 # This avoids adding an extra span to all lazy strategies.
1321 unwrapped = unwrap_strategies(strategy)
1322 if label is None:
1323 label = unwrapped.label
1324 assert isinstance(label, int)
1326 # If the next prefix element is a ValueHole, we are the strategy being
1327 # asked to re-encode its value: replace the hole with our inversion of
1328 # it, and let do_draw consume those choices (under our own constraints)
1329 # as usual. If we can't invert it, leave the hole for _pop_choice to
1330 # treat as a misalignment.
1331 if (
1332 self.prefix is not None
1333 and self.index < len(self.prefix)
1334 and isinstance(hole := self.prefix[self.index], ValueHole)
1335 ):
1336 self._inverting = True
1337 try:
1338 inverted = unwrapped._invert(hole.value)
1339 except Exception:
1340 # Usually CannotInvert, but _invert may execute arbitrary user code, eg
1341 # if a .filter is involved.
1342 pass
1343 else:
1344 self.prefix = (
1345 tuple(self.prefix[: self.index])
1346 + inverted
1347 + tuple(self.prefix[self.index + 1 :])
1348 )
1349 finally:
1350 self._inverting = False
1352 span_index = self.__span_record.span_count
1353 self.start_span(label=label)
1354 try:
1355 if not at_top_level:
1356 try:
1357 v = unwrapped.do_draw(self)
1358 self.__span_record.record_value_for_span(span_index, v)
1359 return v
1360 except FlakyStrategyDefinition as err:
1361 # Record the strategy stack as the error unwinds, so that an
1362 # inconsistent-generation failure is explained in terms of the
1363 # strategies being drawn from, not just the choice sequence.
1364 # The top-level draw adds its own "while generating ..." note.
1365 add_note(err, f"while drawing from {strategy!r}")
1366 raise
1367 assert start_time is not None
1368 assert gc_start_time is not None
1369 key = observe_as or f"generate:unlabeled_{len(self.draw_times)}"
1370 try:
1371 try:
1372 v = unwrapped.do_draw(self)
1373 finally:
1374 # Subtract the time spent in GC to avoid overcounting, as it is
1375 # accounted for at the overall example level.
1376 in_gctime = gc_cumulative_time() - gc_start_time
1377 self.draw_times[key] = time.perf_counter() - start_time - in_gctime
1378 except Exception as err:
1379 add_note(
1380 err,
1381 f"while generating {key.removeprefix('generate:')!r} from {strategy!r}",
1382 )
1383 raise
1384 if observability_enabled():
1385 avoid = self.provider.avoid_realization
1386 self._observability_args[key] = to_jsonable(v, avoid_realization=avoid)
1387 self.__span_record.record_value_for_span(span_index, v)
1388 return v
1389 finally:
1390 self.stop_span()
1392 @property
1393 def next_span_index(self) -> int:
1394 """The index that the next span to start will get. Spans are indexed
1395 in start order, so this also counts the spans started so far."""
1396 return self.__span_record.span_count
1398 @contextmanager
1399 def track_arg_span(self) -> Generator[int]:
1400 # Record the span opened by the draw inside this block in ``arg_spans``,
1401 # for the shrinker's explain phase to vary and comment on.
1402 #
1403 # Yields the span's index, which we know in advance even though Span
1404 # objects are only materialized after the test case is completed. (If the
1405 # draw raises instead, we skip recording, along with the rest of the test
1406 # case.)
1407 span_index = self.next_span_index
1408 yield span_index
1409 self.arg_spans.add(span_index)
1411 @contextmanager
1412 def track_arg_label(self, label: str) -> Generator[ArgLabelsT]:
1413 arg_labels: ArgLabelsT = {}
1415 with self.track_arg_span() as span_index:
1416 yield arg_labels
1418 # Mutate the arg_labels dict so that the pretty-printer knows where to
1419 # place the which-parts-matter comments later.
1420 arg_labels[label] = span_index
1422 def start_span(self, label: int) -> None:
1423 self.provider.span_start(label)
1424 self.__assert_not_frozen("start_span")
1425 self.depth += 1
1426 # Logically it would make sense for this to just be
1427 # ``self.depth = max(self.depth, self.max_depth)``, which is what it used to
1428 # be until we ran the code under tracemalloc and found a rather significant
1429 # chunk of allocation was happening here. This was presumably due to varargs
1430 # or the like, but we didn't investigate further given that it was easy
1431 # to fix with this check.
1432 if self.depth > self.max_depth:
1433 self.max_depth = self.depth
1434 self.__span_record.start_span(label)
1435 self.labels_for_structure_stack.append({label})
1437 def stop_span(self, *, discard: bool = False) -> None:
1438 self.provider.span_end(discard)
1439 if self.frozen:
1440 return
1441 if discard:
1442 self.has_discards = True
1443 self.depth -= 1
1444 assert self.depth >= -1
1445 self.__span_record.stop_span(discard=discard)
1447 labels_for_structure = self.labels_for_structure_stack.pop()
1449 if not discard:
1450 if self.labels_for_structure_stack:
1451 self.labels_for_structure_stack[-1].update(labels_for_structure)
1452 else:
1453 self.tags.update([structural_coverage(l) for l in labels_for_structure])
1455 if discard:
1456 # Once we've discarded a span, every test case starting with
1457 # this prefix contains discards. We prune the tree at that point so
1458 # as to avoid future test cases bothering with this region, on the
1459 # assumption that some span that you could have used instead
1460 # there would *not* trigger the discard. This greatly speeds up
1461 # test case generation in some cases, because it allows us to
1462 # ignore large swathes of the search space that are effectively
1463 # redundant.
1464 #
1465 # A scenario that can cause us problems but which we deliberately
1466 # have decided not to support is that if there are side effects
1467 # during data generation then you may end up with a scenario where
1468 # every good test case generates a discard because the discarded
1469 # section sets up important things for later. This is not terribly
1470 # likely and all that you see in this case is some degradation in
1471 # quality of testing, so we don't worry about it.
1472 #
1473 # Note that killing the branch does *not* mean we will never
1474 # explore below this point, and in particular we may do so during
1475 # shrinking. Any explicit request for a data object that starts
1476 # with the branch here will work just fine, but novel prefix
1477 # generation will avoid it, and we can use it to detect when we
1478 # have explored the entire tree (up to redundancy).
1480 self.observer.kill_branch()
1482 @property
1483 def spans(self) -> Spans:
1484 assert self.frozen
1485 if self.__spans is None:
1486 self.__spans = Spans(record=self.__span_record)
1487 return self.__spans
1489 def freeze(self) -> None:
1490 if self.frozen:
1491 return
1492 self.finish_time = time.perf_counter()
1493 self.gc_finish_time = gc_cumulative_time()
1495 # Always finish by closing all remaining spans so that we have a valid tree.
1496 while self.depth >= 0:
1497 self.stop_span()
1499 self.__span_record.freeze()
1500 self.frozen = True
1501 self.observer.conclude_test(self.status, self.interesting_origin)
1503 def choice(
1504 self,
1505 values: Sequence[T],
1506 *,
1507 forced: T | None = None,
1508 observe: bool = True,
1509 ) -> T:
1510 forced_i = None if forced is None else values.index(forced)
1511 i = self.draw_integer(
1512 0,
1513 len(values) - 1,
1514 forced=forced_i,
1515 observe=observe,
1516 )
1517 return values[i]
1519 def conclude_test(
1520 self,
1521 status: Status,
1522 interesting_origin: InterestingOrigin | None = None,
1523 ) -> NoReturn:
1524 assert (interesting_origin is None) or (status == Status.INTERESTING)
1525 self.__assert_not_frozen("conclude_test")
1526 self.interesting_origin = interesting_origin
1527 self.status = status
1528 self.freeze()
1529 raise StopTest(self.testcounter)
1531 def mark_interesting(self, interesting_origin: InterestingOrigin) -> NoReturn:
1532 self.conclude_test(Status.INTERESTING, interesting_origin)
1534 def mark_invalid(
1535 self, why: str | None = None, *, location: str | None = None
1536 ) -> NoReturn:
1537 if why is not None:
1538 self.events["gave up because"] = why
1539 self.invalid_location = location
1540 self.conclude_test(Status.INVALID)
1542 def mark_overrun(self) -> NoReturn:
1543 self.conclude_test(Status.OVERRUN)
1545 def last_rejected_filter_location(self) -> str | None:
1546 """The location of the most recently rejected filter."""
1547 if self._last_rejected_filter is None:
1548 return None
1549 condition, location = self._last_rejected_filter
1550 # fall back to where the predicate was defined if no location is known
1551 return location or function_location(condition)
1554def draw_choice(
1555 choice_type: ChoiceTypeT, constraints: ChoiceConstraintsT, *, random: Random
1556) -> ChoiceT:
1557 cd = ConjectureData(random=random)
1558 return cast(ChoiceT, getattr(cd.provider, f"draw_{choice_type}")(**constraints))