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 try:
814 for i, element in enumerate(self.elements):
815 if equal_values(self._transform(element, data=None), value):
816 return (i,)
817 except Exception:
818 # `self._transform` may include arbitrary user code (map/filter functions)
819 raise CannotInvert(f"transforming elements of {self!r} errored") from None
820 raise CannotInvert(f"{value!r} is not produced by {self!r}")
822 def do_filtered_draw(self, data: ConjectureData) -> Ex | UniqueIdentifier:
823 # Set of indices that have been tried so far, so that we never test
824 # the same element twice during a draw.
825 known_bad_indices: set[int] = set()
827 # Start with ordinary rejection sampling. It's fast if it works, and
828 # if it doesn't work then it was only a small amount of overhead.
829 for _ in range(3):
830 i = data.draw_integer(0, len(self.elements) - 1)
831 if i not in known_bad_indices:
832 element = self.get_element(i, data)
833 if element is not filter_not_satisfied:
834 return element
835 if not known_bad_indices:
836 data.events[f"Retried draw from {self!r} to satisfy filter"] = ""
837 known_bad_indices.add(i)
839 # If we've tried all the possible elements, give up now.
840 max_good_indices = len(self.elements) - len(known_bad_indices)
841 if not max_good_indices:
842 return filter_not_satisfied
844 # Impose an arbitrary cutoff to prevent us from wasting too much time
845 # on very large element lists.
846 max_good_indices = min(max_good_indices, self._MAX_FILTER_CALLS - 3)
848 # Before building the list of allowed indices, speculatively choose
849 # one of them. We don't yet know how many allowed indices there will be,
850 # so this choice might be out-of-bounds, but that's OK.
851 speculative_index = data.draw_integer(0, max_good_indices - 1)
853 # Calculate the indices of allowed values, so that we can choose one
854 # of them at random. But if we encounter the speculatively-chosen one,
855 # just use that and return immediately. Note that we also track the
856 # allowed elements, in case of .map(some_stateful_function)
857 allowed: list[tuple[int, Ex]] = []
858 for i in range(min(len(self.elements), self._MAX_FILTER_CALLS - 3)):
859 if i not in known_bad_indices:
860 element = self.get_element(i, data)
861 if element is not filter_not_satisfied:
862 assert not isinstance(element, UniqueIdentifier)
863 allowed.append((i, element))
864 if len(allowed) > speculative_index:
865 # Early-exit case: We reached the speculative index, so
866 # we just return the corresponding element.
867 data.draw_integer(0, len(self.elements) - 1, forced=i)
868 return element
870 # The speculative index didn't work out, but at this point we've built
871 # and can choose from the complete list of allowed indices and elements.
872 if allowed:
873 i, element = data.choice(allowed)
874 data.draw_integer(0, len(self.elements) - 1, forced=i)
875 return element
876 # If there are no allowed indices, the filter couldn't be satisfied.
877 return filter_not_satisfied
880# The ids of OneOfStrategy instances an _invert call is currently walking
881# through, per thread. See OneOfStrategy._invert.
882_inverting_one_ofs = threading.local()
885class OneOfStrategy(SearchStrategy[Ex]):
886 """Implements a union of strategies. Given a number of strategies this
887 generates values which could have come from any of them.
889 The conditional distribution draws uniformly at random from some
890 non-empty subset of these strategies and then draws from the
891 conditional distribution of that strategy.
892 """
894 def __init__(self, strategies: Sequence[SearchStrategy[Ex]]):
895 super().__init__()
896 self.original_strategies = tuple(strategies)
897 self.__element_strategies: Sequence[SearchStrategy[Ex]] | None = None
898 self.__in_branches = False
899 self._branches_lock = RLock()
901 def calc_is_empty(self, recur: RecurT) -> bool:
902 return all(recur(e) for e in self.original_strategies)
904 def calc_has_reusable_values(self, recur: RecurT) -> bool:
905 return all(recur(e) for e in self.original_strategies)
907 def calc_is_cacheable(self, recur: RecurT) -> bool:
908 return all(recur(e) for e in self.original_strategies)
910 @property
911 def element_strategies(self) -> Sequence[SearchStrategy[Ex]]:
912 if self.__element_strategies is None:
913 # While strategies are hashable, they use object.__hash__ and are
914 # therefore distinguished only by identity.
915 #
916 # In principle we could "just" define a __hash__ method
917 # (and __eq__, but that's easy in terms of type() and hash())
918 # to make this more powerful, but this is harder than it sounds:
919 #
920 # 1. Strategies are often distinguished by non-hashable attributes,
921 # or by attributes that have the same hash value ("^.+" / b"^.+").
922 # 2. LazyStrategy: can't reify the wrapped strategy without breaking
923 # laziness, so there's a hash each for the lazy and the nonlazy.
924 #
925 # Having made several attempts, the minor benefits of making strategies
926 # hashable are simply not worth the engineering effort it would take.
927 # See also issues #2291 and #2327.
928 seen: set[SearchStrategy] = {self}
929 strategies: list[SearchStrategy] = []
930 for arg in self.original_strategies:
931 check_strategy(arg)
932 if not arg.is_empty:
933 for s in arg.branches:
934 if s not in seen and not s.is_empty:
935 seen.add(s)
936 strategies.append(s)
937 self.__element_strategies = strategies
938 return self.__element_strategies
940 def calc_label(self) -> int:
941 return combine_labels(
942 self.class_label, *(p.label for p in self.original_strategies)
943 )
945 def do_draw(self, data: ConjectureData) -> Ex:
946 strategies = self.element_strategies
947 if len(strategies) == 1:
948 # optimization: skip constructing SampledFromStrategy if we only have one
949 # strategy. This can happen for eg `st.integers() | st.nothing()`.
950 return data.draw(strategies[0])
952 strategy = data.draw(
953 SampledFromStrategy(strategies).filter(
954 lambda s: not s.is_currently_empty(data)
955 )
956 )
957 return data.draw(strategy)
959 def _invert(self, value: Any) -> tuple[ChoiceT, ...]:
960 # do_draw first draws a branch index, then draws from that branch.
961 # Return the simplest candidate across all branches: the shortest
962 # encoding, breaking ties by the lower branch index - matching the
963 # shrinker's ordering over choice sequences.
964 #
965 # Self-referential strategies (e.g. via st.deferred) can route a
966 # branch's inversion of the same value back to this one_of, so guard
967 # against re-entry rather than recursing forever. Unlike a structural
968 # recursion into a subvalue - which terminates because values are
969 # finite - a same-value cycle can make no progress.
970 if len(self.element_strategies) == 1:
971 return self.element_strategies[0]._invert(value)
973 active = getattr(_inverting_one_ofs, "ids", None)
974 if active is None:
975 active = _inverting_one_ofs.ids = set()
976 if id(self) in active:
977 raise CannotInvert(f"recursive inversion of {self!r}")
978 active.add(id(self))
979 try:
980 best: tuple[ChoiceT, ...] | None = None
981 for i, branch in enumerate(self.element_strategies):
982 try:
983 candidate = (i, *branch._invert(value))
984 except CannotInvert:
985 continue
986 if best is None or len(candidate) < len(best):
987 best = candidate
988 if len(best) <= 2:
989 # A selector plus at most one choice: no later branch is
990 # worth trying. (A zero-choice just()-like branch could
991 # still encode in one choice, but we accept two rather
992 # than scanning - possibly recursively - for it.)
993 break
994 finally:
995 active.discard(id(self))
996 if best is None:
997 raise CannotInvert(f"{value!r} is not produced by any branch of {self!r}")
998 return best
1000 def __repr__(self) -> str:
1001 return "one_of({})".format(", ".join(map(repr, self.original_strategies)))
1003 def do_validate(self) -> None:
1004 for e in self.element_strategies:
1005 e.validate()
1007 @property
1008 def branches(self) -> Sequence[SearchStrategy[Ex]]:
1009 if self.__element_strategies is not None:
1010 # common fast path which avoids the lock
1011 return self.element_strategies
1013 with self._branches_lock:
1014 if not self.__in_branches:
1015 try:
1016 self.__in_branches = True
1017 return self.element_strategies
1018 finally:
1019 self.__in_branches = False
1020 else:
1021 return [self]
1023 @overload
1024 def filter(
1025 self, condition: Callable[[Ex], TypeGuard[T]]
1026 ) -> "SearchStrategy[T]": ...
1027 @overload
1028 def filter(self, condition: Callable[[Ex], Any]) -> "SearchStrategy[Ex]": ...
1029 def filter(self, condition):
1030 return FilteredStrategy(
1031 OneOfStrategy([s.filter(condition) for s in self.original_strategies]),
1032 conditions=(),
1033 )
1036@overload
1037def one_of(
1038 __args: Sequence[SearchStrategy[Ex]],
1039) -> SearchStrategy[Ex]: ...
1042@overload
1043def one_of(__a1: SearchStrategy[Ex]) -> SearchStrategy[Ex]: ...
1046@overload
1047def one_of(
1048 __a1: SearchStrategy[Ex], __a2: SearchStrategy[T]
1049) -> SearchStrategy[Ex | T]: ...
1052@overload
1053def one_of(
1054 __a1: SearchStrategy[Ex], __a2: SearchStrategy[T], __a3: SearchStrategy[T3]
1055) -> SearchStrategy[Ex | T | T3]: ...
1058@overload
1059def one_of(
1060 __a1: SearchStrategy[Ex],
1061 __a2: SearchStrategy[T],
1062 __a3: SearchStrategy[T3],
1063 __a4: SearchStrategy[T4],
1064) -> SearchStrategy[Ex | T | T3 | T4]: ...
1067@overload
1068def one_of(
1069 __a1: SearchStrategy[Ex],
1070 __a2: SearchStrategy[T],
1071 __a3: SearchStrategy[T3],
1072 __a4: SearchStrategy[T4],
1073 __a5: SearchStrategy[T5],
1074) -> SearchStrategy[Ex | T | T3 | T4 | T5]: ...
1077@overload
1078def one_of(*args: SearchStrategy[Any]) -> SearchStrategy[Any]: ...
1081@defines_strategy(eager=True)
1082def one_of(
1083 *args: Sequence[SearchStrategy[Any]] | SearchStrategy[Any],
1084) -> SearchStrategy[Any]:
1085 # Mypy workaround alert: Any is too loose above; the return parameter
1086 # should be the union of the input parameters. Unfortunately, Mypy <=0.600
1087 # raises errors due to incompatible inputs instead. See #1270 for links.
1088 # v0.610 doesn't error; it gets inference wrong for 2+ arguments instead.
1089 """Return a strategy which generates values from any of the argument
1090 strategies.
1092 This may be called with one iterable argument instead of multiple
1093 strategy arguments, in which case ``one_of(x)`` and ``one_of(*x)`` are
1094 equivalent.
1096 Examples from this strategy will generally shrink to ones that come from
1097 strategies earlier in the list, then shrink according to behaviour of the
1098 strategy that produced them. In order to get good shrinking behaviour,
1099 try to put simpler strategies first. e.g. ``one_of(none(), text())`` is
1100 better than ``one_of(text(), none())``.
1102 This is especially important when using recursive strategies. e.g.
1103 ``x = st.deferred(lambda: st.none() | st.tuples(x, x))`` will shrink well,
1104 but ``x = st.deferred(lambda: st.tuples(x, x) | st.none())`` will shrink
1105 very badly indeed.
1106 """
1107 if len(args) == 1 and not isinstance(args[0], SearchStrategy):
1108 try:
1109 args = tuple(args[0])
1110 except TypeError:
1111 pass
1112 if len(args) == 1 and isinstance(args[0], SearchStrategy):
1113 # This special-case means that we can one_of over lists of any size
1114 # without incurring any performance overhead when there is only one
1115 # strategy, and keeps our reprs simple.
1116 return args[0]
1117 if args and not any(isinstance(a, SearchStrategy) for a in args):
1118 # And this special case is to give a more-specific error message if it
1119 # seems that the user has confused `one_of()` for `sampled_from()`;
1120 # the remaining validation is left to OneOfStrategy. See PR #2627.
1121 raise InvalidArgument(
1122 f"Did you mean st.sampled_from({list(args)!r})? st.one_of() is used "
1123 "to combine strategies, but all of the arguments were of other types."
1124 )
1125 # we've handled the case where args is a one-element sequence [(s1, s2, ...)]
1126 # above, so we can assume it's an actual sequence of strategies.
1127 args = cast(Sequence[SearchStrategy], args)
1128 return OneOfStrategy(args)
1131class MappedStrategy(SearchStrategy[MappedTo], Generic[MappedFrom, MappedTo]):
1132 """A strategy which is defined purely by conversion to and from another
1133 strategy.
1135 Its parameter and distribution come from that other strategy.
1136 """
1138 def __init__(
1139 self,
1140 strategy: SearchStrategy[MappedFrom],
1141 pack: Callable[[MappedFrom], MappedTo],
1142 ) -> None:
1143 super().__init__()
1144 self.mapped_strategy = strategy
1145 self.pack = pack
1147 def calc_is_empty(self, recur: RecurT) -> bool:
1148 return recur(self.mapped_strategy)
1150 def calc_is_cacheable(self, recur: RecurT) -> bool:
1151 return recur(self.mapped_strategy)
1153 def __repr__(self) -> str:
1154 if not hasattr(self, "_cached_repr"):
1155 self._cached_repr = f"{self.mapped_strategy!r}.map({get_pretty_function_description(self.pack)})"
1156 return self._cached_repr
1158 def do_validate(self) -> None:
1159 self.mapped_strategy.validate()
1161 def do_draw(self, data: ConjectureData) -> MappedTo:
1162 with warnings.catch_warnings():
1163 if isinstance(self.pack, type) and issubclass(
1164 self.pack, (abc.Mapping, abc.Set)
1165 ):
1166 warnings.simplefilter("ignore", BytesWarning)
1167 for _ in range(3):
1168 try:
1169 data.start_span(MAPPED_SEARCH_STRATEGY_DO_DRAW_LABEL)
1170 x = data.draw(self.mapped_strategy)
1171 result = self.pack(x)
1172 data.stop_span()
1173 current_build_context().record_call(
1174 result, self.pack, args=[x], kwargs={}
1175 )
1176 return result
1177 except UnsatisfiedAssumption as err:
1178 # we want to preserve the err.location of the actual exception here if we
1179 # re-throw it outside of the loop at the end. The alternative is a
1180 # bare `raise UnsatisfiedAssumption` outside of the loop, which would
1181 # drop the real err.location for observability.
1182 #
1183 # If we catch multiple exceptions here, we'll just report the location
1184 # of the last one, which is a reasonable tradeoff for a singleton field.
1185 failed_assumption = err
1186 data.stop_span(discard=True)
1187 raise failed_assumption
1189 def _invert(self, value: Any) -> tuple[ChoiceT, ...]:
1190 # map() is not invertible in general, but a dict-like pack - e.g. the
1191 # dict_class which st.dictionaries maps over a unique list of
1192 # (key, value) tuples - inverts as its list of items.
1193 if (
1194 isinstance(self.pack, type)
1195 and issubclass(self.pack, abc.Mapping)
1196 and isinstance(value, self.pack)
1197 ):
1198 return self.mapped_strategy._invert(list(value.items()))
1199 raise CannotInvert(f"cannot invert {self!r} (value={value!r})")
1201 @property
1202 def branches(self) -> Sequence[SearchStrategy[MappedTo]]:
1203 return [
1204 MappedStrategy(strategy, pack=self.pack)
1205 for strategy in self.mapped_strategy.branches
1206 ]
1208 @overload
1209 def filter(
1210 self, condition: Callable[[MappedTo], TypeGuard[T]]
1211 ) -> "SearchStrategy[T]": ...
1212 @overload
1213 def filter(
1214 self, condition: Callable[[MappedTo], Any]
1215 ) -> "SearchStrategy[MappedTo]": ...
1216 def filter(self, condition):
1217 # Includes a special case so that we can rewrite filters on collection
1218 # lengths, when most collections are `st.lists(...).map(the_type)`.
1219 ListStrategy = _list_strategy_type()
1220 if not isinstance(self.mapped_strategy, ListStrategy) or not (
1221 (isinstance(self.pack, type) and issubclass(self.pack, abc.Collection))
1222 or self.pack in _collection_ish_functions()
1223 ):
1224 return super().filter(condition)
1226 # Check whether our inner list strategy can rewrite this filter condition.
1227 # If not, discard the result and _only_ apply a new outer filter.
1228 new = ListStrategy.filter(self.mapped_strategy, condition)
1229 if getattr(new, "filtered_strategy", None) is self.mapped_strategy:
1230 return super().filter(condition) # didn't rewrite
1232 # Apply a new outer filter even though we rewrote the inner strategy,
1233 # because some collections can change the list length (dict, set, etc).
1234 return FilteredStrategy(
1235 type(self)(new, self.pack),
1236 conditions=(condition,),
1237 condition_locations=(current_filter_call_site(),),
1238 )
1241@lru_cache
1242def _list_strategy_type() -> Any:
1243 from hypothesis.strategies._internal.collections import ListStrategy
1245 return ListStrategy
1248def _collection_ish_functions() -> Sequence[Any]:
1249 funcs = [sorted]
1250 if np := sys.modules.get("numpy"):
1251 # c.f. https://numpy.org/doc/stable/reference/routines.array-creation.html
1252 # Probably only `np.array` and `np.asarray` will be used in practice,
1253 # but why should that stop us when we've already gone this far?
1254 funcs += [
1255 np.empty_like,
1256 np.eye,
1257 np.identity,
1258 np.ones_like,
1259 np.zeros_like,
1260 np.array,
1261 np.asarray,
1262 np.asanyarray,
1263 np.ascontiguousarray,
1264 np.asmatrix,
1265 np.copy,
1266 np.rec.array,
1267 np.rec.fromarrays,
1268 np.rec.fromrecords,
1269 np.diag,
1270 # bonus undocumented functions from tab-completion:
1271 np.asarray_chkfinite,
1272 np.asfortranarray,
1273 ]
1275 return funcs
1278filter_not_satisfied = UniqueIdentifier("filter not satisfied")
1281class FilteredStrategy(SearchStrategy[Ex]):
1282 def __init__(
1283 self,
1284 strategy: SearchStrategy[Ex],
1285 conditions: tuple[Callable[[Ex], Any], ...],
1286 condition_locations: tuple[str | None, ...] | None = None,
1287 ):
1288 super().__init__()
1289 # Where each condition's .filter() call was made, for observability
1290 if condition_locations is None:
1291 condition_locations = (None,) * len(conditions)
1292 assert len(condition_locations) == len(conditions)
1293 if isinstance(strategy, FilteredStrategy):
1294 # Flatten chained filters into a single filter with multiple conditions.
1295 self.flat_conditions: tuple[Callable[[Ex], Any], ...] = (
1296 strategy.flat_conditions + conditions
1297 )
1298 self.condition_locations: tuple[str | None, ...] = (
1299 strategy.condition_locations + condition_locations
1300 )
1301 self.filtered_strategy: SearchStrategy[Ex] = strategy.filtered_strategy
1302 else:
1303 self.flat_conditions = conditions
1304 self.condition_locations = condition_locations
1305 self.filtered_strategy = strategy
1307 assert isinstance(self.flat_conditions, tuple)
1308 assert not isinstance(self.filtered_strategy, FilteredStrategy)
1310 self.__condition: Callable[[Ex], Any] | None = None
1312 def calc_is_empty(self, recur: RecurT) -> bool:
1313 return recur(self.filtered_strategy)
1315 def calc_is_cacheable(self, recur: RecurT) -> bool:
1316 return recur(self.filtered_strategy)
1318 def __repr__(self) -> str:
1319 if not hasattr(self, "_cached_repr"):
1320 self._cached_repr = "{!r}{}".format(
1321 self.filtered_strategy,
1322 "".join(
1323 f".filter({get_pretty_function_description(cond)})"
1324 for cond in self.flat_conditions
1325 ),
1326 )
1327 return self._cached_repr
1329 def do_validate(self) -> None:
1330 # Start by validating our inner filtered_strategy. If this was a LazyStrategy,
1331 # validation also reifies it so that subsequent calls to e.g. `.filter()` will
1332 # be passed through.
1333 self.filtered_strategy.validate()
1334 # So now we have a reified inner strategy, we'll replay all our saved
1335 # predicates in case some or all of them can be rewritten. Note that this
1336 # replaces the `fresh` strategy too!
1337 fresh = self.filtered_strategy
1338 for cond, location in zip(
1339 self.flat_conditions, self.condition_locations, strict=True
1340 ):
1341 with _filter_location_override.with_value(location):
1342 fresh = fresh.filter(cond)
1343 if isinstance(fresh, FilteredStrategy):
1344 # In this case we have at least some non-rewritten filter predicates,
1345 # so we just re-initialize the strategy.
1346 FilteredStrategy.__init__(
1347 self,
1348 fresh.filtered_strategy,
1349 fresh.flat_conditions,
1350 fresh.condition_locations,
1351 )
1352 else:
1353 # But if *all* the predicates were rewritten... well, do_validate() is
1354 # an in-place method so we still just re-initialize the strategy!
1355 FilteredStrategy.__init__(self, fresh, ())
1357 @overload
1358 def filter(
1359 self, condition: Callable[[Ex], TypeGuard[T]]
1360 ) -> "FilteredStrategy[T]": ...
1361 @overload
1362 def filter(self, condition: Callable[[Ex], Any]) -> "FilteredStrategy[Ex]": ...
1363 def filter(self, condition):
1364 # If we can, it's more efficient to rewrite our strategy to satisfy the
1365 # condition. We therefore exploit the fact that the order of predicates
1366 # doesn't matter (`f(x) and g(x) == g(x) and f(x)`) by attempting to apply
1367 # condition directly to our filtered strategy as the inner-most filter.
1368 out = self.filtered_strategy.filter(condition)
1369 # If it couldn't be rewritten, we'll get a new FilteredStrategy - and then
1370 # combine the conditions of each in our expected newest=last order.
1371 if isinstance(out, FilteredStrategy):
1372 return FilteredStrategy(
1373 out.filtered_strategy,
1374 self.flat_conditions + out.flat_conditions,
1375 self.condition_locations + out.condition_locations,
1376 )
1377 # But if it *could* be rewritten, we can return the more efficient form!
1378 return FilteredStrategy(out, self.flat_conditions, self.condition_locations)
1380 @property
1381 def condition(self) -> Callable[[Ex], Any]:
1382 # We write this defensively to avoid any threading race conditions
1383 # with our manual FilteredStrategy.__init__ for filter-rewriting.
1384 # See https://github.com/HypothesisWorks/hypothesis/pull/4522.
1385 if (condition := self.__condition) is not None:
1386 return condition
1388 if len(self.flat_conditions) == 1:
1389 # Avoid an extra indirection in the common case of only one condition.
1390 condition = self.flat_conditions[0]
1391 elif len(self.flat_conditions) == 0:
1392 # Possible, if unlikely, due to filter predicate rewriting
1393 condition = lambda _: True
1394 else:
1395 condition = lambda x: all(cond(x) for cond in self.flat_conditions)
1396 self.__condition = condition
1397 return condition
1399 def do_draw(self, data: ConjectureData) -> Ex:
1400 result = self.do_filtered_draw(data)
1401 if result is not filter_not_satisfied:
1402 return cast(Ex, result)
1404 # do_filtered_draw records data._last_rejected_filter.
1405 data.mark_invalid(
1406 f"Aborted test because unable to satisfy {self!r}",
1407 location=data.last_rejected_filter_location(),
1408 )
1410 def do_filtered_draw(self, data: ConjectureData) -> Ex | UniqueIdentifier:
1411 for i in range(3):
1412 data.start_span(FILTERED_SEARCH_STRATEGY_DO_DRAW_LABEL)
1413 value = data.draw(self.filtered_strategy)
1414 # Check the conditions individually rather than via self.condition,
1415 # so that we can set data._last_rejected_filter.
1416 failing = next(
1417 (
1418 (cond, location)
1419 for cond, location in zip(
1420 self.flat_conditions, self.condition_locations, strict=True
1421 )
1422 if not cond(value)
1423 ),
1424 None,
1425 )
1426 if failing is None:
1427 data.stop_span()
1428 return value
1429 else:
1430 data._last_rejected_filter = failing
1431 data.stop_span(discard=True)
1432 if i == 0:
1433 data.events[f"Retried draw from {self!r} to satisfy filter"] = ""
1435 return filter_not_satisfied
1437 def _invert(self, value: Any) -> tuple[ChoiceT, ...]:
1438 # If the condition accepts value, do_draw would have succeeded on its
1439 # first try, drawing exactly the inner strategy's encoding.
1440 try:
1441 satisfied = self.condition(value)
1442 except Exception:
1443 # `self.condition` is user code and may raise arbitrarily.
1444 raise CannotInvert(f"{value!r} errored in filter {self!r}") from None
1445 if not satisfied:
1446 raise CannotInvert(f"{value!r} does not satisfy filter {self!r}")
1447 return self.filtered_strategy._invert(value)
1449 @property
1450 def branches(self) -> Sequence[SearchStrategy[Ex]]:
1451 return [
1452 FilteredStrategy(strategy, self.flat_conditions, self.condition_locations)
1453 for strategy in self.filtered_strategy.branches
1454 ]
1457@check_function
1458def check_strategy(arg: object, name: str = "") -> None:
1459 assert isinstance(name, str)
1460 if not isinstance(arg, SearchStrategy):
1461 hint = ""
1462 if isinstance(arg, (list, tuple)):
1463 hint = ", such as st.sampled_from({}),".format(name or "...")
1464 if name:
1465 name += "="
1466 raise InvalidArgument(
1467 f"Expected a SearchStrategy{hint} but got {name}{arg!r} "
1468 f"(type={type(arg).__name__})"
1469 )