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