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