Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/hypothesis/control.py: 47%
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 inspect
12import math
13import random
14from collections import defaultdict
15from collections.abc import Callable, Generator, Sequence
16from contextlib import contextmanager
17from types import TracebackType
18from typing import TYPE_CHECKING, Any, Literal, NoReturn, Optional, overload
19from weakref import WeakKeyDictionary
21from hypothesis import Verbosity, settings
22from hypothesis.errors import InvalidArgument, UnsatisfiedAssumption
23from hypothesis.internal.compat import BaseExceptionGroup
24from hypothesis.internal.conjecture.data import ConjectureData
25from hypothesis.internal.observability import observability_enabled
26from hypothesis.internal.reflection import get_pretty_function_description
27from hypothesis.internal.validation import check_type
28from hypothesis.reporting import report, verbose_report
29from hypothesis.utils.deprecation import note_deprecation
30from hypothesis.utils.dynamicvariables import DynamicVariable
31from hypothesis.vendor.pretty import ArgLabelsT, IDKey, PrettyPrintFunction, pretty
33if TYPE_CHECKING:
34 from typing_extensions import Self
37def _calling_function_location(what: str, frame: Any) -> str:
38 where = frame.f_back
39 return f"{what}() in {where.f_code.co_name} (line {where.f_lineno})"
42def _calling_frame_location(frame: Any) -> str:
43 where = frame.f_back
44 return f"{where.f_code.co_filename}:{where.f_lineno}"
47def reject() -> NoReturn:
48 if _current_build_context.value is None:
49 note_deprecation(
50 "Using `reject` outside a property-based test is deprecated",
51 since="2023-09-25",
52 has_codemod=False,
53 )
54 frame = inspect.currentframe()
55 where = _calling_function_location("reject", frame)
56 if currently_in_test_context():
57 counts = current_build_context().data._observability_predicates[where]
58 counts.update_count(condition=False)
59 raise UnsatisfiedAssumption(where, location=_calling_frame_location(frame))
62@overload
63def assume(condition: Literal[False] | None) -> NoReturn: ...
64@overload
65def assume(condition: object) -> Literal[True]: ...
68def assume(condition: object) -> Literal[True]:
69 """Calling ``assume`` is like an :ref:`assert <python:assert>` that marks
70 the |test case| as bad, rather than failing the test.
72 This allows you to specify properties that you *assume* will be
73 true, and let Hypothesis try to avoid similar test cases in future.
74 """
75 if _current_build_context.value is None:
76 note_deprecation(
77 "Using `assume` outside a property-based test is deprecated",
78 since="2023-09-25",
79 has_codemod=False,
80 )
81 if observability_enabled() or not condition:
82 frame = inspect.currentframe()
83 where = _calling_function_location("assume", frame)
84 if observability_enabled() and currently_in_test_context():
85 counts = current_build_context().data._observability_predicates[where]
86 counts.update_count(condition=bool(condition))
87 if not condition:
88 raise UnsatisfiedAssumption(
89 f"failed to satisfy {where}",
90 location=_calling_frame_location(frame),
91 )
92 return True
95_current_build_context = DynamicVariable[Optional["BuildContext"]](None)
98def currently_in_test_context() -> bool:
99 """Return ``True`` if the calling code is currently running inside an
100 |@given| or :ref:`stateful <stateful>` test, and ``False`` otherwise.
102 This is useful for third-party integrations and assertion helpers which
103 may be called from either traditional or property-based tests, and can only
104 use e.g. |assume| or |target| in the latter case.
105 """
106 return _current_build_context.value is not None
109def current_build_context() -> "BuildContext":
110 context = _current_build_context.value
111 if context is None:
112 raise InvalidArgument("No build context registered")
113 return context
116@contextmanager
117def deprecate_random_in_strategy(fmt: str, *args: Any) -> Generator[None, None, None]:
118 from hypothesis.internal import entropy
120 state_before = random.getstate()
121 yield
122 state_after = random.getstate()
123 if (
124 # there is a threading race condition here with deterministic_PRNG. Say
125 # we have two threads 1 and 2. We start in global random state A, and
126 # deterministic_PRNG sets to global random state B (which is constant across
127 # threads since we seed to 0 unconditionally). Then we might have state
128 # transitions:
129 #
130 # [1] [2]
131 # A -> B deterministic_PRNG().__enter__
132 # B ->B deterministic_PRNG().__enter__
133 # state_before = B deprecate_random_in_strategy.__enter__
134 # B -> A deterministic_PRNG().__exit__
135 # state_after = A deprecate_random_in_strategy.__exit__
136 #
137 # where state_before != state_after because a different thread has reset
138 # the global random state.
139 #
140 # To fix this, we track the known random states set by deterministic_PRNG,
141 # and will not note a deprecation if it matches one of those.
142 state_after != state_before
143 and hash(state_after) not in entropy._known_random_state_hashes
144 ):
145 note_deprecation(
146 "Do not use the `random` module inside strategies; instead "
147 "consider `st.randoms()`, `st.sampled_from()`, etc. " + fmt.format(*args),
148 since="2024-02-05",
149 has_codemod=False,
150 stacklevel=1,
151 )
154class BuildContext:
155 def __init__(
156 self,
157 data: ConjectureData,
158 *,
159 is_final: bool = False,
160 wrapped_test: Callable,
161 ) -> None:
162 self.data = data
163 self.tasks: list[Callable[[], Any]] = []
164 self.is_final = is_final
165 self.wrapped_test = wrapped_test
167 # Use defaultdict(list) here to handle the possibility of having multiple
168 # functions registered for the same object (due to caching, small ints, etc).
169 # The printer will discard duplicates which return different representations.
170 self.known_object_printers: dict[IDKey, list[PrettyPrintFunction]] = (
171 defaultdict(list)
172 )
174 def record_call(
175 self,
176 obj: object,
177 func: object,
178 *,
179 args: Sequence[object],
180 kwargs: dict[str, object],
181 arg_labels: ArgLabelsT | None = None,
182 ) -> None:
183 self.known_object_printers[IDKey(obj)].append(
184 lambda obj, p, cycle, *, _func=func, _arg_labels=arg_labels: p.maybe_repr_known_object_as_call( # type: ignore
185 obj,
186 cycle,
187 get_pretty_function_description(_func),
188 args,
189 kwargs,
190 arg_labels=_arg_labels,
191 )
192 )
194 def prep_args_kwargs_from_strategies(
195 self,
196 kwarg_strategies: dict[str, Any],
197 ) -> tuple[dict[str, Any], ArgLabelsT]:
198 arg_labels: ArgLabelsT = {}
199 kwargs: dict[str, Any] = {}
201 for k, s in kwarg_strategies.items():
202 with (
203 self.data.track_arg_label(k) as arg_label,
204 deprecate_random_in_strategy("from {}={!r}", k, s),
205 ):
206 kwargs[k] = self.data.draw(s, observe_as=f"generate:{k}")
207 arg_labels |= arg_label
209 return kwargs, arg_labels
211 def __enter__(self) -> "Self":
212 self.assign_variable = _current_build_context.with_value(self)
213 self.assign_variable.__enter__()
214 return self
216 def __exit__(
217 self,
218 exc_type: type[BaseException] | None,
219 exc_value: BaseException | None,
220 tb: TracebackType | None,
221 ) -> None:
222 self.assign_variable.__exit__(exc_type, exc_value, tb)
223 errors = []
224 for task in self.tasks:
225 try:
226 task()
227 except BaseException as err:
228 errors.append(err)
229 if errors:
230 if len(errors) == 1:
231 raise errors[0] from exc_value
232 raise BaseExceptionGroup("Cleanup failed", errors) from exc_value
235def cleanup(teardown: Callable[[], Any]) -> None:
236 """Register a function to be called when the current test has finished
237 executing. Any exceptions thrown in teardown will be printed but not
238 rethrown.
240 Inside a test this isn't very interesting, because you can just use
241 a finally block, but note that you can use this inside map, flatmap,
242 etc. in order to e.g. insist that a value is closed at the end.
243 """
244 context = _current_build_context.value
245 if context is None:
246 raise InvalidArgument("Cannot register cleanup outside of build context")
247 context.tasks.append(teardown)
250def should_note() -> bool:
251 context = _current_build_context.value
252 if context is None:
253 raise InvalidArgument("Cannot make notes outside of a test")
254 assert settings.default is not None
255 return context.is_final or settings.default.verbosity >= Verbosity.verbose
258def note(value: object) -> None:
259 """
260 Record a note on this |test case|. Non-string values will be automatically converted
261 to a string.
263 This value is reported for the |minimal failing test case|, and on |Verbosity.verbose|
264 or higher.
266 Notes are also recorded in the ``metadata.notes`` key of each test case
267 observation, if :ref:`observability <observability>` is enabled.
268 """
269 should_report = should_note()
270 if should_report or observability_enabled():
271 if not isinstance(value, str):
272 value = pretty(value)
273 context = _current_build_context.value
274 assert context is not None
275 if observability_enabled():
276 context.data.note(value)
277 if should_report:
278 report(value)
281def event(value: str, payload: Any = "") -> None:
282 """Record an event that occurred during this test. Statistics on the number of test
283 runs with each event will be reported at the end if you run Hypothesis in
284 statistics reporting mode.
286 Event values should be strings or convertible to them. If an optional
287 payload is given, it will be included in the string for :ref:`statistics`.
288 """
289 context = _current_build_context.value
290 if context is None:
291 raise InvalidArgument("Cannot record events outside of a test")
293 avoid_realization = context.data.provider.avoid_realization
294 payload = _serialize_event(
295 payload, allowed_types=(str, int, float), avoid_realization=avoid_realization
296 )
297 value = _serialize_event(value, avoid_realization=avoid_realization)
298 context.data.events[value] = payload
301_events_to_strings: WeakKeyDictionary[Any, str] = WeakKeyDictionary()
304def _serialize_event(
305 event: Any, *, allowed_types: tuple[type, ...] = (str,), avoid_realization: bool
306) -> Any:
307 if isinstance(event, allowed_types):
308 return event
310 # _events_to_strings is a cache which persists across iterations, causing
311 # problems for symbolic backends. see
312 # https://github.com/pschanely/hypothesis-crosshair/issues/41
313 if avoid_realization:
314 return str(event)
316 try:
317 return _events_to_strings[event]
318 except (KeyError, TypeError):
319 pass
321 result = str(event)
322 try:
323 _events_to_strings[event] = result
324 except TypeError:
325 pass
326 return result
329def target(observation: int | float, *, label: str = "") -> int | float:
330 """Calling this function with an ``int`` or ``float`` observation gives it feedback
331 with which to guide our search for inputs that will cause an error, in
332 addition to all the usual heuristics. Observations must always be finite.
334 Hypothesis will try to maximize the observed value over several |test cases|;
335 almost any metric will work so long as it makes sense to increase it.
336 For example, ``-abs(error)`` is a metric that increases as ``error``
337 approaches zero.
339 Example metrics:
341 - Number of elements in a collection, or tasks in a queue
342 - Mean or maximum runtime of a task (or both, if you use ``label``)
343 - Compression ratio for data (perhaps per-algorithm or per-level)
344 - Number of steps taken by a state machine
346 The optional ``label`` argument can be used to distinguish between
347 and therefore separately optimise distinct observations, such as the
348 mean and standard deviation of a dataset. It is an error to call
349 ``target()`` with any label more than once per test case.
351 .. note::
352 The more test cases you run, the better this technique works.
354 As a rule of thumb, the targeting effect is noticeable above
355 :obj:`max_examples=1000 <hypothesis.settings.max_examples>`,
356 and immediately obvious by around ten thousand test cases
357 *per label* used by your test.
359 :ref:`statistics` include the best score seen for each label,
360 which can help avoid `the threshold problem
361 <https://hypothesis.works/articles/threshold-problem/>`__ when the minimal
362 test case shrinks right down to the threshold of failure (:issue:`2180`).
363 """
364 check_type((int, float), observation, "observation")
365 if not math.isfinite(observation):
366 raise InvalidArgument(f"{observation=} must be a finite float.")
367 check_type(str, label, "label")
369 context = _current_build_context.value
370 if context is None:
371 raise InvalidArgument(
372 "Calling target() outside of a test is invalid. "
373 "Consider guarding this call with `if currently_in_test_context(): ...`"
374 )
375 elif context.data.provider.avoid_realization:
376 # We could in principle realize this in the engine, but it seems more
377 # efficient to have our alternative backend optimize it for us.
378 # See e.g. https://github.com/pschanely/hypothesis-crosshair/issues/3
379 return observation # pragma: no cover
380 verbose_report(f"Saw target({observation!r}, {label=})")
382 if label in context.data.target_observations:
383 raise InvalidArgument(
384 f"Calling target({observation!r}, {label=}) would overwrite "
385 f"target({context.data.target_observations[label]!r}, {label=})"
386 )
387 else:
388 context.data.target_observations[label] = observation
390 return observation