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