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 sys
12from inspect import (
13 Parameter,
14 Signature,
15 isasyncgenfunction,
16 iscoroutinefunction,
17 isgeneratorfunction,
18)
19from weakref import WeakKeyDictionary
21from hypothesis.control import note, should_note
22from hypothesis.errors import InvalidState
23from hypothesis.internal.reflection import (
24 convert_positional_arguments,
25 get_signature,
26 nicerepr,
27 proxies,
28 repr_call,
29)
30from hypothesis.strategies._internal.lazy import unwrap_strategies
31from hypothesis.strategies._internal.strategies import (
32 RecurT,
33 SampledFromStrategy,
34 SearchStrategy,
35)
36from hypothesis.utils.conventions import UniqueIdentifier
38can_vary = UniqueIdentifier("can_vary")
41async def _checkpoint(): # pragma: no cover # depends on installed frameworks
42 # Generated async functions follow Trio-style checkpoint semantics, using
43 # anyio or sniffio to find the right way to checkpoint if the user's async
44 # framework might not be asyncio.
45 if anyio := sys.modules.get("anyio"):
46 await anyio.lowlevel.checkpoint()
47 return
48 if sniffio := sys.modules.get("sniffio"):
49 if sniffio.current_async_library() == "trio":
50 await sys.modules["trio"].lowlevel.checkpoint()
51 return
52 import asyncio # deferred to keep `import hypothesis` fast
54 await asyncio.sleep(0)
57class FunctionStrategy(SearchStrategy):
58 def __init__(self, like, returns, pure):
59 super().__init__()
60 self.like = like
61 self.returns = returns
62 self.pure = pure
63 # If `returns` can only generate a single value - just(), none(), or a
64 # one-element sampled_from() - we know what our functions return before
65 # they are ever called, and show them as constant lambdas instead. We
66 # have to unwrap lazy wrappers to see that, and to bail out if any
67 # .map() or .filter() was applied, since those transformations could
68 # change or reject the value we'd otherwise report.
69 unwrapped = unwrap_strategies(returns)
70 if (
71 isinstance(unwrapped, SampledFromStrategy)
72 and len(unwrapped.elements) == 1
73 and not unwrapped._transformations
74 ):
75 self._constant = unwrapped.elements[0]
76 else:
77 self._constant = can_vary
78 # Using wekrefs-to-generated-functions means that the cache can be
79 # garbage-collected at the end of each example, reducing memory use.
80 self._cache = WeakKeyDictionary()
82 def _pretty_constant_function(self, p, cycle):
83 # Annotations are valid in a signature, but not in a lambda.
84 sig = get_signature(self.like, follow_wrapped=False)
85 params = str(
86 sig.replace(
87 parameters=[
88 param.replace(annotation=Parameter.empty)
89 for param in sig.parameters.values()
90 ],
91 return_annotation=Signature.empty,
92 )
93 )[1:-1]
94 p.text(f"lambda {params}: " if params else "lambda: ")
95 p.pretty(self._constant)
97 def calc_is_empty(self, recur: RecurT) -> bool:
98 return recur(self.returns)
100 def do_draw(self, data):
101 # If we know what the function returns, we show it as a constant lambda
102 # instead of noting every call - the notes would be redundant. A
103 # lambda is a poor description of an async function though, and
104 # generator kinds are never constant, so this is for plain functions.
105 varies = self._constant is can_vary or iscoroutinefunction(self.like)
107 def draw_value(args, kwargs):
108 # `pure=True` is rejected for non-plain `like`s, so only the plain
109 # `inner` below can ever take the caching branch.
110 if data.frozen:
111 raise InvalidState(
112 f"This generated {nicerepr(self.like)} function can only "
113 "be called within the scope of the @given that created it."
114 )
115 if self.pure:
116 args, kwargs = convert_positional_arguments(self.like, args, kwargs)
117 key = (args, frozenset(kwargs.items()))
118 cache = self._cache.setdefault(inner, {})
119 if key not in cache:
120 cache[key] = data.draw(self.returns)
121 # optimization to avoid needless repr_call
122 if varies and should_note():
123 rep = repr_call(self.like, args, kwargs, reorder=False)
124 note(f"Called function: {rep} -> {cache[key]!r}")
125 return cache[key]
126 else:
127 val = data.draw(self.returns)
128 if varies and should_note():
129 rep = repr_call(self.like, args, kwargs, reorder=False)
130 note(f"Called function: {rep} -> {val!r}")
131 return val
133 # Define an inner function of the same kind as `like`, so that the
134 # proxy (see `proxies`) is a coroutine, generator, or async-generator
135 # function whenever `like` is. For generator kinds, the drawn value
136 # is a list of values to yield.
137 if iscoroutinefunction(self.like):
139 @proxies(self.like)
140 async def inner(*args, **kwargs):
141 value = draw_value(args, kwargs)
142 await _checkpoint()
143 return value
145 elif isasyncgenfunction(self.like):
147 @proxies(self.like)
148 async def inner(*args, **kwargs):
149 for value in draw_value(args, kwargs):
150 await _checkpoint()
151 yield value
152 await _checkpoint()
154 elif isgeneratorfunction(self.like):
156 @proxies(self.like)
157 def inner(*args, **kwargs):
158 yield from draw_value(args, kwargs)
160 else:
162 @proxies(self.like)
163 def inner(*args, **kwargs):
164 return draw_value(args, kwargs)
166 if not varies:
167 inner._repr_pretty_ = self._pretty_constant_function
168 return inner