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