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 typing import (
21 TYPE_CHECKING,
22 Any,
23 ClassVar,
24 Generic,
25 Literal,
26 TypeAlias,
27 TypeGuard,
28 TypeVar,
29 cast,
30 overload,
31)
33from hypothesis._settings import HealthCheck, Phase, Verbosity, settings
34from hypothesis.control import _current_build_context, current_build_context
35from hypothesis.errors import (
36 HypothesisException,
37 HypothesisWarning,
38 InvalidArgument,
39 NonInteractiveExampleWarning,
40 UnsatisfiedAssumption,
41)
42from hypothesis.internal.conjecture import utils as cu
43from hypothesis.internal.conjecture.data import ConjectureData
44from hypothesis.internal.conjecture.utils import (
45 calc_label_from_cls,
46 calc_label_from_hash,
47 calc_label_from_name,
48 combine_labels,
49)
50from hypothesis.internal.coverage import check_function
51from hypothesis.internal.reflection import (
52 get_pretty_function_description,
53 is_identity_function,
54)
55from hypothesis.strategies._internal.utils import defines_strategy
56from hypothesis.utils.conventions import UniqueIdentifier
58if TYPE_CHECKING:
59 Ex = TypeVar("Ex", covariant=True, default=Any)
60else:
61 Ex = TypeVar("Ex", covariant=True)
63T = TypeVar("T")
64T3 = TypeVar("T3")
65T4 = TypeVar("T4")
66T5 = TypeVar("T5")
67MappedFrom = TypeVar("MappedFrom")
68MappedTo = TypeVar("MappedTo")
69RecurT: TypeAlias = Callable[["SearchStrategy"], bool]
70calculating = UniqueIdentifier("calculating")
72MAPPED_SEARCH_STRATEGY_DO_DRAW_LABEL = calc_label_from_name(
73 "another attempted draw in MappedStrategy"
74)
76FILTERED_SEARCH_STRATEGY_DO_DRAW_LABEL = calc_label_from_name(
77 "single loop iteration in FilteredStrategy"
78)
80label_lock = RLock()
83def recursive_property(strategy: "SearchStrategy", name: str, default: object) -> Any:
84 """Handle properties which may be mutually recursive among a set of
85 strategies.
87 These are essentially lazily cached properties, with the ability to set
88 an override: If the property has not been explicitly set, we calculate
89 it on first access and memoize the result for later.
91 The problem is that for properties that depend on each other, a naive
92 calculation strategy may hit infinite recursion. Consider for example
93 the property is_empty. A strategy defined as x = st.deferred(lambda: x)
94 is certainly empty (in order to draw a value from x we would have to
95 draw a value from x, for which we would have to draw a value from x,
96 ...), but in order to calculate it the naive approach would end up
97 calling x.is_empty in order to calculate x.is_empty in order to etc.
99 The solution is one of fixed point calculation. We start with a default
100 value that is the value of the property in the absence of evidence to
101 the contrary, and then update the values of the property for all
102 dependent strategies until we reach a fixed point.
104 The approach taken roughly follows that in section 4.2 of Adams,
105 Michael D., Celeste Hollenbeck, and Matthew Might. "On the complexity
106 and performance of parsing with derivatives." ACM SIGPLAN Notices 51.6
107 (2016): 224-236.
108 """
109 assert name in {"is_empty", "has_reusable_values", "is_cacheable"}
110 cache_key = "cached_" + name
111 calculation = "calc_" + name
112 force_key = "force_" + name
114 def forced_or_cached_value(target: SearchStrategy) -> Any:
115 try:
116 return getattr(target, force_key)
117 except AttributeError:
118 return getattr(target, cache_key)
120 try:
121 return forced_or_cached_value(strategy)
122 except AttributeError:
123 pass
125 mapping: dict[SearchStrategy, Any] = {}
126 sentinel = object()
127 hit_recursion = False
129 # For a first pass we do a direct recursive calculation of the
130 # property, but we block recursively visiting a value in the
131 # computation of its property: When that happens, we simply
132 # note that it happened and return the default value.
133 def recur(strat: SearchStrategy) -> Any:
134 nonlocal hit_recursion
135 try:
136 return forced_or_cached_value(strat)
137 except AttributeError:
138 pass
139 result = mapping.get(strat, sentinel)
140 if result is calculating:
141 hit_recursion = True
142 return default
143 elif result is sentinel:
144 mapping[strat] = calculating
145 mapping[strat] = getattr(strat, calculation)(recur)
146 return mapping[strat]
147 return result
149 recur(strategy)
151 # If we hit self-recursion in the computation of any strategy
152 # value, our mapping at the end is imprecise - it may or may
153 # not have the right values in it. We now need to proceed with
154 # a more careful fixed point calculation to get the exact
155 # values. Hopefully our mapping is still pretty good and it
156 # won't take a large number of updates to reach a fixed point.
157 if hit_recursion:
158 needs_update = set(mapping)
160 # We track which strategies use which in the course of
161 # calculating their property value. If A ever uses B in
162 # the course of calculating its value, then whenever the
163 # value of B changes we might need to update the value of
164 # A.
165 listeners: dict[SearchStrategy, set[SearchStrategy]] = defaultdict(set)
166 else:
167 needs_update = None
169 def recur2(strat: SearchStrategy) -> Any:
170 def recur_inner(other: SearchStrategy) -> Any:
171 try:
172 return forced_or_cached_value(other)
173 except AttributeError:
174 pass
175 listeners[other].add(strat)
176 result = mapping.get(other, sentinel)
177 if result is sentinel:
178 assert needs_update is not None
179 needs_update.add(other)
180 mapping[other] = default
181 return default
182 return result
184 return recur_inner
186 count = 0
187 seen = set()
188 while needs_update:
189 count += 1
190 # If we seem to be taking a really long time to stabilize we
191 # start tracking seen values to attempt to detect an infinite
192 # loop. This should be impossible, and most code will never
193 # hit the count, but having an assertion for it means that
194 # testing is easier to debug and we don't just have a hung
195 # test.
196 # Note: This is actually covered, by test_very_deep_deferral
197 # in tests/cover/test_deferred_strategies.py. Unfortunately it
198 # runs into a coverage bug. See
199 # https://github.com/nedbat/coveragepy/issues/605
200 # for details.
201 if count > 50: # pragma: no cover
202 key = frozenset(mapping.items())
203 assert key not in seen, (key, name)
204 seen.add(key)
205 to_update = needs_update
206 needs_update = set()
207 for strat in to_update:
208 new_value = getattr(strat, calculation)(recur2(strat))
209 if new_value != mapping[strat]:
210 needs_update.update(listeners[strat])
211 mapping[strat] = new_value
213 # We now have a complete and accurate calculation of the
214 # property values for everything we have seen in the course of
215 # running this calculation. We simultaneously update all of
216 # them (not just the strategy we started out with).
217 for k, v in mapping.items():
218 setattr(k, cache_key, v)
219 # This used to simply be `getattr(strategy, cache_key)`. That relied on the invariant
220 # that our loop above has set `strategy.cached_* = v` on `strategy` if we've reached
221 # here. However, under threading, this is not necessarily true. If a concurrent thread
222 # sets `strategy.force_* = v` in between the two places we check for `force_*`, we
223 # will not set `strategy.cached_*`.
224 #
225 # There are several places where we might do this. unwrap_strategies sets
226 # force_has_reusable_values = True. our numpy.py's `arrays` strategy also does.
227 #
228 # We guard against this in general by checking the forced and cached values here,
229 # rather than just the cached value.
230 return forced_or_cached_value(strategy)
233class SearchStrategy(Generic[Ex]):
234 """A ``SearchStrategy`` tells Hypothesis how to generate that kind of input.
236 This class is only part of the public API for use in type annotations, so that
237 you can write e.g. ``-> SearchStrategy[Foo]`` for your function which returns
238 ``builds(Foo, ...)``. Do not inherit from or directly instantiate this class.
239 """
241 __module__: str = "hypothesis.strategies"
242 LABELS: ClassVar[dict[type, int]] = {}
243 # triggers `assert isinstance(label, int)` under threading when setting this
244 # in init instead of a classvar. I'm not sure why, init should be safe. But
245 # this works so I'm not looking into it further atm.
246 __label: int | UniqueIdentifier | None = None
248 def __init__(self) -> None:
249 self.validate_called: dict[int, bool] = {}
251 def is_currently_empty(self, data: ConjectureData) -> bool:
252 """
253 Returns whether this strategy is currently empty. Unlike ``empty``,
254 which is computed based on static information and cannot change,
255 ``is_currently_empty`` may change over time based on choices made
256 during the test case.
258 This is currently only used for stateful testing, where |Bundle| grows a
259 list of values to choose from over the course of a test case.
261 ``data`` will only be used for introspection. No values will be drawn
262 from it in a way that modifies the choice sequence.
263 """
264 return self.is_empty
266 @property
267 def is_empty(self) -> Any:
268 # Returns True if this strategy can never draw a value and will always
269 # result in the data being marked invalid.
270 # The fact that this returns False does not guarantee that a valid value
271 # can be drawn - this is not intended to be perfect, and is primarily
272 # intended to be an optimisation for some cases.
273 return recursive_property(self, "is_empty", True)
275 # Returns True if values from this strategy can safely be reused without
276 # this causing unexpected behaviour.
278 # True if values from this strategy can be implicitly reused (e.g. as
279 # background values in a numpy array) without causing surprising
280 # user-visible behaviour. Should be false for built-in strategies that
281 # produce mutable values, and for strategies that have been mapped/filtered
282 # by arbitrary user-provided functions.
283 @property
284 def has_reusable_values(self) -> Any:
285 return recursive_property(self, "has_reusable_values", True)
287 @property
288 def is_cacheable(self) -> Any:
289 """
290 Whether it is safe to hold on to instances of this strategy in a cache.
291 See _STRATEGY_CACHE.
292 """
293 return recursive_property(self, "is_cacheable", True)
295 def calc_is_cacheable(self, recur: RecurT) -> bool:
296 return True
298 def calc_is_empty(self, recur: RecurT) -> bool:
299 # Note: It is correct and significant that the default return value
300 # from calc_is_empty is False despite the default value for is_empty
301 # being true. The reason for this is that strategies should be treated
302 # as empty absent evidence to the contrary, but most basic strategies
303 # are trivially non-empty and it would be annoying to have to override
304 # this method to show that.
305 return False
307 def calc_has_reusable_values(self, recur: RecurT) -> bool:
308 return False
310 def example(self) -> Ex: # FIXME
311 """Provide an example of the sort of value that this strategy generates.
313 This method is designed for use in a REPL, and will raise an error if
314 called from inside |@given| or a strategy definition. For serious use,
315 see |@composite| or |st.data|.
316 """
317 if getattr(sys, "ps1", None) is None and (
318 # The main module's __spec__ is None when running interactively
319 # or running a source file directly.
320 # See https://docs.python.org/3/reference/import.html#main-spec.
321 sys.modules["__main__"].__spec__ is not None
322 # __spec__ is also None under pytest-xdist. To avoid an unfortunate
323 # missed alarm here, always warn under pytest.
324 or os.environ.get("PYTEST_CURRENT_TEST") is not None
325 ): # pragma: no branch
326 # The other branch *is* covered in cover/test_interactive_example.py;
327 # but as that uses `pexpect` for an interactive session `coverage`
328 # doesn't see it.
329 warnings.warn(
330 "The `.example()` method is good for exploring strategies, but should "
331 "only be used interactively. We recommend using `@given` for tests - "
332 "it performs better, saves and replays failures to avoid flakiness, "
333 f"and reports minimal examples. (strategy: {self!r})",
334 NonInteractiveExampleWarning,
335 stacklevel=2,
336 )
338 context = _current_build_context.value
339 if context is not None:
340 if context.data is not None and context.data.depth > 0:
341 raise HypothesisException(
342 "Using example() inside a strategy definition is a bad "
343 "idea. Instead consider using hypothesis.strategies.builds() "
344 "or @hypothesis.strategies.composite to define your strategy."
345 " See https://hypothesis.readthedocs.io/en/latest/reference/"
346 "strategies.html#hypothesis.strategies.builds or "
347 "https://hypothesis.readthedocs.io/en/latest/reference/"
348 "strategies.html#hypothesis.strategies.composite for more "
349 "details."
350 )
351 else:
352 raise HypothesisException(
353 "Using example() inside a test function is a bad "
354 "idea. Instead consider using hypothesis.strategies.data() "
355 "to draw more examples during testing. See "
356 "https://hypothesis.readthedocs.io/en/latest/reference/"
357 "strategies.html#hypothesis.strategies.data for more details."
358 )
360 try:
361 return self.__examples.pop()
362 except (AttributeError, IndexError):
363 self.__examples: list[Ex] = []
365 from hypothesis.core import given
367 # Note: this function has a weird name because it might appear in
368 # tracebacks, and we want users to know that they can ignore it.
369 @given(self)
370 @settings(
371 database=None,
372 # generate only a few examples at a time to avoid slow interactivity
373 # for large strategies. The overhead of @given is very small relative
374 # to generation, so a small batch size is fine.
375 max_examples=10,
376 deadline=None,
377 verbosity=Verbosity.quiet,
378 phases=(Phase.generate,),
379 suppress_health_check=list(HealthCheck),
380 )
381 def example_generating_inner_function(
382 ex: Ex, # type: ignore # mypy is overzealous in preventing covariant params
383 ) -> None:
384 self.__examples.append(ex)
386 example_generating_inner_function()
387 shuffle(self.__examples)
388 return self.__examples.pop()
390 def map(self, pack: Callable[[Ex], T]) -> "SearchStrategy[T]":
391 """Returns a new strategy which generates a value from this one, and
392 then returns ``pack(value)``. For example, ``integers().map(str)``
393 could generate ``str(5)`` == ``"5"``.
394 """
395 if is_identity_function(pack):
396 return self # type: ignore # Mypy has no way to know that `Ex == T`
397 return MappedStrategy(self, pack=pack)
399 def flatmap(
400 self, expand: Callable[[Ex], "SearchStrategy[T]"]
401 ) -> "SearchStrategy[T]": # FIXME
402 """Old syntax for a special case of |@composite|:
404 .. code-block:: python
406 @st.composite
407 def flatmap_like(draw, base_strategy, expand):
408 value = draw(base_strategy)
409 new_strategy = expand(value)
410 return draw(new_strategy)
412 We find that the greater readability of |@composite| usually outweighs
413 the verbosity, with a few exceptions for simple cases or recipes like
414 ``from_type(type).flatmap(from_type)`` ("pick a type, get a strategy for
415 any instance of that type, and then generate one of those").
416 """
417 from hypothesis.strategies._internal.flatmapped import FlatMapStrategy
419 return FlatMapStrategy(self, expand=expand)
421 # Note that we previously had condition extracted to a type alias as
422 # PredicateT. However, that was only useful when not specifying a relationship
423 # between the generic Ts and some other function param / return value.
424 # If we do want to - like here, where we want to say that the Ex arg to condition
425 # is of the same type as the strategy's Ex - then you need to write out the
426 # entire Callable[[Ex], Any] expression rather than use a type alias.
427 # TypeAlias is *not* simply a macro that inserts the text. TypeAlias will not
428 # reference the local TypeVar context.
429 @overload
430 def filter(
431 self, condition: Callable[[Ex], TypeGuard[T]]
432 ) -> "SearchStrategy[T]": ...
433 @overload
434 def filter(self, condition: Callable[[Ex], Any]) -> "SearchStrategy[Ex]": ...
435 def filter(self, condition):
436 """Returns a new strategy that generates values from this strategy
437 which satisfy the provided condition.
439 Note that if the condition is too hard to satisfy this might result
440 in your tests failing with an Unsatisfiable exception.
441 A basic version of the filtering logic would look something like:
443 .. code-block:: python
445 @st.composite
446 def filter_like(draw, strategy, condition):
447 for _ in range(3):
448 value = draw(strategy)
449 if condition(value):
450 return value
451 assume(False)
452 """
453 return FilteredStrategy(self, conditions=(condition,))
455 @property
456 def branches(self) -> Sequence["SearchStrategy[Ex]"]:
457 return [self]
459 def __or__(self, other: "SearchStrategy[T]") -> "SearchStrategy[Ex | T]":
460 """Return a strategy which produces values by randomly drawing from one
461 of this strategy or the other strategy.
463 This method is part of the public API.
464 """
465 if not isinstance(other, SearchStrategy):
466 raise ValueError(f"Cannot | a SearchStrategy with {other!r}")
468 # Unwrap explicitly or'd strategies. This turns the
469 # common case of e.g. st.integers() | st.integers() | st.integers() from
470 #
471 # one_of(one_of(integers(), integers()), integers())
472 #
473 # into
474 #
475 # one_of(integers(), integers(), integers())
476 #
477 # This is purely an aesthetic unwrapping, for e.g. reprs. In practice
478 # we use .branches / .element_strategies to get the list of possible
479 # strategies, so this unwrapping is *not* necessary for correctness.
480 strategies: list[SearchStrategy] = []
481 strategies.extend(
482 self.original_strategies if isinstance(self, OneOfStrategy) else [self]
483 )
484 strategies.extend(
485 other.original_strategies if isinstance(other, OneOfStrategy) else [other]
486 )
487 return OneOfStrategy(strategies)
489 def __bool__(self) -> bool:
490 warnings.warn(
491 f"bool({self!r}) is always True, did you mean to draw a value?",
492 HypothesisWarning,
493 stacklevel=2,
494 )
495 return True
497 def validate(self) -> None:
498 """Throw an exception if the strategy is not valid.
500 Strategies should implement ``do_validate``, which is called by this
501 method. They should not override ``validate``.
503 This can happen due to invalid arguments, or lazy construction.
504 """
505 thread_id = threading.get_ident()
506 if self.validate_called.get(thread_id, False):
507 return
508 # we need to set validate_called before calling do_validate, for
509 # recursive / deferred strategies. But if a thread switches after
510 # validate_called but before do_validate, we might have a strategy
511 # which does weird things like drawing when do_validate would error but
512 # its params are technically valid (e.g. a param was passed as 1.0
513 # instead of 1) and get into weird internal states.
514 #
515 # There are two ways to fix this.
516 # (1) The first is a per-strategy lock around do_validate. Even though we
517 # expect near-zero lock contention, this still adds the lock overhead.
518 # (2) The second is allowing concurrent .validate calls. Since validation
519 # is (assumed to be) deterministic, both threads will produce the same
520 # end state, so the validation order or race conditions does not matter.
521 #
522 # In order to avoid the lock overhead of (1), we use (2) here. See also
523 # discussion in https://github.com/HypothesisWorks/hypothesis/pull/4473.
524 try:
525 self.validate_called[thread_id] = True
526 self.do_validate()
527 self.is_empty
528 self.has_reusable_values
529 except Exception:
530 self.validate_called[thread_id] = False
531 raise
533 @property
534 def class_label(self) -> int:
535 cls = self.__class__
536 try:
537 return cls.LABELS[cls]
538 except KeyError:
539 pass
540 result = calc_label_from_cls(cls)
541 cls.LABELS[cls] = result
542 return result
544 @property
545 def label(self) -> int:
546 if isinstance((label := self.__label), int):
547 # avoid locking if we've already completely computed the label.
548 return label
550 with label_lock:
551 if self.__label is calculating:
552 return 0
553 self.__label = calculating
554 self.__label = self.calc_label()
555 return self.__label
557 def calc_label(self) -> int:
558 return self.class_label
560 def do_validate(self) -> None:
561 pass
563 def do_draw(self, data: ConjectureData) -> Ex:
564 raise NotImplementedError(f"{type(self).__name__}.do_draw")
567def _is_hashable(value: object) -> tuple[bool, int | None]:
568 # hashing can be expensive; return the hash value if we compute it, so that
569 # callers don't have to recompute.
570 try:
571 return (True, hash(value))
572 except TypeError:
573 return (False, None)
576def is_hashable(value: object) -> bool:
577 return _is_hashable(value)[0]
580class SampledFromStrategy(SearchStrategy[Ex]):
581 """A strategy which samples from a set of elements. This is essentially
582 equivalent to using a OneOfStrategy over Just strategies but may be more
583 efficient and convenient.
584 """
586 _MAX_FILTER_CALLS: ClassVar[int] = 10_000
588 def __init__(
589 self,
590 elements: Sequence[Ex],
591 *,
592 force_repr: str | None = None,
593 force_repr_braces: tuple[str, str] | None = None,
594 transformations: tuple[
595 tuple[Literal["filter", "map"], Callable[[Ex], Any]],
596 ...,
597 ] = (),
598 ):
599 super().__init__()
600 self.elements = cu.check_sample(elements, "sampled_from")
601 assert self.elements
602 self.force_repr = force_repr
603 self.force_repr_braces = force_repr_braces
604 self._transformations = transformations
606 self._cached_repr: str | None = None
608 def map(self, pack: Callable[[Ex], T]) -> SearchStrategy[T]:
609 s = type(self)(
610 self.elements,
611 force_repr=self.force_repr,
612 force_repr_braces=self.force_repr_braces,
613 transformations=(*self._transformations, ("map", pack)),
614 )
615 # guaranteed by the ("map", pack) transformation
616 return cast(SearchStrategy[T], s)
618 @overload
619 def filter(
620 self, condition: Callable[[Ex], TypeGuard[T]]
621 ) -> "SearchStrategy[T]": ...
622 @overload
623 def filter(self, condition: Callable[[Ex], Any]) -> "SearchStrategy[Ex]": ...
624 def filter(self, condition):
625 return type(self)(
626 self.elements,
627 force_repr=self.force_repr,
628 force_repr_braces=self.force_repr_braces,
629 transformations=(*self._transformations, ("filter", condition)),
630 )
632 def __repr__(self):
633 if self._cached_repr is None:
634 rep = get_pretty_function_description
635 elements_s = (
636 ", ".join(rep(v) for v in self.elements[:512]) + ", ..."
637 if len(self.elements) > 512
638 else ", ".join(rep(v) for v in self.elements)
639 )
640 braces = self.force_repr_braces or ("(", ")")
641 instance_s = (
642 self.force_repr or f"sampled_from({braces[0]}{elements_s}{braces[1]})"
643 )
644 transforms_s = "".join(
645 f".{name}({get_pretty_function_description(f)})"
646 for name, f in self._transformations
647 )
648 repr_s = instance_s + transforms_s
649 self._cached_repr = repr_s
650 return self._cached_repr
652 def calc_label(self) -> int:
653 # strategy.label is effectively an under-approximation of structural
654 # equality (i.e., some strategies may have the same label when they are not
655 # structurally identical). More importantly for calculating the
656 # SampledFromStrategy label, we might have hash(s1) != hash(s2) even
657 # when s1 and s2 are structurally identical. For instance:
658 #
659 # s1 = st.sampled_from([st.none()])
660 # s2 = st.sampled_from([st.none()])
661 # assert hash(s1) != hash(s2)
662 #
663 # (see also test cases in test_labels.py).
664 #
665 # We therefore use the labels of any component strategies when calculating
666 # our label, and only use the hash if it is not a strategy.
667 #
668 # That's the ideal, anyway. In reality the logic is more complicated than
669 # necessary in order to be efficient in the presence of (very) large sequences:
670 # * add an unabashed special case for range, to avoid iteration over an
671 # enormous range when we know it is entirely integers.
672 # * if there is at least one strategy in self.elements, use strategy label,
673 # and the element hash otherwise.
674 # * if there are no strategies in self.elements, take the hash of the
675 # entire sequence. This prevents worst-case performance of hashing each
676 # element when a hash of the entire sequence would have sufficed.
677 #
678 # The worst case performance of this scheme is
679 # itertools.chain(range(2**100), [st.none()]), where it degrades to
680 # hashing every int in the range.
681 elements_is_hashable, hash_value = _is_hashable(self.elements)
682 if isinstance(self.elements, range) or (
683 elements_is_hashable
684 and not any(isinstance(e, SearchStrategy) for e in self.elements)
685 ):
686 return combine_labels(
687 self.class_label, calc_label_from_name(str(hash_value))
688 )
690 labels = [self.class_label]
691 for element in self.elements:
692 if not is_hashable(element):
693 continue
695 labels.append(
696 element.label
697 if isinstance(element, SearchStrategy)
698 else calc_label_from_hash(element)
699 )
701 return combine_labels(*labels)
703 def calc_has_reusable_values(self, recur: RecurT) -> bool:
704 # Because our custom .map/.filter implementations skip the normal
705 # wrapper strategies (which would automatically return False for us),
706 # we need to manually return False here if any transformations have
707 # been applied.
708 return not self._transformations
710 def calc_is_cacheable(self, recur: RecurT) -> bool:
711 return is_hashable(self.elements)
713 def _transform(
714 self,
715 # https://github.com/python/mypy/issues/7049, we're not writing `element`
716 # anywhere in the class so this is still type-safe. mypy is being more
717 # conservative than necessary
718 element: Ex, # type: ignore
719 ) -> Ex | UniqueIdentifier:
720 # Used in UniqueSampledListStrategy
721 for name, f in self._transformations:
722 if name == "map":
723 result = f(element)
724 if build_context := _current_build_context.value:
725 build_context.record_call(result, f, args=[element], kwargs={})
726 element = result
727 else:
728 assert name == "filter"
729 if not f(element):
730 return filter_not_satisfied
731 return element
733 def do_draw(self, data: ConjectureData) -> Ex:
734 result = self.do_filtered_draw(data)
735 if isinstance(result, SearchStrategy) and all(
736 isinstance(x, SearchStrategy) for x in self.elements
737 ):
738 data._sampled_from_all_strategies_elements_message = (
739 "sampled_from was given a collection of strategies: "
740 "{!r}. Was one_of intended?",
741 self.elements,
742 )
743 if result is filter_not_satisfied:
744 data.mark_invalid(f"Aborted test because unable to satisfy {self!r}")
745 assert not isinstance(result, UniqueIdentifier)
746 return result
748 def get_element(self, i: int) -> Ex | UniqueIdentifier:
749 return self._transform(self.elements[i])
751 def do_filtered_draw(self, data: ConjectureData) -> Ex | UniqueIdentifier:
752 # Set of indices that have been tried so far, so that we never test
753 # the same element twice during a draw.
754 known_bad_indices: set[int] = set()
756 # Start with ordinary rejection sampling. It's fast if it works, and
757 # if it doesn't work then it was only a small amount of overhead.
758 for _ in range(3):
759 i = data.draw_integer(0, len(self.elements) - 1)
760 if i not in known_bad_indices:
761 element = self.get_element(i)
762 if element is not filter_not_satisfied:
763 return element
764 if not known_bad_indices:
765 data.events[f"Retried draw from {self!r} to satisfy filter"] = ""
766 known_bad_indices.add(i)
768 # If we've tried all the possible elements, give up now.
769 max_good_indices = len(self.elements) - len(known_bad_indices)
770 if not max_good_indices:
771 return filter_not_satisfied
773 # Impose an arbitrary cutoff to prevent us from wasting too much time
774 # on very large element lists.
775 max_good_indices = min(max_good_indices, self._MAX_FILTER_CALLS - 3)
777 # Before building the list of allowed indices, speculatively choose
778 # one of them. We don't yet know how many allowed indices there will be,
779 # so this choice might be out-of-bounds, but that's OK.
780 speculative_index = data.draw_integer(0, max_good_indices - 1)
782 # Calculate the indices of allowed values, so that we can choose one
783 # of them at random. But if we encounter the speculatively-chosen one,
784 # just use that and return immediately. Note that we also track the
785 # allowed elements, in case of .map(some_stateful_function)
786 allowed: list[tuple[int, Ex]] = []
787 for i in range(min(len(self.elements), self._MAX_FILTER_CALLS - 3)):
788 if i not in known_bad_indices:
789 element = self.get_element(i)
790 if element is not filter_not_satisfied:
791 assert not isinstance(element, UniqueIdentifier)
792 allowed.append((i, element))
793 if len(allowed) > speculative_index:
794 # Early-exit case: We reached the speculative index, so
795 # we just return the corresponding element.
796 data.draw_integer(0, len(self.elements) - 1, forced=i)
797 return element
799 # The speculative index didn't work out, but at this point we've built
800 # and can choose from the complete list of allowed indices and elements.
801 if allowed:
802 i, element = data.choice(allowed)
803 data.draw_integer(0, len(self.elements) - 1, forced=i)
804 return element
805 # If there are no allowed indices, the filter couldn't be satisfied.
806 return filter_not_satisfied
809class OneOfStrategy(SearchStrategy[Ex]):
810 """Implements a union of strategies. Given a number of strategies this
811 generates values which could have come from any of them.
813 The conditional distribution draws uniformly at random from some
814 non-empty subset of these strategies and then draws from the
815 conditional distribution of that strategy.
816 """
818 def __init__(self, strategies: Sequence[SearchStrategy[Ex]]):
819 super().__init__()
820 self.original_strategies = tuple(strategies)
821 self.__element_strategies: Sequence[SearchStrategy[Ex]] | None = None
822 self.__in_branches = False
823 self._branches_lock = RLock()
825 def calc_is_empty(self, recur: RecurT) -> bool:
826 return all(recur(e) for e in self.original_strategies)
828 def calc_has_reusable_values(self, recur: RecurT) -> bool:
829 return all(recur(e) for e in self.original_strategies)
831 def calc_is_cacheable(self, recur: RecurT) -> bool:
832 return all(recur(e) for e in self.original_strategies)
834 @property
835 def element_strategies(self) -> Sequence[SearchStrategy[Ex]]:
836 if self.__element_strategies is None:
837 # While strategies are hashable, they use object.__hash__ and are
838 # therefore distinguished only by identity.
839 #
840 # In principle we could "just" define a __hash__ method
841 # (and __eq__, but that's easy in terms of type() and hash())
842 # to make this more powerful, but this is harder than it sounds:
843 #
844 # 1. Strategies are often distinguished by non-hashable attributes,
845 # or by attributes that have the same hash value ("^.+" / b"^.+").
846 # 2. LazyStrategy: can't reify the wrapped strategy without breaking
847 # laziness, so there's a hash each for the lazy and the nonlazy.
848 #
849 # Having made several attempts, the minor benefits of making strategies
850 # hashable are simply not worth the engineering effort it would take.
851 # See also issues #2291 and #2327.
852 seen: set[SearchStrategy] = {self}
853 strategies: list[SearchStrategy] = []
854 for arg in self.original_strategies:
855 check_strategy(arg)
856 if not arg.is_empty:
857 for s in arg.branches:
858 if s not in seen and not s.is_empty:
859 seen.add(s)
860 strategies.append(s)
861 self.__element_strategies = strategies
862 return self.__element_strategies
864 def calc_label(self) -> int:
865 return combine_labels(
866 self.class_label, *(p.label for p in self.original_strategies)
867 )
869 def do_draw(self, data: ConjectureData) -> Ex:
870 strategy = data.draw(
871 SampledFromStrategy(self.element_strategies).filter(
872 lambda s: not s.is_currently_empty(data)
873 )
874 )
875 return data.draw(strategy)
877 def __repr__(self) -> str:
878 return "one_of({})".format(", ".join(map(repr, self.original_strategies)))
880 def do_validate(self) -> None:
881 for e in self.element_strategies:
882 e.validate()
884 @property
885 def branches(self) -> Sequence[SearchStrategy[Ex]]:
886 if self.__element_strategies is not None:
887 # common fast path which avoids the lock
888 return self.element_strategies
890 with self._branches_lock:
891 if not self.__in_branches:
892 try:
893 self.__in_branches = True
894 return self.element_strategies
895 finally:
896 self.__in_branches = False
897 else:
898 return [self]
900 @overload
901 def filter(
902 self, condition: Callable[[Ex], TypeGuard[T]]
903 ) -> "SearchStrategy[T]": ...
904 @overload
905 def filter(self, condition: Callable[[Ex], Any]) -> "SearchStrategy[Ex]": ...
906 def filter(self, condition):
907 return FilteredStrategy(
908 OneOfStrategy([s.filter(condition) for s in self.original_strategies]),
909 conditions=(),
910 )
913@overload
914def one_of(
915 __args: Sequence[SearchStrategy[Ex]],
916) -> SearchStrategy[Ex]: # pragma: no cover
917 ...
920@overload
921def one_of(__a1: SearchStrategy[Ex]) -> SearchStrategy[Ex]: # pragma: no cover
922 ...
925@overload
926def one_of(
927 __a1: SearchStrategy[Ex], __a2: SearchStrategy[T]
928) -> SearchStrategy[Ex | T]: # pragma: no cover
929 ...
932@overload
933def one_of(
934 __a1: SearchStrategy[Ex], __a2: SearchStrategy[T], __a3: SearchStrategy[T3]
935) -> SearchStrategy[Ex | T | T3]: # pragma: no cover
936 ...
939@overload
940def one_of(
941 __a1: SearchStrategy[Ex],
942 __a2: SearchStrategy[T],
943 __a3: SearchStrategy[T3],
944 __a4: SearchStrategy[T4],
945) -> SearchStrategy[Ex | T | T3 | T4]: # pragma: no cover
946 ...
949@overload
950def one_of(
951 __a1: SearchStrategy[Ex],
952 __a2: SearchStrategy[T],
953 __a3: SearchStrategy[T3],
954 __a4: SearchStrategy[T4],
955 __a5: SearchStrategy[T5],
956) -> SearchStrategy[Ex | T | T3 | T4 | T5]: # pragma: no cover
957 ...
960@overload
961def one_of(*args: SearchStrategy[Any]) -> SearchStrategy[Any]: # pragma: no cover
962 ...
965@defines_strategy(eager=True)
966def one_of(
967 *args: Sequence[SearchStrategy[Any]] | SearchStrategy[Any],
968) -> SearchStrategy[Any]:
969 # Mypy workaround alert: Any is too loose above; the return parameter
970 # should be the union of the input parameters. Unfortunately, Mypy <=0.600
971 # raises errors due to incompatible inputs instead. See #1270 for links.
972 # v0.610 doesn't error; it gets inference wrong for 2+ arguments instead.
973 """Return a strategy which generates values from any of the argument
974 strategies.
976 This may be called with one iterable argument instead of multiple
977 strategy arguments, in which case ``one_of(x)`` and ``one_of(*x)`` are
978 equivalent.
980 Examples from this strategy will generally shrink to ones that come from
981 strategies earlier in the list, then shrink according to behaviour of the
982 strategy that produced them. In order to get good shrinking behaviour,
983 try to put simpler strategies first. e.g. ``one_of(none(), text())`` is
984 better than ``one_of(text(), none())``.
986 This is especially important when using recursive strategies. e.g.
987 ``x = st.deferred(lambda: st.none() | st.tuples(x, x))`` will shrink well,
988 but ``x = st.deferred(lambda: st.tuples(x, x) | st.none())`` will shrink
989 very badly indeed.
990 """
991 if len(args) == 1 and not isinstance(args[0], SearchStrategy):
992 try:
993 args = tuple(args[0])
994 except TypeError:
995 pass
996 if len(args) == 1 and isinstance(args[0], SearchStrategy):
997 # This special-case means that we can one_of over lists of any size
998 # without incurring any performance overhead when there is only one
999 # strategy, and keeps our reprs simple.
1000 return args[0]
1001 if args and not any(isinstance(a, SearchStrategy) for a in args):
1002 # And this special case is to give a more-specific error message if it
1003 # seems that the user has confused `one_of()` for `sampled_from()`;
1004 # the remaining validation is left to OneOfStrategy. See PR #2627.
1005 raise InvalidArgument(
1006 f"Did you mean st.sampled_from({list(args)!r})? st.one_of() is used "
1007 "to combine strategies, but all of the arguments were of other types."
1008 )
1009 # we've handled the case where args is a one-element sequence [(s1, s2, ...)]
1010 # above, so we can assume it's an actual sequence of strategies.
1011 args = cast(Sequence[SearchStrategy], args)
1012 return OneOfStrategy(args)
1015class MappedStrategy(SearchStrategy[MappedTo], Generic[MappedFrom, MappedTo]):
1016 """A strategy which is defined purely by conversion to and from another
1017 strategy.
1019 Its parameter and distribution come from that other strategy.
1020 """
1022 def __init__(
1023 self,
1024 strategy: SearchStrategy[MappedFrom],
1025 pack: Callable[[MappedFrom], MappedTo],
1026 ) -> None:
1027 super().__init__()
1028 self.mapped_strategy = strategy
1029 self.pack = pack
1031 def calc_is_empty(self, recur: RecurT) -> bool:
1032 return recur(self.mapped_strategy)
1034 def calc_is_cacheable(self, recur: RecurT) -> bool:
1035 return recur(self.mapped_strategy)
1037 def __repr__(self) -> str:
1038 if not hasattr(self, "_cached_repr"):
1039 self._cached_repr = f"{self.mapped_strategy!r}.map({get_pretty_function_description(self.pack)})"
1040 return self._cached_repr
1042 def do_validate(self) -> None:
1043 self.mapped_strategy.validate()
1045 def do_draw(self, data: ConjectureData) -> MappedTo:
1046 with warnings.catch_warnings():
1047 if isinstance(self.pack, type) and issubclass(
1048 self.pack, (abc.Mapping, abc.Set)
1049 ):
1050 warnings.simplefilter("ignore", BytesWarning)
1051 for _ in range(3):
1052 try:
1053 data.start_span(MAPPED_SEARCH_STRATEGY_DO_DRAW_LABEL)
1054 x = data.draw(self.mapped_strategy)
1055 result = self.pack(x)
1056 data.stop_span()
1057 current_build_context().record_call(
1058 result, self.pack, args=[x], kwargs={}
1059 )
1060 return result
1061 except UnsatisfiedAssumption:
1062 data.stop_span(discard=True)
1063 raise UnsatisfiedAssumption
1065 @property
1066 def branches(self) -> Sequence[SearchStrategy[MappedTo]]:
1067 return [
1068 MappedStrategy(strategy, pack=self.pack)
1069 for strategy in self.mapped_strategy.branches
1070 ]
1072 @overload
1073 def filter(
1074 self, condition: Callable[[MappedTo], TypeGuard[T]]
1075 ) -> "SearchStrategy[T]": ...
1076 @overload
1077 def filter(
1078 self, condition: Callable[[MappedTo], Any]
1079 ) -> "SearchStrategy[MappedTo]": ...
1080 def filter(self, condition):
1081 # Includes a special case so that we can rewrite filters on collection
1082 # lengths, when most collections are `st.lists(...).map(the_type)`.
1083 ListStrategy = _list_strategy_type()
1084 if not isinstance(self.mapped_strategy, ListStrategy) or not (
1085 (isinstance(self.pack, type) and issubclass(self.pack, abc.Collection))
1086 or self.pack in _collection_ish_functions()
1087 ):
1088 return super().filter(condition)
1090 # Check whether our inner list strategy can rewrite this filter condition.
1091 # If not, discard the result and _only_ apply a new outer filter.
1092 new = ListStrategy.filter(self.mapped_strategy, condition)
1093 if getattr(new, "filtered_strategy", None) is self.mapped_strategy:
1094 return super().filter(condition) # didn't rewrite
1096 # Apply a new outer filter even though we rewrote the inner strategy,
1097 # because some collections can change the list length (dict, set, etc).
1098 return FilteredStrategy(type(self)(new, self.pack), conditions=(condition,))
1101@lru_cache
1102def _list_strategy_type() -> Any:
1103 from hypothesis.strategies._internal.collections import ListStrategy
1105 return ListStrategy
1108def _collection_ish_functions() -> Sequence[Any]:
1109 funcs = [sorted]
1110 if np := sys.modules.get("numpy"):
1111 # c.f. https://numpy.org/doc/stable/reference/routines.array-creation.html
1112 # Probably only `np.array` and `np.asarray` will be used in practice,
1113 # but why should that stop us when we've already gone this far?
1114 funcs += [
1115 np.empty_like,
1116 np.eye,
1117 np.identity,
1118 np.ones_like,
1119 np.zeros_like,
1120 np.array,
1121 np.asarray,
1122 np.asanyarray,
1123 np.ascontiguousarray,
1124 np.asmatrix,
1125 np.copy,
1126 np.rec.array,
1127 np.rec.fromarrays,
1128 np.rec.fromrecords,
1129 np.diag,
1130 # bonus undocumented functions from tab-completion:
1131 np.asarray_chkfinite,
1132 np.asfortranarray,
1133 ]
1135 return funcs
1138filter_not_satisfied = UniqueIdentifier("filter not satisfied")
1141class FilteredStrategy(SearchStrategy[Ex]):
1142 def __init__(
1143 self, strategy: SearchStrategy[Ex], conditions: tuple[Callable[[Ex], Any], ...]
1144 ):
1145 super().__init__()
1146 if isinstance(strategy, FilteredStrategy):
1147 # Flatten chained filters into a single filter with multiple conditions.
1148 self.flat_conditions: tuple[Callable[[Ex], Any], ...] = (
1149 strategy.flat_conditions + conditions
1150 )
1151 self.filtered_strategy: SearchStrategy[Ex] = strategy.filtered_strategy
1152 else:
1153 self.flat_conditions = conditions
1154 self.filtered_strategy = strategy
1156 assert isinstance(self.flat_conditions, tuple)
1157 assert not isinstance(self.filtered_strategy, FilteredStrategy)
1159 self.__condition: Callable[[Ex], Any] | None = None
1161 def calc_is_empty(self, recur: RecurT) -> bool:
1162 return recur(self.filtered_strategy)
1164 def calc_is_cacheable(self, recur: RecurT) -> bool:
1165 return recur(self.filtered_strategy)
1167 def __repr__(self) -> str:
1168 if not hasattr(self, "_cached_repr"):
1169 self._cached_repr = "{!r}{}".format(
1170 self.filtered_strategy,
1171 "".join(
1172 f".filter({get_pretty_function_description(cond)})"
1173 for cond in self.flat_conditions
1174 ),
1175 )
1176 return self._cached_repr
1178 def do_validate(self) -> None:
1179 # Start by validating our inner filtered_strategy. If this was a LazyStrategy,
1180 # validation also reifies it so that subsequent calls to e.g. `.filter()` will
1181 # be passed through.
1182 self.filtered_strategy.validate()
1183 # So now we have a reified inner strategy, we'll replay all our saved
1184 # predicates in case some or all of them can be rewritten. Note that this
1185 # replaces the `fresh` strategy too!
1186 fresh = self.filtered_strategy
1187 for cond in self.flat_conditions:
1188 fresh = fresh.filter(cond)
1189 if isinstance(fresh, FilteredStrategy):
1190 # In this case we have at least some non-rewritten filter predicates,
1191 # so we just re-initialize the strategy.
1192 FilteredStrategy.__init__(
1193 self, fresh.filtered_strategy, fresh.flat_conditions
1194 )
1195 else:
1196 # But if *all* the predicates were rewritten... well, do_validate() is
1197 # an in-place method so we still just re-initialize the strategy!
1198 FilteredStrategy.__init__(self, fresh, ())
1200 @overload
1201 def filter(
1202 self, condition: Callable[[Ex], TypeGuard[T]]
1203 ) -> "FilteredStrategy[T]": ...
1204 @overload
1205 def filter(self, condition: Callable[[Ex], Any]) -> "FilteredStrategy[Ex]": ...
1206 def filter(self, condition):
1207 # If we can, it's more efficient to rewrite our strategy to satisfy the
1208 # condition. We therefore exploit the fact that the order of predicates
1209 # doesn't matter (`f(x) and g(x) == g(x) and f(x)`) by attempting to apply
1210 # condition directly to our filtered strategy as the inner-most filter.
1211 out = self.filtered_strategy.filter(condition)
1212 # If it couldn't be rewritten, we'll get a new FilteredStrategy - and then
1213 # combine the conditions of each in our expected newest=last order.
1214 if isinstance(out, FilteredStrategy):
1215 return FilteredStrategy(
1216 out.filtered_strategy, self.flat_conditions + out.flat_conditions
1217 )
1218 # But if it *could* be rewritten, we can return the more efficient form!
1219 return FilteredStrategy(out, self.flat_conditions)
1221 @property
1222 def condition(self) -> Callable[[Ex], Any]:
1223 # We write this defensively to avoid any threading race conditions
1224 # with our manual FilteredStrategy.__init__ for filter-rewriting.
1225 # See https://github.com/HypothesisWorks/hypothesis/pull/4522.
1226 if (condition := self.__condition) is not None:
1227 return condition
1229 if len(self.flat_conditions) == 1:
1230 # Avoid an extra indirection in the common case of only one condition.
1231 condition = self.flat_conditions[0]
1232 elif len(self.flat_conditions) == 0:
1233 # Possible, if unlikely, due to filter predicate rewriting
1234 condition = lambda _: True
1235 else:
1236 condition = lambda x: all(cond(x) for cond in self.flat_conditions)
1237 self.__condition = condition
1238 return condition
1240 def do_draw(self, data: ConjectureData) -> Ex:
1241 result = self.do_filtered_draw(data)
1242 if result is not filter_not_satisfied:
1243 return cast(Ex, result)
1245 data.mark_invalid(f"Aborted test because unable to satisfy {self!r}")
1247 def do_filtered_draw(self, data: ConjectureData) -> Ex | UniqueIdentifier:
1248 for i in range(3):
1249 data.start_span(FILTERED_SEARCH_STRATEGY_DO_DRAW_LABEL)
1250 value = data.draw(self.filtered_strategy)
1251 if self.condition(value):
1252 data.stop_span()
1253 return value
1254 else:
1255 data.stop_span(discard=True)
1256 if i == 0:
1257 data.events[f"Retried draw from {self!r} to satisfy filter"] = ""
1259 return filter_not_satisfied
1261 @property
1262 def branches(self) -> Sequence[SearchStrategy[Ex]]:
1263 return [
1264 FilteredStrategy(strategy=strategy, conditions=self.flat_conditions)
1265 for strategy in self.filtered_strategy.branches
1266 ]
1269@check_function
1270def check_strategy(arg: object, name: str = "") -> None:
1271 assert isinstance(name, str)
1272 if not isinstance(arg, SearchStrategy):
1273 hint = ""
1274 if isinstance(arg, (list, tuple)):
1275 hint = ", such as st.sampled_from({}),".format(name or "...")
1276 if name:
1277 name += "="
1278 raise InvalidArgument(
1279 f"Expected a SearchStrategy{hint} but got {name}{arg!r} "
1280 f"(type={type(arg).__name__})"
1281 )