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 os
12import sys
13import threading
14import warnings
15from collections import abc, defaultdict
16from collections.abc import Callable, Sequence
17from functools import lru_cache
18from random import shuffle
19from threading import RLock
20from types import FrameType
21from typing import (
22 TYPE_CHECKING,
23 Any,
24 ClassVar,
25 Generic,
26 Literal,
27 TypeAlias,
28 TypeGuard,
29 TypeVar,
30 cast,
31 overload,
32)
34from hypothesis._settings import HealthCheck, Phase, Verbosity, settings
35from hypothesis.control import _current_build_context, current_build_context
36from hypothesis.errors import (
37 CannotInvert,
38 HypothesisException,
39 HypothesisWarning,
40 InvalidArgument,
41 NonInteractiveExampleWarning,
42 UnsatisfiedAssumption,
43)
44from hypothesis.internal.conjecture import utils as cu
45from hypothesis.internal.conjecture.choice import ChoiceT
46from hypothesis.internal.conjecture.data import ConjectureData
47from hypothesis.internal.conjecture.junkdrawer import equal_values
48from hypothesis.internal.conjecture.utils import (
49 calc_label_from_cls,
50 calc_label_from_hash,
51 calc_label_from_name,
52 combine_labels,
53)
54from hypothesis.internal.coverage import check_function
55from hypothesis.internal.escalation import is_hypothesis_file
56from hypothesis.internal.reflection import (
57 get_pretty_function_description,
58 is_identity_function,
59)
60from hypothesis.strategies._internal.utils import defines_strategy
61from hypothesis.utils.conventions import UniqueIdentifier, not_set
62from hypothesis.utils.dynamicvariables import DynamicVariable
64if TYPE_CHECKING:
65 Ex = TypeVar("Ex", covariant=True, default=Any)
66else:
67 Ex = TypeVar("Ex", covariant=True)
69T = TypeVar("T")
70T1 = TypeVar("T1")
71T3 = TypeVar("T3")
72T4 = TypeVar("T4")
73T5 = TypeVar("T5")
74MappedFrom = TypeVar("MappedFrom")
75MappedTo = TypeVar("MappedTo")
76RecurT: TypeAlias = Callable[["SearchStrategy"], bool]
77calculating = UniqueIdentifier("calculating")
79MAPPED_SEARCH_STRATEGY_DO_DRAW_LABEL = calc_label_from_name(
80 "another attempted draw in MappedStrategy"
81)
83FILTERED_SEARCH_STRATEGY_DO_DRAW_LABEL = calc_label_from_name(
84 "single loop iteration in FilteredStrategy"
85)
87label_lock = RLock()
89# When hypothesis re-applies a filter condition internally, eg in LazyStrategy, we override
90# stack walking with the correct original location here.
91_filter_location_override: DynamicVariable[Any] = DynamicVariable(not_set)
94def current_filter_call_site() -> str | None:
95 """The filename:lineno of the nearest enclosing non-hypothesis .filter() call.
97 Callers are expected to only call this from within a .filter() implementation. This
98 function simply respects _filter_location_override or walks to the nearest non-hypothesis
99 frame.
100 """
101 if (location := _filter_location_override.value) is not not_set:
102 return location
103 frame: FrameType | None = sys._getframe(1)
104 while frame is not None and is_hypothesis_file(frame.f_code.co_filename):
105 frame = frame.f_back
106 if frame is None: # pragma: no cover # ran out of frames
107 return None
108 return f"{frame.f_code.co_filename}:{frame.f_lineno}"
111def recursive_property(strategy: "SearchStrategy", name: str, default: object) -> Any:
112 """Handle properties which may be mutually recursive among a set of
113 strategies.
115 These are essentially lazily cached properties, with the ability to set
116 an override: If the property has not been explicitly set, we calculate
117 it on first access and memoize the result for later.
119 The problem is that for properties that depend on each other, a naive
120 calculation strategy may hit infinite recursion. Consider for example
121 the property is_empty. A strategy defined as x = st.deferred(lambda: x)
122 is certainly empty (in order to draw a value from x we would have to
123 draw a value from x, for which we would have to draw a value from x,
124 ...), but in order to calculate it the naive approach would end up
125 calling x.is_empty in order to calculate x.is_empty in order to etc.
127 The solution is one of fixed point calculation. We start with a default
128 value that is the value of the property in the absence of evidence to
129 the contrary, and then update the values of the property for all
130 dependent strategies until we reach a fixed point.
132 The approach taken roughly follows that in section 4.2 of Adams,
133 Michael D., Celeste Hollenbeck, and Matthew Might. "On the complexity
134 and performance of parsing with derivatives." ACM SIGPLAN Notices 51.6
135 (2016): 224-236.
136 """
137 assert name in {"is_empty", "has_reusable_values", "is_cacheable"}
138 cache_key = "cached_" + name
139 calculation = "calc_" + name
140 force_key = "force_" + name
142 def forced_or_cached_value(target: SearchStrategy) -> Any:
143 # Fast path: check the instance dict directly, which avoids raising and
144 # catching an AttributeError for the (usually absent) force_key.
145 d = target.__dict__
146 if force_key in d:
147 return d[force_key]
148 if cache_key in d:
149 return d[cache_key]
150 try:
151 return getattr(target, force_key)
152 except AttributeError:
153 return getattr(target, cache_key)
155 try:
156 return forced_or_cached_value(strategy)
157 except AttributeError:
158 pass
160 mapping: dict[SearchStrategy, Any] = {}
161 sentinel = object()
162 hit_recursion = False
164 # For a first pass we do a direct recursive calculation of the
165 # property, but we block recursively visiting a value in the
166 # computation of its property: When that happens, we simply
167 # note that it happened and return the default value.
168 def recur(strat: SearchStrategy) -> Any:
169 nonlocal hit_recursion
170 try:
171 return forced_or_cached_value(strat)
172 except AttributeError:
173 pass
174 result = mapping.get(strat, sentinel)
175 if result is calculating:
176 hit_recursion = True
177 return default
178 elif result is sentinel:
179 mapping[strat] = calculating
180 mapping[strat] = getattr(strat, calculation)(recur)
181 return mapping[strat]
182 return result
184 recur(strategy)
186 # If we hit self-recursion in the computation of any strategy
187 # value, our mapping at the end is imprecise - it may or may
188 # not have the right values in it. We now need to proceed with
189 # a more careful fixed point calculation to get the exact
190 # values. Hopefully our mapping is still pretty good and it
191 # won't take a large number of updates to reach a fixed point.
193 # We track which strategies use which in the course of
194 # calculating their property value. If A ever uses B in
195 # the course of calculating its value, then whenever the
196 # value of B changes we might need to update the value of
197 # A.
198 listeners: dict[SearchStrategy, set[SearchStrategy]] = defaultdict(set)
199 needs_update = set(mapping) if hit_recursion else None
201 def recur2(strat: SearchStrategy) -> Any:
202 def recur_inner(other: SearchStrategy) -> Any:
203 try:
204 return forced_or_cached_value(other)
205 except AttributeError:
206 pass
207 listeners[other].add(strat)
208 result = mapping.get(other, sentinel)
209 if result is sentinel:
210 assert needs_update is not None
211 needs_update.add(other)
212 mapping[other] = default
213 return default
214 return result
216 return recur_inner
218 count = 0
219 seen = set()
220 while needs_update:
221 count += 1
222 # If we seem to be taking a really long time to stabilize we
223 # start tracking seen values to attempt to detect an infinite
224 # loop. This should be impossible, and most code will never
225 # hit the count, but having an assertion for it means that
226 # testing is easier to debug and we don't just have a hung
227 # test.
228 if count > 50:
229 key = frozenset(mapping.items())
230 assert key not in seen, (key, name)
231 seen.add(key)
232 to_update = needs_update
233 needs_update = set()
234 for strat in to_update:
235 new_value = getattr(strat, calculation)(recur2(strat))
236 if new_value != mapping[strat]:
237 needs_update.update(listeners[strat])
238 mapping[strat] = new_value
240 # We now have a complete and accurate calculation of the
241 # property values for everything we have seen in the course of
242 # running this calculation. We simultaneously update all of
243 # them (not just the strategy we started out with).
244 for k, v in mapping.items():
245 setattr(k, cache_key, v)
246 # This used to simply be `getattr(strategy, cache_key)`. That relied on the invariant
247 # that our loop above has set `strategy.cached_* = v` on `strategy` if we've reached
248 # here. However, under threading, this is not necessarily true. If a concurrent thread
249 # sets `strategy.force_* = v` in between the two places we check for `force_*`, we
250 # will not set `strategy.cached_*`.
251 #
252 # There are several places where we might do this. unwrap_strategies sets
253 # force_has_reusable_values = True. our numpy.py's `arrays` strategy also does.
254 #
255 # We guard against this in general by checking the forced and cached values here,
256 # rather than just the cached value.
257 return forced_or_cached_value(strategy)
260class SearchStrategy(Generic[Ex]):
261 """A ``SearchStrategy`` tells Hypothesis how to generate that kind of input.
263 This class is only part of the public API for use in type annotations, so that
264 you can write e.g. ``-> SearchStrategy[Foo]`` for your function which returns
265 ``builds(Foo, ...)``. Do not inherit from or directly instantiate this class.
266 """
268 __module__: str = "hypothesis.strategies"
269 LABELS: ClassVar[dict[type, int]] = {}
270 # triggers `assert isinstance(label, int)` under threading when setting this
271 # in init instead of a classvar. I'm not sure why, init should be safe. But
272 # this works so I'm not looking into it further atm.
273 __label: int | UniqueIdentifier | None = None
275 def __init__(self) -> None:
276 self.validate_called: dict[int, bool] = {}
278 def is_currently_empty(self, data: ConjectureData) -> bool:
279 """
280 Returns whether this strategy is currently empty. Unlike ``empty``,
281 which is computed based on static information and cannot change,
282 ``is_currently_empty`` may change over time based on choices made
283 during the test case.
285 This is currently only used for stateful testing, where |Bundle| grows a
286 list of values to choose from over the course of a test case.
288 ``data`` will only be used for introspection. No values will be drawn
289 from it in a way that modifies the choice sequence.
290 """
291 return self.is_empty
293 @property
294 def is_empty(self) -> Any:
295 # Returns True if this strategy can never draw a value and will always
296 # result in the data being marked invalid.
297 # The fact that this returns False does not guarantee that a valid value
298 # can be drawn - this is not intended to be perfect, and is primarily
299 # intended to be an optimisation for some cases.
300 d = self.__dict__
301 if "cached_is_empty" in d and "force_is_empty" not in d:
302 return d["cached_is_empty"]
303 return recursive_property(self, "is_empty", True)
305 # Returns True if values from this strategy can safely be reused without
306 # this causing unexpected behaviour.
308 # True if values from this strategy can be implicitly reused (e.g. as
309 # background values in a numpy array) without causing surprising
310 # user-visible behaviour. Should be false for built-in strategies that
311 # produce mutable values, and for strategies that have been mapped/filtered
312 # by arbitrary user-provided functions.
313 @property
314 def has_reusable_values(self) -> Any:
315 return recursive_property(self, "has_reusable_values", True)
317 @property
318 def is_cacheable(self) -> Any:
319 """
320 Whether it is safe to hold on to instances of this strategy in a cache.
321 See _STRATEGY_CACHE.
322 """
323 return recursive_property(self, "is_cacheable", True)
325 def calc_is_cacheable(self, recur: RecurT) -> bool:
326 return True
328 def calc_is_empty(self, recur: RecurT) -> bool:
329 # Note: It is correct and significant that the default return value
330 # from calc_is_empty is False despite the default value for is_empty
331 # being true. The reason for this is that strategies should be treated
332 # as empty absent evidence to the contrary, but most basic strategies
333 # are trivially non-empty and it would be annoying to have to override
334 # this method to show that.
335 return False
337 def calc_has_reusable_values(self, recur: RecurT) -> bool:
338 return False
340 def example(self) -> Ex: # FIXME
341 """Provide an example of the sort of value that this strategy generates.
343 This method is designed for use in a REPL, and will raise an error if
344 called from inside |@given| or a strategy definition. For serious use,
345 see |@composite| or |st.data|.
346 """
347 if getattr(sys, "ps1", None) is None and (
348 # The main module's __spec__ is None when running interactively
349 # or running a source file directly.
350 # See https://docs.python.org/3/reference/import.html#main-spec.
351 sys.modules["__main__"].__spec__ is not None
352 # __spec__ is also None under pytest-xdist. To avoid an unfortunate
353 # missed alarm here, always warn under pytest.
354 or os.environ.get("PYTEST_CURRENT_TEST") is not None
355 ): # pragma: no branch
356 # The other branch *is* covered in cover/test_interactive_example.py;
357 # but as that uses `pexpect` for an interactive session `coverage`
358 # doesn't see it.
359 warnings.warn(
360 "The `.example()` method is good for exploring strategies, but should "
361 "only be used interactively. We recommend using `@given` for tests - "
362 "it performs better, saves and replays failures to avoid flakiness, "
363 f"and reports minimal failing test cases. (strategy: {self!r})",
364 NonInteractiveExampleWarning,
365 stacklevel=2,
366 )
368 context = _current_build_context.value
369 if context is not None:
370 if context.data is not None and context.data.depth > 0:
371 raise HypothesisException(
372 "Using example() inside a strategy definition is a bad "
373 "idea. Instead consider using hypothesis.strategies.builds() "
374 "or @hypothesis.strategies.composite to define your strategy."
375 " See https://hypothesis.readthedocs.io/en/latest/reference/"
376 "strategies.html#hypothesis.strategies.builds or "
377 "https://hypothesis.readthedocs.io/en/latest/reference/"
378 "strategies.html#hypothesis.strategies.composite for more "
379 "details."
380 )
381 else:
382 raise HypothesisException(
383 "Using example() inside a test function is a bad "
384 "idea. Instead consider using hypothesis.strategies.data() "
385 "to draw more values during testing. See "
386 "https://hypothesis.readthedocs.io/en/latest/reference/"
387 "strategies.html#hypothesis.strategies.data for more details."
388 )
390 try:
391 return self.__examples.pop()
392 except (AttributeError, IndexError):
393 self.__examples: list[Ex] = []
395 from hypothesis.core import given
397 # Note: this function has a weird name because it might appear in
398 # tracebacks, and we want users to know that they can ignore it.
399 @given(self)
400 @settings(
401 database=None,
402 # generate only a few examples at a time to avoid slow interactivity
403 # for large strategies. The overhead of @given is very small relative
404 # to generation, so a small batch size is fine.
405 max_examples=10,
406 deadline=None,
407 verbosity=Verbosity.quiet,
408 phases=(Phase.generate,),
409 suppress_health_check=list(HealthCheck),
410 )
411 def example_generating_inner_function(
412 ex: Ex, # type: ignore # mypy is overzealous in preventing covariant params
413 ) -> None:
414 self.__examples.append(ex)
416 example_generating_inner_function()
417 shuffle(self.__examples)
418 return self.__examples.pop()
420 def map(self, pack: Callable[[Ex], T]) -> "SearchStrategy[T]":
421 """Returns a new strategy which generates a value from this one, and
422 then returns ``pack(value)``. For example, ``integers().map(str)``
423 could generate ``str(5)`` == ``"5"``.
424 """
425 if is_identity_function(pack):
426 return self # type: ignore # Mypy has no way to know that `Ex == T`
427 return MappedStrategy(self, pack=pack)
429 def flatmap(
430 self, expand: Callable[[Ex], "SearchStrategy[T]"]
431 ) -> "SearchStrategy[T]": # FIXME
432 """Old syntax for a special case of |@composite|:
434 .. code-block:: python
436 @st.composite
437 def flatmap_like(draw, base_strategy, expand):
438 value = draw(base_strategy)
439 new_strategy = expand(value)
440 return draw(new_strategy)
442 We find that the greater readability of |@composite| usually outweighs
443 the verbosity, with a few exceptions for simple cases or recipes like
444 ``from_type(type).flatmap(from_type)`` ("pick a type, get a strategy for
445 any instance of that type, and then generate one of those").
446 """
447 from hypothesis.strategies._internal.flatmapped import FlatMapStrategy
449 return FlatMapStrategy(self, expand=expand)
451 # Note that we previously had condition extracted to a type alias as
452 # PredicateT. However, that was only useful when not specifying a relationship
453 # between the generic Ts and some other function param / return value.
454 # If we do want to - like here, where we want to say that the Ex arg to condition
455 # is of the same type as the strategy's Ex - then you need to write out the
456 # entire Callable[[Ex], Any] expression rather than use a type alias.
457 # TypeAlias is *not* simply a macro that inserts the text. TypeAlias will not
458 # reference the local TypeVar context.
459 @overload
460 def filter(
461 self, condition: Callable[[Ex], TypeGuard[T]]
462 ) -> "SearchStrategy[T]": ...
463 @overload
464 def filter(self, condition: Callable[[Ex], Any]) -> "SearchStrategy[Ex]": ...
465 def filter(self, condition):
466 """Returns a new strategy that generates values from this strategy
467 which satisfy the provided condition.
469 Note that if the condition is too hard to satisfy this might result
470 in your tests failing with an Unsatisfiable exception.
471 A basic version of the filtering logic would look something like:
473 .. code-block:: python
475 @st.composite
476 def filter_like(draw, strategy, condition):
477 for _ in range(3):
478 value = draw(strategy)
479 if condition(value):
480 return value
481 assume(False)
482 """
483 return FilteredStrategy(
484 self,
485 conditions=(condition,),
486 condition_locations=(current_filter_call_site(),),
487 )
489 @property
490 def branches(self) -> Sequence["SearchStrategy[Ex]"]:
491 return [self]
493 def __or__(self, other: "SearchStrategy[T]") -> "SearchStrategy[Ex | T]":
494 """Return a strategy which produces values by randomly drawing from one
495 of this strategy or the other strategy.
497 This method is part of the public API.
498 """
499 if not isinstance(other, SearchStrategy):
500 raise ValueError(f"Cannot | a SearchStrategy with {other!r}")
502 # Unwrap explicitly or'd strategies. This turns the
503 # common case of e.g. st.integers() | st.integers() | st.integers() from
504 #
505 # one_of(one_of(integers(), integers()), integers())
506 #
507 # into
508 #
509 # one_of(integers(), integers(), integers())
510 #
511 # This is purely an aesthetic unwrapping, for e.g. reprs. In practice
512 # we use .branches / .element_strategies to get the list of possible
513 # strategies, so this unwrapping is *not* necessary for correctness.
514 strategies: list[SearchStrategy] = []
515 strategies.extend(
516 self.original_strategies if isinstance(self, OneOfStrategy) else [self]
517 )
518 strategies.extend(
519 other.original_strategies if isinstance(other, OneOfStrategy) else [other]
520 )
521 return OneOfStrategy(strategies)
523 def __bool__(self) -> bool:
524 warnings.warn(
525 f"bool({self!r}) is always True, did you mean to draw a value?",
526 HypothesisWarning,
527 stacklevel=2,
528 )
529 return True
531 def validate(self) -> None:
532 """Throw an exception if the strategy is not valid.
534 Strategies should implement ``do_validate``, which is called by this
535 method. They should not override ``validate``.
537 This can happen due to invalid arguments, or lazy construction.
538 """
539 thread_id = threading.get_ident()
540 if self.validate_called.get(thread_id, False):
541 return
542 # we need to set validate_called before calling do_validate, for
543 # recursive / deferred strategies. But if a thread switches after
544 # validate_called but before do_validate, we might have a strategy
545 # which does weird things like drawing when do_validate would error but
546 # its params are technically valid (e.g. a param was passed as 1.0
547 # instead of 1) and get into weird internal states.
548 #
549 # There are two ways to fix this.
550 # (1) The first is a per-strategy lock around do_validate. Even though we
551 # expect near-zero lock contention, this still adds the lock overhead.
552 # (2) The second is allowing concurrent .validate calls. Since validation
553 # is (assumed to be) deterministic, both threads will produce the same
554 # end state, so the validation order or race conditions does not matter.
555 #
556 # In order to avoid the lock overhead of (1), we use (2) here. See also
557 # discussion in https://github.com/HypothesisWorks/hypothesis/pull/4473.
558 try:
559 self.validate_called[thread_id] = True
560 self.do_validate()
561 self.is_empty
562 self.has_reusable_values
563 except Exception:
564 self.validate_called[thread_id] = False
565 raise
567 @property
568 def class_label(self) -> int:
569 cls = self.__class__
570 try:
571 return cls.LABELS[cls]
572 except KeyError:
573 pass
574 result = calc_label_from_cls(cls)
575 cls.LABELS[cls] = result
576 return result
578 @property
579 def label(self) -> int:
580 if isinstance((label := self.__label), int):
581 # avoid locking if we've already completely computed the label.
582 return label
584 with label_lock:
585 if self.__label is calculating:
586 return 0
587 self.__label = calculating
588 self.__label = self.calc_label()
589 return self.__label
591 def calc_label(self) -> int:
592 return self.class_label
594 def do_validate(self) -> None:
595 pass
597 def do_draw(self, data: ConjectureData) -> Ex:
598 raise NotImplementedError(f"{type(self).__name__}.do_draw")
600 def _invert(self, value: Any) -> tuple[ChoiceT, ...]:
601 """
602 Return a choice sequence ``choices`` such that we expect
604 data = ConjectureData.for_choices(choices)
605 drawn = data.draw(self)
607 would produce ``drawn == value``. Inversion is best-effort: it should
608 satisfy this property with high probability, but callers must replay
609 the returned choices and check the outcome rather than rely on it.
611 Raises CannotInvert if we cannot construct such a choice sequence,
612 whether because ``value`` is not produced by this strategy or because
613 inversion isn't implemented for it.
614 """
615 raise CannotInvert(f"{type(self).__name__} does not support inversion")
618def _is_hashable(value: object) -> tuple[bool, int | None]:
619 # hashing can be expensive; return the hash value if we compute it, so that
620 # callers don't have to recompute.
621 try:
622 return (True, hash(value))
623 except TypeError:
624 return (False, None)
627def is_hashable(value: object) -> bool:
628 return _is_hashable(value)[0]
631class SampledFromStrategy(SearchStrategy[Ex]):
632 """A strategy which samples from a set of elements. This is essentially
633 equivalent to using a OneOfStrategy over Just strategies but may be more
634 efficient and convenient.
635 """
637 _MAX_FILTER_CALLS: ClassVar[int] = 10_000
639 def __init__(
640 self,
641 elements: Sequence[Ex],
642 *,
643 force_repr: str | None = None,
644 force_repr_braces: tuple[str, str] | None = None,
645 # (name, function, location of the .filter()/.map() call, if known)
646 transformations: tuple[
647 tuple[Literal["filter", "map"], Callable[[Ex], Any], str | None],
648 ...,
649 ] = (),
650 ):
651 super().__init__()
652 self.elements = cu.check_sample(elements, "sampled_from")
653 assert self.elements
654 self.force_repr = force_repr
655 self.force_repr_braces = force_repr_braces
656 self._transformations = transformations
658 self._cached_repr: str | None = None
660 def map(self, pack: Callable[[Ex], T]) -> SearchStrategy[T]:
661 s = type(self)(
662 self.elements,
663 force_repr=self.force_repr,
664 force_repr_braces=self.force_repr_braces,
665 transformations=(*self._transformations, ("map", pack, None)),
666 )
667 # guaranteed by the ("map", pack) transformation
668 return cast(SearchStrategy[T], s)
670 @overload
671 def filter(
672 self, condition: Callable[[Ex], TypeGuard[T]]
673 ) -> "SearchStrategy[T]": ...
674 @overload
675 def filter(self, condition: Callable[[Ex], Any]) -> "SearchStrategy[Ex]": ...
676 def filter(self, condition):
677 return type(self)(
678 self.elements,
679 force_repr=self.force_repr,
680 force_repr_braces=self.force_repr_braces,
681 transformations=(
682 *self._transformations,
683 ("filter", condition, current_filter_call_site()),
684 ),
685 )
687 def __repr__(self):
688 if self._cached_repr is None:
689 rep = get_pretty_function_description
690 elements_s = (
691 ", ".join(rep(v) for v in self.elements[:512]) + ", ..."
692 if len(self.elements) > 512
693 else ", ".join(rep(v) for v in self.elements)
694 )
695 braces = self.force_repr_braces or ("(", ")")
696 instance_s = (
697 self.force_repr or f"sampled_from({braces[0]}{elements_s}{braces[1]})"
698 )
699 transforms_s = "".join(
700 f".{name}({get_pretty_function_description(f)})"
701 for name, f, _ in self._transformations
702 )
703 repr_s = instance_s + transforms_s
704 self._cached_repr = repr_s
705 return self._cached_repr
707 def calc_label(self) -> int:
708 # strategy.label is effectively an under-approximation of structural
709 # equality (i.e., some strategies may have the same label when they are not
710 # structurally identical). More importantly for calculating the
711 # SampledFromStrategy label, we might have hash(s1) != hash(s2) even
712 # when s1 and s2 are structurally identical. For instance:
713 #
714 # s1 = st.sampled_from([st.none()])
715 # s2 = st.sampled_from([st.none()])
716 # assert hash(s1) != hash(s2)
717 #
718 # (see also test cases in test_labels.py).
719 #
720 # We therefore use the labels of any component strategies when calculating
721 # our label, and only use the hash if it is not a strategy.
722 #
723 # That's the ideal, anyway. In reality the logic is more complicated than
724 # necessary in order to be efficient in the presence of (very) large sequences:
725 # * add an unabashed special case for range, to avoid iteration over an
726 # enormous range when we know it is entirely integers.
727 # * if there is at least one strategy in self.elements, use strategy label,
728 # and the element hash otherwise.
729 # * if there are no strategies in self.elements, take the hash of the
730 # entire sequence. This prevents worst-case performance of hashing each
731 # element when a hash of the entire sequence would have sufficed.
732 #
733 # The worst case performance of this scheme is
734 # itertools.chain(range(2**100), [st.none()]), where it degrades to
735 # hashing every int in the range.
736 elements_is_hashable, hash_value = _is_hashable(self.elements)
737 if isinstance(self.elements, range) or (
738 elements_is_hashable
739 and not any(isinstance(e, SearchStrategy) for e in self.elements)
740 ):
741 return combine_labels(
742 self.class_label, calc_label_from_name(str(hash_value))
743 )
745 labels = [self.class_label]
746 for element in self.elements:
747 if not is_hashable(element):
748 continue
750 labels.append(
751 element.label
752 if isinstance(element, SearchStrategy)
753 else calc_label_from_hash(element)
754 )
756 return combine_labels(*labels)
758 def calc_has_reusable_values(self, recur: RecurT) -> bool:
759 # Because our custom .map/.filter implementations skip the normal
760 # wrapper strategies (which would automatically return False for us),
761 # we need to manually return False here if any transformations have
762 # been applied.
763 return not self._transformations
765 def calc_is_cacheable(self, recur: RecurT) -> bool:
766 return is_hashable(self.elements)
768 def _transform(
769 self,
770 # https://github.com/python/mypy/issues/7049, we're not writing `element`
771 # anywhere in the class so this is still type-safe. mypy is being more
772 # conservative than necessary
773 element: Ex, # type: ignore
774 *,
775 # None for _invert, which has no ConjectureData
776 data: ConjectureData | None,
777 ) -> Ex | UniqueIdentifier:
778 # Used in UniqueSampledListStrategy
779 for name, f, location in self._transformations:
780 if name == "map":
781 result = f(element)
782 if build_context := _current_build_context.value:
783 build_context.record_call(result, f, args=[element], kwargs={})
784 element = result
785 else:
786 assert name == "filter"
787 if not f(element):
788 if data is not None:
789 data._last_rejected_filter = (f, location)
790 return filter_not_satisfied
791 return element
793 def do_draw(self, data: ConjectureData) -> Ex:
794 result = self.do_filtered_draw(data)
795 if isinstance(result, SearchStrategy) and all(
796 isinstance(x, SearchStrategy) for x in self.elements
797 ):
798 data._sampled_from_all_strategies_elements_message = (
799 (
800 "sampled_from was given a collection of strategies: "
801 "{!r}. Was one_of intended?"
802 ),
803 self.elements,
804 )
805 if result is filter_not_satisfied:
806 # do_filtered_draw records data._last_rejected_filter.
807 data.mark_invalid(
808 f"Aborted test because unable to satisfy {self!r}",
809 location=data.last_rejected_filter_location(),
810 )
811 assert not isinstance(result, UniqueIdentifier)
812 return result
814 def get_element(self, i: int, data: ConjectureData) -> Ex | UniqueIdentifier:
815 return self._transform(self.elements[i], data=data)
817 def _invert(self, value: Any) -> tuple[ChoiceT, ...]:
818 # The smallest index whose (possibly transformed) element equals value.
819 # _transform might depend on external state and give us a wrong answer
820 # here; that's fine, since _invert is allowed to be fallible.
821 try:
822 for i, element in enumerate(self.elements):
823 if equal_values(self._transform(element, data=None), value):
824 return (i,)
825 except Exception:
826 # `self._transform` may include arbitrary user code (map/filter functions)
827 raise CannotInvert(f"transforming elements of {self!r} errored") from None
828 raise CannotInvert(f"{value!r} is not produced by {self!r}")
830 def do_filtered_draw(self, data: ConjectureData) -> Ex | UniqueIdentifier:
831 # Set of indices that have been tried so far, so that we never test
832 # the same element twice during a draw.
833 known_bad_indices: set[int] = set()
835 # Start with ordinary rejection sampling. It's fast if it works, and
836 # if it doesn't work then it was only a small amount of overhead.
837 for _ in range(3):
838 i = data.draw_integer(0, len(self.elements) - 1)
839 if i not in known_bad_indices:
840 element = self.get_element(i, data)
841 if element is not filter_not_satisfied:
842 return element
843 if not known_bad_indices:
844 data.events[f"Retried draw from {self!r} to satisfy filter"] = ""
845 known_bad_indices.add(i)
847 # If we've tried all the possible elements, give up now.
848 max_good_indices = len(self.elements) - len(known_bad_indices)
849 if not max_good_indices:
850 return filter_not_satisfied
852 # Impose an arbitrary cutoff to prevent us from wasting too much time
853 # on very large element lists.
854 max_good_indices = min(max_good_indices, self._MAX_FILTER_CALLS - 3)
856 # Before building the list of allowed indices, speculatively choose
857 # one of them. We don't yet know how many allowed indices there will be,
858 # so this choice might be out-of-bounds, but that's OK.
859 speculative_index = data.draw_integer(0, max_good_indices - 1)
861 # Calculate the indices of allowed values, so that we can choose one
862 # of them at random. But if we encounter the speculatively-chosen one,
863 # just use that and return immediately. Note that we also track the
864 # allowed elements, in case of .map(some_stateful_function)
865 allowed: list[tuple[int, Ex]] = []
866 for i in range(min(len(self.elements), self._MAX_FILTER_CALLS - 3)):
867 if i not in known_bad_indices:
868 element = self.get_element(i, data)
869 if element is not filter_not_satisfied:
870 assert not isinstance(element, UniqueIdentifier)
871 allowed.append((i, element))
872 if len(allowed) > speculative_index:
873 # Early-exit case: We reached the speculative index, so
874 # we just return the corresponding element.
875 data.draw_integer(0, len(self.elements) - 1, forced=i)
876 return element
878 # The speculative index didn't work out, but at this point we've built
879 # and can choose from the complete list of allowed indices and elements.
880 if allowed:
881 i, element = data.choice(allowed)
882 data.draw_integer(0, len(self.elements) - 1, forced=i)
883 return element
884 # If there are no allowed indices, the filter couldn't be satisfied.
885 return filter_not_satisfied
888# The ids of OneOfStrategy instances an _invert call is currently walking
889# through, per thread. See OneOfStrategy._invert.
890_inverting_one_ofs = threading.local()
893class OneOfStrategy(SearchStrategy[Ex]):
894 """Implements a union of strategies. Given a number of strategies this
895 generates values which could have come from any of them.
897 The conditional distribution draws uniformly at random from some
898 non-empty subset of these strategies and then draws from the
899 conditional distribution of that strategy.
900 """
902 def __init__(self, strategies: Sequence[SearchStrategy[Ex]]):
903 super().__init__()
904 self.original_strategies = tuple(strategies)
905 self.__element_strategies: Sequence[SearchStrategy[Ex]] | None = None
906 self.__in_branches = False
907 self.__branch_sampler: SampledFromStrategy | Literal[False] | None = None
908 self._branches_lock = RLock()
910 def calc_is_empty(self, recur: RecurT) -> bool:
911 return all(recur(e) for e in self.original_strategies)
913 def calc_has_reusable_values(self, recur: RecurT) -> bool:
914 return all(recur(e) for e in self.original_strategies)
916 def calc_is_cacheable(self, recur: RecurT) -> bool:
917 return all(recur(e) for e in self.original_strategies)
919 @property
920 def element_strategies(self) -> Sequence[SearchStrategy[Ex]]:
921 if self.__element_strategies is None:
922 # While strategies are hashable, they use object.__hash__ and are
923 # therefore distinguished only by identity.
924 #
925 # In principle we could "just" define a __hash__ method
926 # (and __eq__, but that's easy in terms of type() and hash())
927 # to make this more powerful, but this is harder than it sounds:
928 #
929 # 1. Strategies are often distinguished by non-hashable attributes,
930 # or by attributes that have the same hash value ("^.+" / b"^.+").
931 # 2. LazyStrategy: can't reify the wrapped strategy without breaking
932 # laziness, so there's a hash each for the lazy and the nonlazy.
933 #
934 # Having made several attempts, the minor benefits of making strategies
935 # hashable are simply not worth the engineering effort it would take.
936 # See also issues #2291 and #2327.
937 seen: set[SearchStrategy] = {self}
938 strategies: list[SearchStrategy] = []
939 for arg in self.original_strategies:
940 check_strategy(arg)
941 if not arg.is_empty:
942 for s in arg.branches:
943 if s not in seen and not s.is_empty:
944 seen.add(s)
945 strategies.append(s)
946 self.__element_strategies = strategies
947 return self.__element_strategies
949 def calc_label(self) -> int:
950 return combine_labels(
951 self.class_label, *(p.label for p in self.original_strategies)
952 )
954 def do_draw(self, data: ConjectureData) -> Ex:
955 strategies = self.element_strategies
956 if len(strategies) == 1:
957 # optimization: skip constructing SampledFromStrategy if we only have one
958 # strategy. This can happen for eg `st.integers() | st.nothing()`.
959 return data.draw(strategies[0])
961 if self.__branch_sampler is None:
962 # Only strategies which override is_currently_empty (e.g. stateful
963 # Bundles) can become empty dynamically - otherwise element_strategies
964 # has already excluded every empty branch, so we can skip the filter
965 # and reuse a single sampler rather than building one per draw.
966 if any(
967 type(s).is_currently_empty is not SearchStrategy.is_currently_empty
968 for s in strategies
969 ):
970 self.__branch_sampler = False
971 else:
972 self.__branch_sampler = SampledFromStrategy(strategies)
973 if self.__branch_sampler is not False:
974 return data.draw(data.draw(self.__branch_sampler))
976 strategy = data.draw(
977 SampledFromStrategy(strategies).filter(
978 lambda s: not s.is_currently_empty(data)
979 )
980 )
981 return data.draw(strategy)
983 def _invert(self, value: Any) -> tuple[ChoiceT, ...]:
984 # do_draw first draws a branch index, then draws from that branch.
985 # Return the simplest candidate across all branches: the shortest
986 # encoding, breaking ties by the lower branch index - matching the
987 # shrinker's ordering over choice sequences.
988 #
989 # Self-referential strategies (e.g. via st.deferred) can route a
990 # branch's inversion of the same value back to this one_of, so guard
991 # against re-entry rather than recursing forever. Unlike a structural
992 # recursion into a subvalue - which terminates because values are
993 # finite - a same-value cycle can make no progress.
994 if len(self.element_strategies) == 1:
995 return self.element_strategies[0]._invert(value)
997 active = getattr(_inverting_one_ofs, "ids", None)
998 if active is None:
999 active = _inverting_one_ofs.ids = set()
1000 if id(self) in active:
1001 raise CannotInvert(f"recursive inversion of {self!r}")
1002 active.add(id(self))
1003 try:
1004 best: tuple[ChoiceT, ...] | None = None
1005 for i, branch in enumerate(self.element_strategies):
1006 try:
1007 candidate = (i, *branch._invert(value))
1008 except CannotInvert:
1009 continue
1010 if best is None or len(candidate) < len(best):
1011 best = candidate
1012 if len(best) <= 2:
1013 # A selector plus at most one choice: no later branch is
1014 # worth trying. (A zero-choice just()-like branch could
1015 # still encode in one choice, but we accept two rather
1016 # than scanning - possibly recursively - for it.)
1017 break
1018 finally:
1019 active.discard(id(self))
1020 if best is None:
1021 raise CannotInvert(f"{value!r} is not produced by any branch of {self!r}")
1022 return best
1024 def __repr__(self) -> str:
1025 return "one_of({})".format(", ".join(map(repr, self.original_strategies)))
1027 def do_validate(self) -> None:
1028 for e in self.element_strategies:
1029 e.validate()
1031 @property
1032 def branches(self) -> Sequence[SearchStrategy[Ex]]:
1033 if self.__element_strategies is not None:
1034 # common fast path which avoids the lock
1035 return self.element_strategies
1037 with self._branches_lock:
1038 if not self.__in_branches:
1039 try:
1040 self.__in_branches = True
1041 return self.element_strategies
1042 finally:
1043 self.__in_branches = False
1044 else:
1045 return [self]
1047 @overload
1048 def filter(
1049 self, condition: Callable[[Ex], TypeGuard[T]]
1050 ) -> "SearchStrategy[T]": ...
1051 @overload
1052 def filter(self, condition: Callable[[Ex], Any]) -> "SearchStrategy[Ex]": ...
1053 def filter(self, condition):
1054 return FilteredStrategy(
1055 OneOfStrategy([s.filter(condition) for s in self.original_strategies]),
1056 conditions=(),
1057 )
1060@overload
1061def one_of(
1062 __args: Sequence[SearchStrategy[Ex]],
1063) -> SearchStrategy[Ex]: ...
1066@overload
1067def one_of(__a1: SearchStrategy[Ex]) -> SearchStrategy[Ex]: ...
1070@overload
1071def one_of(
1072 __a1: SearchStrategy[T1], __a2: SearchStrategy[T]
1073) -> SearchStrategy[T1 | T]: ...
1076@overload
1077def one_of(
1078 __a1: SearchStrategy[T1], __a2: SearchStrategy[T], __a3: SearchStrategy[T3]
1079) -> SearchStrategy[T1 | T | T3]: ...
1082@overload
1083def one_of(
1084 __a1: SearchStrategy[T1],
1085 __a2: SearchStrategy[T],
1086 __a3: SearchStrategy[T3],
1087 __a4: SearchStrategy[T4],
1088) -> SearchStrategy[T1 | T | T3 | T4]: ...
1091@overload
1092def one_of(
1093 __a1: SearchStrategy[T1],
1094 __a2: SearchStrategy[T],
1095 __a3: SearchStrategy[T3],
1096 __a4: SearchStrategy[T4],
1097 __a5: SearchStrategy[T5],
1098) -> SearchStrategy[T1 | T | T3 | T4 | T5]: ...
1101@overload
1102def one_of(*args: SearchStrategy[Any]) -> SearchStrategy[Any]: ...
1105@defines_strategy(eager=True)
1106def one_of(
1107 *args: Sequence[SearchStrategy[Any]] | SearchStrategy[Any],
1108) -> SearchStrategy[Any]:
1109 # Mypy workaround alert: Any is too loose above; the return parameter
1110 # should be the union of the input parameters. Unfortunately, Mypy <=0.600
1111 # raises errors due to incompatible inputs instead. See #1270 for links.
1112 # v0.610 doesn't error; it gets inference wrong for 2+ arguments instead.
1113 """Return a strategy which generates values from any of the argument
1114 strategies.
1116 This may be called with one iterable argument instead of multiple
1117 strategy arguments, in which case ``one_of(x)`` and ``one_of(*x)`` are
1118 equivalent.
1120 Examples from this strategy will generally shrink to ones that come from
1121 strategies earlier in the list, then shrink according to behaviour of the
1122 strategy that produced them. In order to get good shrinking behaviour,
1123 try to put simpler strategies first. e.g. ``one_of(none(), text())`` is
1124 better than ``one_of(text(), none())``.
1126 This is especially important when using recursive strategies. e.g.
1127 ``x = st.deferred(lambda: st.none() | st.tuples(x, x))`` will shrink well,
1128 but ``x = st.deferred(lambda: st.tuples(x, x) | st.none())`` will shrink
1129 very badly indeed.
1130 """
1131 if len(args) == 1 and not isinstance(args[0], SearchStrategy):
1132 try:
1133 args = tuple(args[0])
1134 except TypeError:
1135 pass
1136 if len(args) == 1 and isinstance(args[0], SearchStrategy):
1137 # This special-case means that we can one_of over lists of any size
1138 # without incurring any performance overhead when there is only one
1139 # strategy, and keeps our reprs simple.
1140 return args[0]
1141 if args and not any(isinstance(a, SearchStrategy) for a in args):
1142 # And this special case is to give a more-specific error message if it
1143 # seems that the user has confused `one_of()` for `sampled_from()`;
1144 # the remaining validation is left to OneOfStrategy. See PR #2627.
1145 raise InvalidArgument(
1146 f"Did you mean st.sampled_from({list(args)!r})? st.one_of() is used "
1147 "to combine strategies, but all of the arguments were of other types."
1148 )
1149 # we've handled the case where args is a one-element sequence [(s1, s2, ...)]
1150 # above, so we can assume it's an actual sequence of strategies.
1151 return OneOfStrategy(cast(Sequence[SearchStrategy], args))
1154class MappedStrategy(SearchStrategy[MappedTo], Generic[MappedFrom, MappedTo]):
1155 """A strategy which is defined purely by conversion to and from another
1156 strategy.
1158 Its parameter and distribution come from that other strategy.
1159 """
1161 def __init__(
1162 self,
1163 strategy: SearchStrategy[MappedFrom],
1164 pack: Callable[[MappedFrom], MappedTo],
1165 ) -> None:
1166 super().__init__()
1167 self.mapped_strategy = strategy
1168 self.pack = pack
1170 def calc_is_empty(self, recur: RecurT) -> bool:
1171 return recur(self.mapped_strategy)
1173 def calc_is_cacheable(self, recur: RecurT) -> bool:
1174 return recur(self.mapped_strategy)
1176 def __repr__(self) -> str:
1177 if not hasattr(self, "_cached_repr"):
1178 self._cached_repr = f"{self.mapped_strategy!r}.map({get_pretty_function_description(self.pack)})"
1179 return self._cached_repr
1181 def do_validate(self) -> None:
1182 self.mapped_strategy.validate()
1184 def do_draw(self, data: ConjectureData) -> MappedTo:
1185 with warnings.catch_warnings():
1186 if isinstance(self.pack, type) and issubclass(
1187 self.pack, (abc.Mapping, abc.Set)
1188 ):
1189 warnings.simplefilter("ignore", BytesWarning)
1190 for _ in range(3):
1191 data.start_span(MAPPED_SEARCH_STRATEGY_DO_DRAW_LABEL)
1192 discard = False
1193 try:
1194 x = data.draw(self.mapped_strategy)
1195 result = self.pack(x)
1196 except UnsatisfiedAssumption as err:
1197 # we want to preserve the err.location of the actual exception here if we
1198 # re-throw it outside of the loop at the end. The alternative is a
1199 # bare `raise UnsatisfiedAssumption` outside of the loop, which would
1200 # drop the real err.location for observability.
1201 #
1202 # If we catch multiple exceptions here, we'll just report the location
1203 # of the last one, which is a reasonable tradeoff for a singleton field.
1204 failed_assumption = err
1205 discard = True
1206 continue
1207 finally:
1208 data.stop_span(discard=discard)
1209 current_build_context().record_call(
1210 result, self.pack, args=[x], kwargs={}
1211 )
1212 return result
1213 raise failed_assumption
1215 def _invert(self, value: Any) -> tuple[ChoiceT, ...]:
1216 # map() is not invertible in general, but a dict-like pack - e.g. the
1217 # dict_class which st.dictionaries maps over a unique list of
1218 # (key, value) tuples - inverts as its list of items.
1219 if (
1220 isinstance(self.pack, type)
1221 and issubclass(self.pack, abc.Mapping)
1222 and isinstance(value, self.pack)
1223 ):
1224 return self.mapped_strategy._invert(list(value.items()))
1225 raise CannotInvert(f"cannot invert {self!r} (value={value!r})")
1227 @property
1228 def branches(self) -> Sequence[SearchStrategy[MappedTo]]:
1229 return [
1230 MappedStrategy(strategy, pack=self.pack)
1231 for strategy in self.mapped_strategy.branches
1232 ]
1234 @overload
1235 def filter(
1236 self, condition: Callable[[MappedTo], TypeGuard[T]]
1237 ) -> "SearchStrategy[T]": ...
1238 @overload
1239 def filter(
1240 self, condition: Callable[[MappedTo], Any]
1241 ) -> "SearchStrategy[MappedTo]": ...
1242 def filter(self, condition):
1243 # Includes a special case so that we can rewrite filters on collection
1244 # lengths, when most collections are `st.lists(...).map(the_type)`.
1245 ListStrategy = _list_strategy_type()
1246 if not isinstance(self.mapped_strategy, ListStrategy) or not (
1247 (isinstance(self.pack, type) and issubclass(self.pack, abc.Collection))
1248 or self.pack in _collection_ish_functions()
1249 ):
1250 return super().filter(condition)
1252 # Check whether our inner list strategy can rewrite this filter condition.
1253 # If not, discard the result and _only_ apply a new outer filter.
1254 new = ListStrategy.filter(self.mapped_strategy, condition)
1255 if getattr(new, "filtered_strategy", None) is self.mapped_strategy:
1256 return super().filter(condition) # didn't rewrite
1258 # Apply a new outer filter even though we rewrote the inner strategy,
1259 # because some collections can change the list length (dict, set, etc).
1260 return FilteredStrategy(
1261 type(self)(new, self.pack),
1262 conditions=(condition,),
1263 condition_locations=(current_filter_call_site(),),
1264 )
1267@lru_cache
1268def _list_strategy_type() -> Any:
1269 from hypothesis.strategies._internal.collections import ListStrategy
1271 return ListStrategy
1274def _collection_ish_functions() -> Sequence[Any]:
1275 funcs = [sorted]
1276 if np := sys.modules.get("numpy"):
1277 # c.f. https://numpy.org/doc/stable/reference/routines.array-creation.html
1278 # Probably only `np.array` and `np.asarray` will be used in practice,
1279 # but why should that stop us when we've already gone this far?
1280 funcs += [
1281 np.empty_like,
1282 np.eye,
1283 np.identity,
1284 np.ones_like,
1285 np.zeros_like,
1286 np.array,
1287 np.asarray,
1288 np.asanyarray,
1289 np.ascontiguousarray,
1290 np.asmatrix,
1291 np.copy,
1292 np.rec.array,
1293 np.rec.fromarrays,
1294 np.rec.fromrecords,
1295 np.diag,
1296 # bonus undocumented functions from tab-completion:
1297 np.asarray_chkfinite,
1298 np.asfortranarray,
1299 ]
1301 return funcs
1304filter_not_satisfied = UniqueIdentifier("filter not satisfied")
1307class FilteredStrategy(SearchStrategy[Ex]):
1308 def __init__(
1309 self,
1310 strategy: SearchStrategy[Ex],
1311 conditions: tuple[Callable[[Ex], Any], ...],
1312 condition_locations: tuple[str | None, ...] | None = None,
1313 ):
1314 super().__init__()
1315 # Where each condition's .filter() call was made, for observability
1316 if condition_locations is None:
1317 condition_locations = (None,) * len(conditions)
1318 assert len(condition_locations) == len(conditions)
1319 if isinstance(strategy, FilteredStrategy):
1320 # Flatten chained filters into a single filter with multiple conditions.
1321 self.flat_conditions: tuple[Callable[[Ex], Any], ...] = (
1322 strategy.flat_conditions + conditions
1323 )
1324 self.condition_locations: tuple[str | None, ...] = (
1325 strategy.condition_locations + condition_locations
1326 )
1327 self.filtered_strategy: SearchStrategy[Ex] = strategy.filtered_strategy
1328 else:
1329 self.flat_conditions = conditions
1330 self.condition_locations = condition_locations
1331 self.filtered_strategy = strategy
1333 assert isinstance(self.flat_conditions, tuple)
1334 assert not isinstance(self.filtered_strategy, FilteredStrategy)
1336 self.__condition: Callable[[Ex], Any] | None = None
1338 def calc_is_empty(self, recur: RecurT) -> bool:
1339 return recur(self.filtered_strategy)
1341 def calc_is_cacheable(self, recur: RecurT) -> bool:
1342 return recur(self.filtered_strategy)
1344 def __repr__(self) -> str:
1345 if not hasattr(self, "_cached_repr"):
1346 self._cached_repr = "{!r}{}".format(
1347 self.filtered_strategy,
1348 "".join(
1349 f".filter({get_pretty_function_description(cond)})"
1350 for cond in self.flat_conditions
1351 ),
1352 )
1353 return self._cached_repr
1355 def do_validate(self) -> None:
1356 # Start by validating our inner filtered_strategy. If this was a LazyStrategy,
1357 # validation also reifies it so that subsequent calls to e.g. `.filter()` will
1358 # be passed through.
1359 self.filtered_strategy.validate()
1360 # So now we have a reified inner strategy, we'll replay all our saved
1361 # predicates in case some or all of them can be rewritten. Note that this
1362 # replaces the `fresh` strategy too!
1363 fresh = self.filtered_strategy
1364 for cond, location in zip(
1365 self.flat_conditions, self.condition_locations, strict=True
1366 ):
1367 with _filter_location_override.with_value(location):
1368 fresh = fresh.filter(cond)
1369 if isinstance(fresh, FilteredStrategy):
1370 # In this case we have at least some non-rewritten filter predicates,
1371 # so we just re-initialize the strategy.
1372 FilteredStrategy.__init__(
1373 self,
1374 fresh.filtered_strategy,
1375 fresh.flat_conditions,
1376 fresh.condition_locations,
1377 )
1378 else:
1379 # But if *all* the predicates were rewritten... well, do_validate() is
1380 # an in-place method so we still just re-initialize the strategy!
1381 FilteredStrategy.__init__(self, fresh, ())
1383 @overload
1384 def filter(
1385 self, condition: Callable[[Ex], TypeGuard[T]]
1386 ) -> "FilteredStrategy[T]": ...
1387 @overload
1388 def filter(self, condition: Callable[[Ex], Any]) -> "FilteredStrategy[Ex]": ...
1389 def filter(self, condition):
1390 # If we can, it's more efficient to rewrite our strategy to satisfy the
1391 # condition. We therefore exploit the fact that the order of predicates
1392 # doesn't matter (`f(x) and g(x) == g(x) and f(x)`) by attempting to apply
1393 # condition directly to our filtered strategy as the inner-most filter.
1394 out = self.filtered_strategy.filter(condition)
1395 # If it couldn't be rewritten, we'll get a new FilteredStrategy - and then
1396 # combine the conditions of each in our expected newest=last order.
1397 if isinstance(out, FilteredStrategy):
1398 return FilteredStrategy(
1399 out.filtered_strategy,
1400 self.flat_conditions + out.flat_conditions,
1401 self.condition_locations + out.condition_locations,
1402 )
1403 # But if it *could* be rewritten, we can return the more efficient form!
1404 return FilteredStrategy(out, self.flat_conditions, self.condition_locations)
1406 @property
1407 def condition(self) -> Callable[[Ex], Any]:
1408 # We write this defensively to avoid any threading race conditions
1409 # with our manual FilteredStrategy.__init__ for filter-rewriting.
1410 # See https://github.com/HypothesisWorks/hypothesis/pull/4522.
1411 if (condition := self.__condition) is not None:
1412 return condition
1414 if len(self.flat_conditions) == 1:
1415 # Avoid an extra indirection in the common case of only one condition.
1416 condition = self.flat_conditions[0]
1417 elif len(self.flat_conditions) == 0:
1418 # Possible, if unlikely, due to filter predicate rewriting
1419 condition = lambda _: True
1420 else:
1421 condition = lambda x: all(cond(x) for cond in self.flat_conditions)
1422 self.__condition = condition
1423 return condition
1425 def do_draw(self, data: ConjectureData) -> Ex:
1426 result = self.do_filtered_draw(data)
1427 if result is not filter_not_satisfied:
1428 return cast(Ex, result)
1430 # do_filtered_draw records data._last_rejected_filter.
1431 data.mark_invalid(
1432 f"Aborted test because unable to satisfy {self!r}",
1433 location=data.last_rejected_filter_location(),
1434 )
1436 def do_filtered_draw(self, data: ConjectureData) -> Ex | UniqueIdentifier:
1437 for i in range(3):
1438 data.start_span(FILTERED_SEARCH_STRATEGY_DO_DRAW_LABEL)
1439 failing = None
1440 try:
1441 value = data.draw(self.filtered_strategy)
1442 # Check the conditions individually rather than via self.condition,
1443 # so that we can set data._last_rejected_filter.
1444 failing = next(
1445 (
1446 (cond, location)
1447 for cond, location in zip(
1448 self.flat_conditions, self.condition_locations, strict=True
1449 )
1450 if not cond(value)
1451 ),
1452 None,
1453 )
1454 finally:
1455 data.stop_span(discard=failing is not None)
1456 if failing is None:
1457 return value
1458 data._last_rejected_filter = failing
1459 if i == 0:
1460 data.events[f"Retried draw from {self!r} to satisfy filter"] = ""
1462 return filter_not_satisfied
1464 def _invert(self, value: Any) -> tuple[ChoiceT, ...]:
1465 # If the condition accepts value, do_draw would have succeeded on its
1466 # first try, drawing exactly the inner strategy's encoding.
1467 try:
1468 satisfied = self.condition(value)
1469 except Exception:
1470 # `self.condition` is user code and may raise arbitrarily.
1471 raise CannotInvert(f"{value!r} errored in filter {self!r}") from None
1472 if not satisfied:
1473 raise CannotInvert(f"{value!r} does not satisfy filter {self!r}")
1474 return self.filtered_strategy._invert(value)
1476 @property
1477 def branches(self) -> Sequence[SearchStrategy[Ex]]:
1478 return [
1479 FilteredStrategy(strategy, self.flat_conditions, self.condition_locations)
1480 for strategy in self.filtered_strategy.branches
1481 ]
1484@check_function
1485def check_strategy(arg: object, name: str = "") -> None:
1486 assert isinstance(name, str)
1487 if not isinstance(arg, SearchStrategy):
1488 hint = ""
1489 if isinstance(arg, (list, tuple)):
1490 hint = ", such as st.sampled_from({}),".format(name or "...")
1491 if name:
1492 name += "="
1493 raise InvalidArgument(
1494 f"Expected a SearchStrategy{hint} but got {name}{arg!r} "
1495 f"(type={type(arg).__name__})"
1496 )