Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/hypothesis/core.py: 34%
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/.
11"""This module provides the core primitives of Hypothesis, such as given."""
13import base64
14import contextlib
15import dataclasses
16import datetime
17import inspect
18import io
19import math
20import os
21import sys
22import threading
23import time
24import types
25import unittest
26import warnings
27import zlib
28from collections import defaultdict
29from collections.abc import Callable, Coroutine, Generator, Hashable, Iterable, Sequence
30from dataclasses import dataclass, field
31from functools import partial
32from inspect import Parameter
33from random import Random
34from threading import Lock
35from types import EllipsisType
36from typing import (
37 Any,
38 BinaryIO,
39 TypeVar,
40 overload,
41)
42from unittest import TestCase
44from hypothesis import strategies as st
45from hypothesis._settings import (
46 HealthCheck,
47 Phase,
48 Verbosity,
49 all_settings,
50 local_settings,
51 settings as Settings,
52)
53from hypothesis.control import (
54 BuildContext,
55 current_build_context,
56 currently_in_test_context,
57)
58from hypothesis.database import choices_from_bytes, choices_to_bytes
59from hypothesis.errors import (
60 BackendCannotProceed,
61 DeadlineExceeded,
62 DidNotReproduce,
63 FailedHealthCheck,
64 FlakyFailure,
65 FlakyReplay,
66 Found,
67 Frozen,
68 HypothesisException,
69 HypothesisWarning,
70 InvalidArgument,
71 NoSuchExample,
72 StopTest,
73 Unsatisfiable,
74 UnsatisfiedAssumption,
75)
76from hypothesis.internal import observability
77from hypothesis.internal.compat import (
78 PYPY,
79 add_note,
80 bad_django_TestCase,
81 get_type_hints,
82 int_from_bytes,
83)
84from hypothesis.internal.conjecture.choice import ChoiceT
85from hypothesis.internal.conjecture.data import ConjectureData, Status
86from hypothesis.internal.conjecture.engine import (
87 BUFFER_SIZE,
88 ConjectureRunner,
89 ExitReason,
90)
91from hypothesis.internal.conjecture.junkdrawer import (
92 ensure_free_stackframes,
93 gc_cumulative_time,
94)
95from hypothesis.internal.conjecture.providers import (
96 BytestringProvider,
97 PrimitiveProvider,
98)
99from hypothesis.internal.conjecture.shrinker import sort_key
100from hypothesis.internal.entropy import deterministic_PRNG
101from hypothesis.internal.escalation import (
102 InterestingOrigin,
103 _last_frame_location,
104 current_pytest_item,
105 format_exception,
106 get_trimmed_traceback,
107 is_hypothesis_file,
108)
109from hypothesis.internal.healthcheck import fail_health_check
110from hypothesis.internal.observability import (
111 InfoObservation,
112 InfoObservationType,
113 deliver_observation,
114 make_testcase,
115 observability_enabled,
116)
117from hypothesis.internal.reflection import (
118 convert_positional_arguments,
119 define_function_signature,
120 function_digest,
121 get_pretty_function_description,
122 get_signature,
123 impersonate,
124 is_mock,
125 nicerepr,
126 proxies,
127 repr_call,
128)
129from hypothesis.internal.scrutineer import (
130 MONITORING_TOOL_ID,
131 Trace,
132 Tracer,
133 explanatory_lines,
134 tractable_coverage_report,
135)
136from hypothesis.internal.validation import check_type
137from hypothesis.reporting import (
138 current_verbosity,
139 report,
140 verbose_report,
141 with_reporter,
142)
143from hypothesis.statistics import describe_statistics, describe_targets, note_statistics
144from hypothesis.strategies._internal.misc import NOTHING
145from hypothesis.strategies._internal.strategies import (
146 Ex,
147 SearchStrategy,
148 check_strategy,
149)
150from hypothesis.utils.conventions import not_set
151from hypothesis.utils.threading import ThreadLocal
152from hypothesis.vendor.pretty import RepresentationPrinter
153from hypothesis.version import __version__
155TestFunc = TypeVar("TestFunc", bound=Callable)
158running_under_pytest = False
159pytest_shows_exceptiongroups = True
160global_force_seed = None
161# this variable stores "engine-global" constants, which are global relative to a
162# ConjectureRunner instance (roughly speaking). Since only one conjecture runner
163# instance can be active per thread, making engine constants thread-local prevents
164# the ConjectureRunner instances of concurrent threads from treading on each other.
165threadlocal = ThreadLocal(_hypothesis_global_random=lambda: None)
168@dataclass(slots=True, frozen=False)
169class Example:
170 args: Any
171 kwargs: Any
172 # Plus two optional arguments for .xfail()
173 raises: Any = field(default=None)
174 reason: Any = field(default=None)
177@dataclass(slots=True, frozen=True)
178class ReportableError:
179 fragments: list[str]
180 exception: BaseException
183# TODO_DOCS link to not-yet-existent patch-dumping docs
186class example:
187 """
188 Add an explicit input to a Hypothesis test, which Hypothesis will always
189 try before generating random inputs. This combines the randomized nature of
190 Hypothesis generation with a traditional parametrized test.
192 For example:
194 .. code-block:: python
196 @example("Hello world")
197 @example("some string with special significance")
198 @given(st.text())
199 def test_strings(s):
200 pass
202 will call ``test_strings("Hello World")`` and
203 ``test_strings("some string with special significance")`` before generating
204 any random inputs. |@example| may be placed in any order relative to |@given|
205 and |@settings|.
207 |Explicit examples| from |@example| are run in the
208 |Phase.explicit| phase. Explicit examples do not count towards
209 |settings.max_examples|. Note that explicit examples added by |@example| do
210 not shrink. If an explicit example fails, Hypothesis will stop and report
211 the failure without generating any random inputs.
213 |@example| can also be used to easily reproduce a failure. For instance, if
214 Hypothesis reports that ``f(n=[0, math.nan])`` fails, you can add
215 ``@example(n=[0, math.nan])`` to your test to quickly reproduce that failure.
217 Arguments to ``@example``
218 -------------------------
220 Arguments to |@example| have the same behavior and restrictions as arguments
221 to |@given|. This means they may be either positional or keyword arguments
222 (but not both in the same |@example|):
224 .. code-block:: python
226 @example(1, 2)
227 @example(x=1, y=2)
228 @given(st.integers(), st.integers())
229 def test(x, y):
230 pass
232 Noting that while arguments to |@given| are strategies (like |st.integers|),
233 arguments to |@example| are values instead (like ``1``).
235 See the :ref:`given-arguments` section for full details.
236 """
238 def __init__(self, *args: Any, **kwargs: Any) -> None:
239 if args and kwargs:
240 raise InvalidArgument(
241 "Cannot mix positional and keyword arguments for examples"
242 )
243 if not (args or kwargs):
244 raise InvalidArgument("An example must provide at least one argument")
246 self.hypothesis_explicit_examples: list[Example] = []
247 self._this_example = Example(tuple(args), kwargs)
249 def __call__(self, test: TestFunc) -> TestFunc:
250 if not hasattr(test, "hypothesis_explicit_examples"):
251 test.hypothesis_explicit_examples = self.hypothesis_explicit_examples # type: ignore
252 test.hypothesis_explicit_examples.append(self._this_example) # type: ignore
253 return test
255 def xfail(
256 self,
257 condition: bool = True, # noqa: FBT002
258 *,
259 reason: str = "",
260 raises: type[BaseException] | tuple[type[BaseException], ...] = BaseException,
261 ) -> "example":
262 """Mark this example as an expected failure, similarly to
263 :obj:`pytest.mark.xfail(strict=True) <pytest.mark.xfail>`.
265 Expected-failing examples allow you to check that your test does fail on
266 some examples, and therefore build confidence that *passing* tests are
267 because your code is working, not because the test is missing something.
269 .. code-block:: python
271 @example(...).xfail()
272 @example(...).xfail(reason="Prices must be non-negative")
273 @example(...).xfail(raises=(KeyError, ValueError))
274 @example(...).xfail(sys.version_info[:2] >= (3, 12), reason="needs py 3.12")
275 @example(...).xfail(condition=sys.platform != "linux", raises=OSError)
276 def test(x):
277 pass
279 .. note::
281 Expected-failing examples are handled separately from those generated
282 by strategies, so you should usually ensure that there is no overlap.
284 .. code-block:: python
286 @example(x=1, y=0).xfail(raises=ZeroDivisionError)
287 @given(x=st.just(1), y=st.integers()) # Missing `.filter(bool)`!
288 def test_fraction(x, y):
289 # This test will try the explicit example and see it fail as
290 # expected, then go on to generate more examples from the
291 # strategy. If we happen to generate y=0, the test will fail
292 # because only the explicit example is treated as xfailing.
293 x / y
294 """
295 check_type(bool, condition, "condition")
296 check_type(str, reason, "reason")
297 if not (
298 isinstance(raises, type) and issubclass(raises, BaseException)
299 ) and not (
300 isinstance(raises, tuple)
301 and raises # () -> expected to fail with no error, which is impossible
302 and all(
303 isinstance(r, type) and issubclass(r, BaseException) for r in raises
304 )
305 ):
306 raise InvalidArgument(
307 f"{raises=} must be an exception type or tuple of exception types"
308 )
309 if condition:
310 self._this_example = dataclasses.replace(
311 self._this_example, raises=raises, reason=reason
312 )
313 return self
315 def via(self, whence: str, /) -> "example":
316 """Attach a machine-readable label noting what the origin of this example
317 was. |example.via| is completely optional and does not change runtime
318 behavior.
320 |example.via| is intended to support self-documenting behavior, as well as
321 tooling which might add (or remove) |@example| decorators automatically.
322 For example:
324 .. code-block:: python
326 # Annotating examples is optional and does not change runtime behavior
327 @example(...)
328 @example(...).via("regression test for issue #42")
329 @example(...).via("discovered failure")
330 def test(x):
331 pass
333 .. note::
335 `HypoFuzz <https://hypofuzz.com/>`_ uses |example.via| to tag examples
336 in the patch of its high-coverage set of explicit inputs, on
337 `the patches page <https://hypofuzz.com/example-dashboard/#/patches>`_.
338 """
339 if not isinstance(whence, str):
340 raise InvalidArgument(".via() must be passed a string")
341 # This is deliberately a no-op at runtime; the tools operate on source code.
342 return self
345def seed(seed: Hashable) -> Callable[[TestFunc], TestFunc]:
346 """
347 Seed the randomness for this test.
349 ``seed`` may be any hashable object. No exact meaning for ``seed`` is provided
350 other than that for a fixed seed value Hypothesis will produce the same
351 |test cases| (assuming that there are no other sources of nondeterminisim, such
352 as timing, hash randomization, or external state).
354 For example, the following test function and |RuleBasedStateMachine| will
355 each generate the same series of test cases each time they are executed:
357 .. code-block:: python
359 @seed(1234)
360 @given(st.integers())
361 def test(n): ...
363 @seed(6789)
364 class MyMachine(RuleBasedStateMachine): ...
366 If using pytest, you can alternatively pass ``--hypothesis-seed`` on the
367 command line.
369 Setting a seed overrides |settings.derandomize|, which is designed to enable
370 deterministic CI tests rather than reproducing observed failures.
372 Hypothesis will only print the seed which would reproduce a failure if a test
373 fails in an unexpected way, for instance inside Hypothesis internals.
374 """
376 def accept(test):
377 test._hypothesis_internal_use_seed = seed
378 current_settings = getattr(test, "_hypothesis_internal_use_settings", None)
379 test._hypothesis_internal_use_settings = Settings(
380 current_settings, database=None
381 )
382 return test
384 return accept
387# TODO_DOCS: link to /explanation/choice-sequence
390def reproduce_failure(version: str, blob: bytes) -> Callable[[TestFunc], TestFunc]:
391 """
392 Run the |test case| corresponding to the binary ``blob`` in order to reproduce a
393 failure. ``blob`` is a serialized version of the internal input representation
394 of Hypothesis.
396 A test decorated with |@reproduce_failure| always runs exactly one test case,
397 which is expected to cause a failure. If the provided ``blob`` does not
398 cause a failure, Hypothesis will raise |DidNotReproduce|.
400 Hypothesis will print an |@reproduce_failure| decorator if
401 |settings.print_blob| is ``True`` (which is the default in CI).
403 |@reproduce_failure| is intended to be temporarily added to your test suite in
404 order to reproduce a failure. It is not intended to be a permanent addition to
405 your test suite. Because of this, no compatibility guarantees are made across
406 Hypothesis versions, and |@reproduce_failure| will error if used on a different
407 Hypothesis version than it was created for.
409 .. seealso::
411 See also the :doc:`/tutorial/replaying-failures` tutorial.
412 """
414 def accept(test):
415 test._hypothesis_internal_use_reproduce_failure = (version, blob)
416 return test
418 return accept
421def reproduction_decorator(choices: Iterable[ChoiceT]) -> str:
422 return f"@reproduce_failure({__version__!r}, {encode_failure(choices)!r})"
425def encode_failure(choices: Iterable[ChoiceT]) -> bytes:
426 blob = choices_to_bytes(choices)
427 compressed = zlib.compress(blob)
428 if len(compressed) < len(blob):
429 blob = b"\1" + compressed
430 else:
431 blob = b"\0" + blob
432 return base64.b64encode(blob)
435def decode_failure(blob: bytes) -> Sequence[ChoiceT]:
436 try:
437 decoded = base64.b64decode(blob)
438 except Exception:
439 raise InvalidArgument(f"Invalid base64 encoded string: {blob!r}") from None
441 prefix = decoded[:1]
442 if prefix == b"\0":
443 decoded = decoded[1:]
444 elif prefix == b"\1":
445 try:
446 decoded = zlib.decompress(decoded[1:])
447 except zlib.error as err:
448 raise InvalidArgument(
449 f"Invalid zlib compression for blob {blob!r}"
450 ) from err
451 else:
452 raise InvalidArgument(
453 f"Could not decode blob {blob!r}: Invalid start byte {prefix!r}"
454 )
456 choices = choices_from_bytes(decoded)
457 if choices is None:
458 raise InvalidArgument(f"Invalid serialized choice sequence for blob {blob!r}")
460 return choices
463def _invalid(message, *, exc=InvalidArgument, test, given_kwargs):
464 @impersonate(test)
465 def wrapped_test(*arguments, **kwargs): # pragma: no cover # coverage limitation
466 raise exc(message)
468 wrapped_test.is_hypothesis_test = True
469 wrapped_test.hypothesis = HypothesisHandle(
470 inner_test=test,
471 _get_fuzz_target=wrapped_test,
472 _given_kwargs=given_kwargs,
473 )
474 return wrapped_test
477def is_invalid_test(test, original_sig, given_arguments, given_kwargs):
478 """Check the arguments to ``@given`` for basic usage constraints.
480 Most errors are not raised immediately; instead we return a dummy test
481 function that will raise the appropriate error if it is actually called.
482 When the user runs a subset of tests (e.g via ``pytest -k``), errors will
483 only be reported for tests that actually ran.
484 """
485 invalid = partial(_invalid, test=test, given_kwargs=given_kwargs)
487 if not (given_arguments or given_kwargs):
488 return invalid("given must be called with at least one argument")
490 params = list(original_sig.parameters.values())
491 pos_params = [p for p in params if p.kind is p.POSITIONAL_OR_KEYWORD]
492 kwonly_params = [p for p in params if p.kind is p.KEYWORD_ONLY]
493 if given_arguments and params != pos_params:
494 return invalid(
495 "positional arguments to @given are not supported with varargs, "
496 "varkeywords, positional-only, or keyword-only arguments"
497 )
499 if len(given_arguments) > len(pos_params):
500 return invalid(
501 f"Too many positional arguments for {test.__name__}() were passed to "
502 f"@given - expected at most {len(pos_params)} "
503 f"arguments, but got {len(given_arguments)} {given_arguments!r}"
504 )
506 if ... in given_arguments:
507 return invalid(
508 "... was passed as a positional argument to @given, but may only be "
509 "passed as a keyword argument or as the sole argument of @given"
510 )
512 if given_arguments and given_kwargs:
513 return invalid("cannot mix positional and keyword arguments to @given")
514 extra_kwargs = [
515 k for k in given_kwargs if k not in {p.name for p in pos_params + kwonly_params}
516 ]
517 if extra_kwargs and (params == [] or params[-1].kind is not params[-1].VAR_KEYWORD):
518 arg = extra_kwargs[0]
519 extra = ""
520 if arg in all_settings:
521 extra = f". Did you mean @settings({arg}={given_kwargs[arg]!r})?"
522 return invalid(
523 f"{test.__name__}() got an unexpected keyword argument {arg!r}, "
524 f"from `{arg}={given_kwargs[arg]!r}` in @given{extra}"
525 )
526 if any(p.default is not p.empty for p in params):
527 return invalid("Cannot apply @given to a function with defaults.")
529 # This case would raise Unsatisfiable *anyway*, but by detecting it here we can
530 # provide a much more helpful error message for people e.g. using the Ghostwriter.
531 empty = [
532 f"{s!r} (arg {idx})" for idx, s in enumerate(given_arguments) if s is NOTHING
533 ] + [f"{name}={s!r}" for name, s in given_kwargs.items() if s is NOTHING]
534 if empty:
535 strats = "strategies" if len(empty) > 1 else "strategy"
536 return invalid(
537 f"Cannot generate test cases from empty {strats}: " + ", ".join(empty),
538 exc=Unsatisfiable,
539 )
542def execute_explicit_examples(state, wrapped_test, arguments, kwargs, original_sig):
543 assert isinstance(state, StateForActualGivenExecution)
544 posargs = [
545 p.name
546 for p in original_sig.parameters.values()
547 if p.kind is p.POSITIONAL_OR_KEYWORD
548 ]
550 for example in reversed(getattr(wrapped_test, "hypothesis_explicit_examples", ())):
551 assert isinstance(example, Example)
552 # All of this validation is to check that @example() got "the same" arguments
553 # as @given, i.e. corresponding to the same parameters, even though they might
554 # be any mixture of positional and keyword arguments.
555 if example.args:
556 assert not example.kwargs
557 if any(
558 p.kind is p.POSITIONAL_ONLY for p in original_sig.parameters.values()
559 ):
560 raise InvalidArgument(
561 "Cannot pass positional arguments to @example() when decorating "
562 "a test function which has positional-only parameters."
563 )
564 if len(example.args) > len(posargs):
565 raise InvalidArgument(
566 "example has too many arguments for test. Expected at most "
567 f"{len(posargs)} but got {len(example.args)}"
568 )
569 example_kwargs = dict(
570 zip(posargs[-len(example.args) :], example.args, strict=True)
571 )
572 else:
573 example_kwargs = dict(example.kwargs)
574 given_kws = ", ".join(
575 repr(k) for k in sorted(wrapped_test.hypothesis._given_kwargs)
576 )
577 example_kws = ", ".join(repr(k) for k in sorted(example_kwargs))
578 if given_kws != example_kws:
579 raise InvalidArgument(
580 f"Inconsistent args: @given() got strategies for {given_kws}, "
581 f"but @example() got arguments for {example_kws}"
582 ) from None
584 # This is certainly true because the example_kwargs exactly match the params
585 # reserved by @given(), which are then remove from the function signature.
586 assert set(example_kwargs).isdisjoint(kwargs)
587 example_kwargs.update(kwargs)
589 if Phase.explicit not in state.settings.phases:
590 continue
592 with local_settings(state.settings):
593 fragments_reported = []
594 empty_data = ConjectureData.for_choices([])
595 try:
596 execute_example = partial(
597 state.execute_once,
598 empty_data,
599 is_final=True,
600 print_test_case=True,
601 example_kwargs=example_kwargs,
602 )
603 with with_reporter(fragments_reported.append):
604 if example.raises is None:
605 execute_example()
606 else:
607 # @example(...).xfail(...)
608 bits = ", ".join(nicerepr(x) for x in arguments) + ", ".join(
609 f"{k}={nicerepr(v)}" for k, v in example_kwargs.items()
610 )
611 try:
612 execute_example()
613 except failure_exceptions_to_catch() as err:
614 if not isinstance(err, example.raises):
615 raise
616 # Save a string form of this example; we'll warn if it's
617 # ever generated by the strategy (which can't be xfailed)
618 state.xfail_example_reprs.add(
619 repr_call(state.test, arguments, example_kwargs)
620 )
621 except example.raises as err:
622 # We'd usually check this as early as possible, but it's
623 # possible for failure_exceptions_to_catch() to grow when
624 # e.g. pytest is imported between import- and test-time.
625 raise InvalidArgument(
626 f"@example({bits}) raised an expected {err!r}, "
627 "but Hypothesis does not treat this as a test failure"
628 ) from err
629 else:
630 # Unexpectedly passing; always raise an error in this case.
631 reason = f" because {example.reason}" * bool(example.reason)
632 if example.raises is BaseException:
633 name = "exception" # special-case no raises= arg
634 elif not isinstance(example.raises, tuple):
635 name = example.raises.__name__
636 elif len(example.raises) == 1:
637 name = example.raises[0].__name__
638 else:
639 name = (
640 ", ".join(ex.__name__ for ex in example.raises[:-1])
641 + f", or {example.raises[-1].__name__}"
642 )
643 vowel = name.upper()[0] in "AEIOU"
644 raise AssertionError(
645 f"Expected a{'n' * vowel} {name} from @example({bits})"
646 f"{reason}, but no exception was raised."
647 )
648 except UnsatisfiedAssumption:
649 # Odd though it seems, we deliberately support explicit examples that
650 # are then rejected by a call to `assume()`. As well as iterative
651 # development, this is rather useful to replay Hypothesis' part of
652 # a saved failure when other arguments are supplied by e.g. pytest.
653 # See https://github.com/HypothesisWorks/hypothesis/issues/2125
654 with contextlib.suppress(StopTest):
655 empty_data.conclude_test(Status.INVALID)
656 except BaseException as err:
657 # In order to support reporting of multiple failing test cases, we yield
658 # each of the (report text, error) pairs we find back to the top-level
659 # runner. This also ensures that user-facing stack traces have as few
660 # frames of Hypothesis internals as possible.
661 err = err.with_traceback(get_trimmed_traceback())
663 # One user error - whether misunderstanding or typo - we've seen a few
664 # times is to pass strategies to @example() where values are expected.
665 # Checking is easy, and false-positives not much of a problem, so:
666 if isinstance(err, failure_exceptions_to_catch()) and any(
667 isinstance(arg, SearchStrategy)
668 for arg in example.args + tuple(example.kwargs.values())
669 ):
670 new = HypothesisWarning(
671 "The @example() decorator expects to be passed values, but "
672 "you passed strategies instead. See https://hypothesis."
673 "readthedocs.io/en/latest/reference/api.html#hypothesis"
674 ".example for details."
675 )
676 new.__cause__ = err
677 err = new
679 with contextlib.suppress(StopTest):
680 empty_data.conclude_test(Status.INVALID)
681 yield ReportableError(fragments_reported, err)
682 if (
683 state.settings.report_multiple_bugs
684 and pytest_shows_exceptiongroups
685 and isinstance(err, failure_exceptions_to_catch())
686 and not isinstance(err, skip_exceptions_to_reraise())
687 ):
688 continue
689 break
690 finally:
691 if fragments_reported:
692 assert fragments_reported[0].startswith("Failing test case")
693 fragments_reported[0] = fragments_reported[0].replace(
694 "Failing test case", "Failing explicit example", 1
695 )
697 empty_data.freeze()
698 if observability_enabled():
699 tc = make_testcase(
700 run_start=state._start_timestamp,
701 property=state.test_identifier,
702 data=empty_data,
703 how_generated="explicit example",
704 representation=state._string_repr,
705 timing=state._timing_features,
706 )
707 deliver_observation(tc)
709 if fragments_reported:
710 verbose_report(fragments_reported[0].replace("Failing", "Trying", 1))
711 for f in fragments_reported[1:]:
712 verbose_report(f)
715def get_random_for_wrapped_test(test, wrapped_test):
716 settings = wrapped_test._hypothesis_internal_use_settings
717 wrapped_test._hypothesis_internal_use_generated_seed = None
719 if wrapped_test._hypothesis_internal_use_seed is not None:
720 return Random(wrapped_test._hypothesis_internal_use_seed)
722 if settings.derandomize:
723 return Random(int_from_bytes(function_digest(test)))
725 if global_force_seed is not None:
726 return Random(global_force_seed)
728 if threadlocal._hypothesis_global_random is None:
729 threadlocal._hypothesis_global_random = Random()
730 seed = threadlocal._hypothesis_global_random.getrandbits(128)
731 wrapped_test._hypothesis_internal_use_generated_seed = seed
732 return Random(seed)
735@dataclass(slots=True, frozen=False)
736class Stuff:
737 selfy: Any
738 args: tuple
739 kwargs: dict
740 given_kwargs: dict
743def process_arguments_to_given(
744 wrapped_test: Any,
745 arguments: Sequence[object],
746 kwargs: dict[str, object],
747 given_kwargs: dict[str, SearchStrategy],
748 params: dict[str, Parameter],
749) -> tuple[Sequence[object], dict[str, object], Stuff]:
750 selfy = None
751 arguments, kwargs = convert_positional_arguments(wrapped_test, arguments, kwargs)
753 # If the test function is a method of some kind, the bound object
754 # will be the first named argument if there are any, otherwise the
755 # first vararg (if any).
756 posargs = [p.name for p in params.values() if p.kind is p.POSITIONAL_OR_KEYWORD]
757 if posargs:
758 selfy = kwargs.get(posargs[0])
759 elif arguments:
760 selfy = arguments[0]
762 # Ensure that we don't mistake mocks for self here.
763 # This can cause the mock to be used as the test runner.
764 if is_mock(selfy):
765 selfy = None
767 arguments = tuple(arguments)
769 with ensure_free_stackframes():
770 for k, s in given_kwargs.items():
771 check_strategy(s, name=k)
772 s.validate()
774 stuff = Stuff(selfy=selfy, args=arguments, kwargs=kwargs, given_kwargs=given_kwargs)
776 return arguments, kwargs, stuff
779def skip_exceptions_to_reraise():
780 """Return a tuple of exceptions meaning 'skip this test', to re-raise.
782 This is intended to cover most common test runners; if you would
783 like another to be added please open an issue or pull request adding
784 it to this function and to tests/cover/test_lazy_import.py
785 """
786 # This is a set in case any library simply re-exports another's Skip exception
787 exceptions = set()
788 # We use this sys.modules trick to avoid importing libraries -
789 # you can't be an instance of a type from an unimported module!
790 # This is fast enough that we don't need to cache the result,
791 # and more importantly it avoids possible side-effects :-)
792 if "unittest" in sys.modules:
793 exceptions.add(sys.modules["unittest"].SkipTest)
794 if "_pytest.outcomes" in sys.modules:
795 exceptions.add(sys.modules["_pytest.outcomes"].Skipped)
796 return tuple(sorted(exceptions, key=str))
799def failure_exceptions_to_catch() -> tuple[type[BaseException], ...]:
800 """Return a tuple of exceptions meaning 'this test has failed', to catch.
802 This is intended to cover most common test runners; if you would
803 like another to be added please open an issue or pull request.
804 """
805 # While SystemExit and GeneratorExit are instances of BaseException, we also
806 # expect them to be deterministic - unlike KeyboardInterrupt - and so we treat
807 # them as standard exceptions, check for flakiness, etc.
808 # See https://github.com/HypothesisWorks/hypothesis/issues/2223 for details.
809 exceptions = [Exception, SystemExit, GeneratorExit]
810 if "_pytest.outcomes" in sys.modules:
811 exceptions.append(sys.modules["_pytest.outcomes"].Failed)
812 return tuple(exceptions)
815def new_given_signature(original_sig, given_kwargs):
816 """Make an updated signature for the wrapped test."""
817 return original_sig.replace(
818 parameters=[
819 p
820 for p in original_sig.parameters.values()
821 if not (
822 p.name in given_kwargs
823 and p.kind in (p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY)
824 )
825 ],
826 return_annotation=None,
827 )
830def default_executor(data, function):
831 return function(data)
834def get_executor(runner):
835 try:
836 execute_example = runner.execute_example
837 except AttributeError:
838 pass
839 else:
840 return lambda data, function: execute_example(partial(function, data))
842 if hasattr(runner, "setup_example") or hasattr(runner, "teardown_example"):
843 setup = getattr(runner, "setup_example", None) or (lambda: None)
844 teardown = getattr(runner, "teardown_example", None) or (lambda ex: None)
846 def execute(data, function):
847 token = None
848 try:
849 token = setup()
850 return function(data)
851 finally:
852 teardown(token)
854 return execute
856 return default_executor
859# This function is a crude solution, a better way of resolving it would probably
860# be to rewrite a bunch of exception handlers to use except*.
861T = TypeVar("T", bound=BaseException)
864def _flatten_group(excgroup: BaseExceptionGroup[T]) -> list[T]:
865 found_exceptions: list[T] = []
866 for exc in excgroup.exceptions:
867 if isinstance(exc, BaseExceptionGroup):
868 found_exceptions.extend(_flatten_group(exc))
869 else:
870 found_exceptions.append(exc)
871 return found_exceptions
874@contextlib.contextmanager
875def unwrap_markers_from_group() -> Generator[None, None, None]:
876 try:
877 yield
878 except BaseExceptionGroup as excgroup:
879 _frozen_exceptions, non_frozen_exceptions = excgroup.split(Frozen)
881 # group only contains Frozen, reraise the group
882 # it doesn't matter what we raise, since any exceptions get disregarded
883 # and reraised as StopTest if data got frozen.
884 if non_frozen_exceptions is None:
885 raise
886 # in all other cases they are discarded
888 # Can RewindRecursive end up in this group?
889 _, user_exceptions = non_frozen_exceptions.split(
890 lambda e: isinstance(e, (StopTest, HypothesisException))
891 )
893 # this might contain marker exceptions, or internal errors, but not frozen.
894 if user_exceptions is not None:
895 raise
897 # single marker exception - reraise it
898 flattened_non_frozen_exceptions: list[BaseException] = _flatten_group(
899 non_frozen_exceptions
900 )
901 if len(flattened_non_frozen_exceptions) == 1:
902 e = flattened_non_frozen_exceptions[0]
903 # preserve the cause of the original exception to not hinder debugging
904 # note that __context__ is still lost though
905 raise e from e.__cause__
907 # multiple marker exceptions. If we re-raise the whole group we break
908 # a bunch of logic so ....?
909 stoptests, non_stoptests = non_frozen_exceptions.split(StopTest)
911 # TODO: stoptest+hypothesisexception ...? Is it possible? If so, what do?
913 if non_stoptests:
914 # TODO: multiple marker exceptions is easy to produce, but the logic in the
915 # engine does not handle it... so we just reraise the first one for now.
916 e = _flatten_group(non_stoptests)[0]
917 raise e from e.__cause__
918 assert stoptests is not None
920 # multiple stoptests: raising the one with the lowest testcounter
921 raise min(_flatten_group(stoptests), key=lambda s_e: s_e.testcounter)
924class StateForActualGivenExecution:
925 def __init__(
926 self,
927 stuff: Stuff,
928 test: Callable[..., Any],
929 settings: Settings,
930 random: Random,
931 wrapped_test: Any,
932 *,
933 thread_overlap: dict[int, bool] | None = None,
934 ):
935 self.stuff = stuff
936 self.test = test
937 self.settings = settings
938 self.random = random
939 self.wrapped_test = wrapped_test
940 self.thread_overlap = {} if thread_overlap is None else thread_overlap
942 self.test_runner = get_executor(stuff.selfy)
943 self.print_given_args = getattr(
944 wrapped_test, "_hypothesis_internal_print_given_args", True
945 )
946 self.known_safe_to_repr = {
947 name
948 for name, strategy in stuff.given_kwargs.items()
949 if strategy == st.data()
950 }
952 self.last_exception = None
953 self.falsifying_test_cases = ()
954 self.ever_executed = False
955 self.xfail_example_reprs: set[str] = set()
956 self.failed_normally = False
957 self.failed_due_to_deadline = False
959 self.explain_traces: dict[InterestingOrigin | None, set[Trace]] = defaultdict(
960 set
961 )
962 self._start_timestamp = time.time()
963 self._string_repr = ""
964 self._proxied_tests: dict[bool, Callable] = {}
965 self._timing_features: dict[str, float] = {}
967 self._runner: ConjectureRunner | None = None
969 @property
970 def test_identifier(self) -> str:
971 return getattr(
972 current_pytest_item.value, "nodeid", None
973 ) or get_pretty_function_description(self.wrapped_test)
975 def _should_trace(self):
976 # NOTE: we explicitly support monkeypatching this. Keep the namespace
977 # access intact.
978 _trace_obs = (
979 observability_enabled() and observability.OBSERVABILITY_COLLECT_COVERAGE
980 )
981 _trace_failure = (
982 self.failed_normally
983 and not self.failed_due_to_deadline
984 and {Phase.shrink, Phase.explain}.issubset(self.settings.phases)
985 )
986 return _trace_obs or _trace_failure
988 def _proxied_test(self, *, timed: bool) -> Callable:
989 # Building a signature-preserving proxy is expensive, so we do it once
990 # per test rather than once per example. Per-example state is read
991 # from the current build context.
992 try:
993 return self._proxied_tests[timed]
994 except KeyError:
995 pass
996 if not timed:
998 @proxies(self.test)
999 def test(*args, **kwargs):
1000 with unwrap_markers_from_group(), ensure_free_stackframes():
1001 return self.test(*args, **kwargs)
1003 else:
1005 @proxies(self.test)
1006 def test(*args, **kwargs):
1007 context = current_build_context()
1008 data, is_final = context.data, context.is_final
1009 arg_drawtime = math.fsum(data.draw_times.values())
1010 arg_stateful = math.fsum(data._stateful_run_times.values())
1011 arg_gctime = gc_cumulative_time()
1012 with unwrap_markers_from_group(), ensure_free_stackframes():
1013 start = time.perf_counter()
1014 try:
1015 result = self.test(*args, **kwargs)
1016 finally:
1017 finish = time.perf_counter()
1018 in_drawtime = math.fsum(data.draw_times.values()) - arg_drawtime
1019 in_stateful = (
1020 math.fsum(data._stateful_run_times.values()) - arg_stateful
1021 )
1022 in_gctime = gc_cumulative_time() - arg_gctime
1023 runtime = finish - start - in_drawtime - in_stateful - in_gctime
1024 self._timing_features = {
1025 "execute:test": runtime,
1026 "overall:gc": in_gctime,
1027 **data.draw_times,
1028 **data._stateful_run_times,
1029 }
1031 if (
1032 (current_deadline := self.settings.deadline) is not None
1033 # we disable the deadline check under concurrent threads, since
1034 # cpython may switch away from a thread for arbitrarily long.
1035 and not self.thread_overlap.get(threading.get_ident(), False)
1036 ):
1037 if not is_final:
1038 current_deadline = (current_deadline // 4) * 5
1039 if runtime >= current_deadline.total_seconds():
1040 raise DeadlineExceeded(
1041 datetime.timedelta(seconds=runtime), self.settings.deadline
1042 )
1043 return result
1045 self._proxied_tests[timed] = test
1046 return test
1048 def execute_once(
1049 self,
1050 data,
1051 *,
1052 print_test_case=False,
1053 is_final=False,
1054 expected_failure=None,
1055 example_kwargs=None,
1056 ):
1057 """Run the test function once, using ``data`` as input.
1059 If the test raises an exception, it will propagate through to the
1060 caller of this method. Depending on its type, this could represent
1061 an ordinary test failure, or a fatal error, or a control exception.
1063 If this method returns normally, the test might have passed, or
1064 it might have placed ``data`` in an unsuccessful state and then
1065 swallowed the corresponding control exception.
1066 """
1068 self.ever_executed = True
1070 self._string_repr = ""
1071 text_repr = None
1072 test = self._proxied_test(
1073 timed=self.settings.deadline is not None or observability_enabled()
1074 )
1076 def run(data: ConjectureData) -> None:
1077 # Set up dynamic context needed by a single test run.
1078 if self.stuff.selfy is not None:
1079 data.hypothesis_runner = self.stuff.selfy
1080 # Generate all arguments to the test function.
1081 args = self.stuff.args
1082 kwargs = dict(self.stuff.kwargs)
1083 if example_kwargs is None:
1084 kw, arglabels = context.prep_args_kwargs_from_strategies(
1085 self.stuff.given_kwargs
1086 )
1087 else:
1088 kw = example_kwargs
1089 arglabels = {}
1090 kwargs.update(kw)
1091 if expected_failure is not None:
1092 nonlocal text_repr
1093 text_repr = repr_call(test, args, kwargs)
1095 if print_test_case or current_verbosity() >= Verbosity.verbose:
1096 printer = RepresentationPrinter(context=context)
1097 if print_test_case:
1098 printer.text("Failing test case:")
1099 else:
1100 printer.text("Test case:")
1102 if self.print_given_args:
1103 printer.text(" ")
1104 printer.repr_call(
1105 test.__name__,
1106 args,
1107 kwargs,
1108 force_split=True,
1109 arg_labels=arglabels,
1110 leading_comment=(
1111 "# " + context.data.span_comments[None]
1112 if None in context.data.span_comments
1113 else None
1114 ),
1115 avoid_realization=data.provider.avoid_realization,
1116 known_safe_to_repr=self.known_safe_to_repr,
1117 )
1118 else:
1119 # print_given_args is only disabled for stateful tests,
1120 # whose steps are printed separately (and only on failure
1121 # or at Verbosity.debug) - so say why this line is bare.
1122 printer.text(" <state machine>")
1123 report(printer.getvalue())
1125 if observability_enabled():
1126 printer = RepresentationPrinter(context=context)
1127 printer.repr_call(
1128 test.__name__,
1129 args,
1130 kwargs,
1131 force_split=True,
1132 arg_labels=arglabels,
1133 leading_comment=(
1134 "# " + context.data.span_comments[None]
1135 if None in context.data.span_comments
1136 else None
1137 ),
1138 avoid_realization=data.provider.avoid_realization,
1139 known_safe_to_repr=self.known_safe_to_repr,
1140 )
1141 self._string_repr = printer.getvalue()
1143 try:
1144 return test(*args, **kwargs)
1145 except TypeError as e:
1146 # If we sampled from a sequence of strategies, AND failed with a
1147 # TypeError, *AND that exception mentions SearchStrategy*, add a note:
1148 if (
1149 "SearchStrategy" in str(e)
1150 and data._sampled_from_all_strategies_elements_message is not None
1151 ):
1152 msg, format_arg = data._sampled_from_all_strategies_elements_message
1153 add_note(e, msg.format(format_arg))
1154 raise
1155 finally:
1156 if data._stateful_repr_parts is not None:
1157 self._string_repr = "\n".join(data._stateful_repr_parts)
1159 if observability_enabled():
1160 printer = RepresentationPrinter(context=context)
1161 for name, value in data._observability_args.items():
1162 if name.startswith("generate:Draw "):
1163 try:
1164 value = data.provider.realize(value)
1165 except BackendCannotProceed: # pragma: no cover
1166 value = "<backend failed to realize symbolic>"
1167 printer.text(f"\n{name.removeprefix('generate:')}: ")
1168 printer.pretty(value)
1170 self._string_repr += printer.getvalue()
1172 # self.test_runner can include the execute_example method, or setup/teardown
1173 # _example, so it's important to get the PRNG and build context in place first.
1174 with (
1175 local_settings(self.settings),
1176 deterministic_PRNG(),
1177 BuildContext(
1178 data, is_final=is_final, wrapped_test=self.wrapped_test
1179 ) as context,
1180 ):
1181 # providers may throw in per_case_context_fn, and we'd like
1182 # `result` to still be set in these cases.
1183 result = None
1184 with data.provider.per_test_case_context_manager():
1185 # Run the test function once, via the executor hook.
1186 # In most cases this will delegate straight to `run(data)`.
1187 result = self.test_runner(data, run)
1189 # If a failure was expected, it should have been raised already, so
1190 # instead raise an appropriate diagnostic error.
1191 if expected_failure is not None:
1192 exception, traceback = expected_failure
1193 if isinstance(exception, DeadlineExceeded) and (
1194 runtime_secs := math.fsum(
1195 v
1196 for k, v in self._timing_features.items()
1197 if k.startswith("execute:")
1198 )
1199 ):
1200 report(
1201 "Unreliable test timings! On an initial run, this "
1202 f"test took {exception.runtime.total_seconds() * 1000:.2f}ms, "
1203 "which exceeded the deadline of "
1204 f"{self.settings.deadline.total_seconds() * 1000:.2f}ms, but "
1205 f"on a subsequent run it took {runtime_secs * 1000:.2f} ms, "
1206 "which did not. If you expect this sort of "
1207 "variability in your test timings, consider turning "
1208 "deadlines off for this test by setting deadline=None."
1209 )
1210 else:
1211 report("Failed to reproduce exception. Expected: \n" + traceback)
1212 raise FlakyFailure(
1213 f"Hypothesis {text_repr} produces unreliable results: "
1214 "Failed on the first call but did not on a subsequent one",
1215 [exception],
1216 )
1217 return result
1219 def _flaky_replay_to_failure(
1220 self, err: FlakyReplay, context: BaseException
1221 ) -> FlakyFailure:
1222 assert self._runner is not None
1223 # Note that in the mark_interesting case, _context_
1224 # is part of err._interesting_origins - but it's not in
1225 # _runner.interesting_test_cases - this is fine, as the context
1226 # (i.e., immediate exception) is appended.
1227 interesting_test_cases = [
1228 self._runner.interesting_test_cases[origin]
1229 for origin in err._interesting_origins
1230 if origin in self._runner.interesting_test_cases
1231 ]
1232 exceptions = [result.expected_exception for result in interesting_test_cases]
1233 exceptions.append(context) # the immediate exception
1234 return FlakyFailure(err.reason, exceptions)
1236 def _execute_once_for_engine(self, data: ConjectureData) -> None:
1237 """Wrapper around ``execute_once`` that intercepts test failure
1238 exceptions and single-test control exceptions, and turns them into
1239 appropriate method calls to `data` instead.
1241 This allows the engine to assume that any exception other than
1242 ``StopTest`` must be a fatal error, and should stop the entire engine.
1243 """
1244 trace: Trace = frozenset()
1245 backend_cannot_proceed = False
1246 try:
1247 with Tracer(should_trace=self._should_trace()) as tracer:
1248 try:
1249 result = self.execute_once(data)
1250 if data.status == Status.VALID and tracer.branches:
1251 self.explain_traces[None].add(tracer.branches)
1252 finally:
1253 trace = tracer.branches
1254 if result is not None:
1255 fail_health_check(
1256 self.settings,
1257 "Tests run under @given should return None, but "
1258 f"{self.test.__name__} returned {result!r} instead.",
1259 HealthCheck.return_value,
1260 )
1261 except UnsatisfiedAssumption as e:
1262 # An "assume" check failed, so instead we inform the engine that
1263 # this test run was invalid.
1264 try:
1265 data.mark_invalid(e.reason, location=e.location)
1266 except FlakyReplay as err:
1267 # This was unexpected, meaning that the assume was flaky.
1268 # Report it as such.
1269 raise self._flaky_replay_to_failure(err, e) from None
1270 except BackendCannotProceed:
1271 # The engine discards this iteration entirely (see engine.py,
1272 # "we're pretending this never happened"), so we shouldn't emit a
1273 # test_case observation for it either -- otherwise an alternative
1274 # backend that aborts before running the test (e.g. crosshair when
1275 # it has exhausted its paths) surfaces a spurious, draw-less
1276 # "passed" observation with an empty representation.
1277 backend_cannot_proceed = True
1278 raise
1279 except StopTest:
1280 # The engine knows how to handle this control exception, so it's
1281 # OK to re-raise it.
1282 raise
1283 except (
1284 FailedHealthCheck,
1285 *skip_exceptions_to_reraise(),
1286 ):
1287 # These are fatal errors or control exceptions that should stop the
1288 # engine, so we re-raise them.
1289 raise
1290 except failure_exceptions_to_catch() as e:
1291 # If an unhandled (i.e., non-Hypothesis) error was raised by
1292 # Hypothesis-internal code, re-raise it as a fatal error instead
1293 # of treating it as a test failure.
1294 if isinstance(e, BaseExceptionGroup) and len(e.exceptions) == 1:
1295 # When a naked exception is implicitly wrapped in an ExceptionGroup
1296 # due to a re-raising "except*", the ExceptionGroup is constructed in
1297 # the caller's stack frame (see #4183). This workaround is specifically
1298 # for implicit wrapping of naked exceptions by "except*", since explicit
1299 # raising of ExceptionGroup gets the proper traceback in the first place
1300 # - there's no need to handle hierarchical groups here, at least if no
1301 # such implicit wrapping happens inside hypothesis code (we only care
1302 # about the hypothesis-or-not distinction).
1303 #
1304 # 01-25-2025: this was patched to give the correct
1305 # stacktrace in cpython https://github.com/python/cpython/issues/128799.
1306 # can remove once python3.11 is EOL.
1307 tb = e.exceptions[0].__traceback__ or e.__traceback__
1308 else:
1309 tb = e.__traceback__
1310 assert tb is not None
1311 filepath = _last_frame_location(tb)[0]
1312 if (
1313 is_hypothesis_file(filepath)
1314 and not isinstance(e, HypothesisException)
1315 # We expect backend authors to use the provider_conformance test
1316 # to test their backends. If an error occurs there, it is probably
1317 # from their backend, and we would like to treat it as a standard
1318 # error, not a hypothesis-internal error.
1319 and not filepath.endswith(
1320 f"internal{os.sep}conjecture{os.sep}provider_conformance.py"
1321 )
1322 ):
1323 raise
1325 if data.frozen:
1326 # This can happen if an error occurred in a finally
1327 # block somewhere, suppressing our original StopTest.
1328 # We raise a new one here to resume normal operation.
1329 raise StopTest(data.testcounter) from e
1330 else:
1331 # The test failed by raising an exception, so we inform the
1332 # engine that this test run was interesting. This is the normal
1333 # path for test runs that fail.
1334 data.expected_exception = e
1335 # Formatting tracebacks is expensive, and we usually only need the
1336 # traceback for the final reported failure - so defer it unless
1337 # we're about to print it or record it for observability.
1338 if current_verbosity() >= Verbosity.verbose or observability_enabled():
1339 data.expected_traceback = format_exception(
1340 e, get_trimmed_traceback()
1341 )
1342 verbose_report(data.expected_traceback)
1344 self.failed_normally = True
1346 interesting_origin = InterestingOrigin.from_exception(e)
1347 if trace:
1348 self.explain_traces[interesting_origin].add(trace)
1349 if interesting_origin.exc_type == DeadlineExceeded:
1350 self.failed_due_to_deadline = True
1351 self.explain_traces.clear()
1352 try:
1353 data.mark_interesting(interesting_origin)
1354 except FlakyReplay as err:
1355 raise self._flaky_replay_to_failure(err, e) from None
1357 finally:
1358 # Conditional here so we can save some time constructing the payload; in
1359 # other cases (without coverage) it's cheap enough to do that regardless.
1360 #
1361 # Note that we have to unconditionally realize data.events, because
1362 # the statistics reported by the pytest plugin use a different flow
1363 # than observability, but still access symbolic events.
1365 try:
1366 data.events = data.provider.realize(data.events)
1367 except BackendCannotProceed:
1368 data.events = {}
1370 if observability_enabled() and not backend_cannot_proceed:
1371 if runner := getattr(self, "_runner", None):
1372 phase = runner._current_phase
1373 else: # pragma: no cover # in case of messing with internals
1374 if self.failed_normally or self.failed_due_to_deadline:
1375 phase = "shrink"
1376 else:
1377 phase = "unknown"
1378 backend_desc = f", using backend={self.settings.backend!r}" * (
1379 self.settings.backend != "hypothesis"
1380 and not getattr(runner, "_switch_to_hypothesis_provider", False)
1381 )
1382 try:
1383 data._observability_args = data.provider.realize(
1384 data._observability_args
1385 )
1386 except BackendCannotProceed:
1387 data._observability_args = {}
1389 try:
1390 self._string_repr = data.provider.realize(self._string_repr)
1391 except BackendCannotProceed:
1392 self._string_repr = "<backend failed to realize symbolic arguments>"
1394 try:
1395 data.notes = data.provider.realize(data.notes)
1396 except BackendCannotProceed:
1397 data.notes = []
1399 data.freeze()
1400 tc = make_testcase(
1401 run_start=self._start_timestamp,
1402 property=self.test_identifier,
1403 data=data,
1404 how_generated=f"during {phase} phase{backend_desc}",
1405 representation=self._string_repr,
1406 arguments=data._observability_args,
1407 timing=self._timing_features,
1408 coverage=tractable_coverage_report(trace) or None,
1409 phase=phase,
1410 backend_metadata=data.provider.observe_test_case(),
1411 )
1412 deliver_observation(tc)
1414 for msg in data.provider.observe_information_messages(
1415 lifetime="test_case"
1416 ):
1417 self._deliver_information_message(**msg)
1418 self._timing_features = {}
1420 def _deliver_information_message(
1421 self, *, type: InfoObservationType, title: str, content: str | dict
1422 ) -> None:
1423 deliver_observation(
1424 InfoObservation(
1425 type=type,
1426 run_start=self._start_timestamp,
1427 property=self.test_identifier,
1428 title=title,
1429 content=content,
1430 )
1431 )
1433 def run_engine(self):
1434 """Run the test function many times, on database input and generated
1435 input, using the Conjecture engine.
1436 """
1437 # Tell pytest to omit the body of this function from tracebacks
1438 __tracebackhide__ = True
1439 try:
1440 database_key = self.wrapped_test._hypothesis_internal_database_key
1441 except AttributeError:
1442 if global_force_seed is None:
1443 database_key = function_digest(self.test)
1444 else:
1445 database_key = None
1447 runner = ConjectureRunner(
1448 self._execute_once_for_engine,
1449 settings=self.settings,
1450 random=self.random,
1451 database_key=database_key,
1452 thread_overlap=self.thread_overlap,
1453 )
1454 self._runner = runner
1455 # Use the Conjecture engine to run the test function many times
1456 # on different inputs.
1457 runner.run()
1458 note_statistics(runner.statistics)
1459 if observability_enabled():
1460 self._deliver_information_message(
1461 type="info",
1462 title="Hypothesis Statistics",
1463 content=describe_statistics(runner.statistics),
1464 )
1465 for msg in (
1466 p if isinstance(p := runner.provider, PrimitiveProvider) else p(None)
1467 ).observe_information_messages(lifetime="test_function"):
1468 self._deliver_information_message(**msg)
1470 if runner.call_count == 0:
1471 return
1472 if runner.interesting_test_cases:
1473 self.falsifying_test_cases = sorted(
1474 runner.interesting_test_cases.values(),
1475 key=lambda d: sort_key(d.nodes),
1476 reverse=True,
1477 )
1478 else:
1479 if runner.valid_test_cases == 0:
1480 explanations = []
1481 if runner.exit_reason is ExitReason.finished:
1482 explanations.append(
1483 "Hypothesis tried every possible input, so the "
1484 "assumptions of this test are impossible to satisfy - "
1485 "not merely unlikely - and running more test cases "
1486 "cannot help."
1487 )
1488 # use a somewhat arbitrary cutoff to avoid recommending spurious
1489 # fixes.
1490 # eg, a few invalid test cases from internal filters when the
1491 # problem is the user generating large inputs, or a
1492 # few overruns during internal mutation when the problem is
1493 # impossible user filters/assumes.
1494 if runner.invalid_test_cases > min(20, runner.call_count // 5):
1495 explanations.append(
1496 f"{runner.invalid_test_cases} of {runner.call_count} "
1497 "test cases failed a .filter() or assume() condition. Try "
1498 "making your filters or assumes less strict, or rewrite "
1499 "using strategy parameters: "
1500 "st.integers().filter(lambda x: x > 0) fails less often "
1501 "(that is, never) when rewritten as st.integers(min_value=1)."
1502 )
1503 if runner.overrun_test_cases > min(20, runner.call_count // 5):
1504 explanations.append(
1505 f"{runner.overrun_test_cases} of {runner.call_count} "
1506 "test cases were too large to finish generating; try "
1507 "reducing the typical size of your inputs?"
1508 )
1509 rep = get_pretty_function_description(self.test)
1510 raise Unsatisfiable(
1511 f"Unable to satisfy assumptions of {rep}. "
1512 f"{' Also, '.join(explanations)}"
1513 )
1515 # If we have not traced executions, warn about that now (but only when
1516 # we'd expect to do so reliably, i.e. on CPython>=3.12)
1517 if (
1518 hasattr(sys, "monitoring")
1519 and not PYPY
1520 and self._should_trace()
1521 and not Tracer.can_trace()
1522 ):
1523 warnings.warn(
1524 "avoiding tracing test function because tool id "
1525 f"{MONITORING_TOOL_ID} is already taken by tool "
1526 f"{sys.monitoring.get_tool(MONITORING_TOOL_ID)}.",
1527 HypothesisWarning,
1528 stacklevel=3,
1529 )
1531 if not self.falsifying_test_cases:
1532 return
1533 elif not (self.settings.report_multiple_bugs and pytest_shows_exceptiongroups):
1534 # Pretend that we only found one failure, by discarding the others.
1535 del self.falsifying_test_cases[:-1]
1537 # The engine found one or more failures, so we need to reproduce and
1538 # report them.
1540 errors_to_report = []
1542 report_lines = describe_targets(runner.best_observed_targets)
1543 if report_lines:
1544 report_lines.append("")
1546 explanations = explanatory_lines(self.explain_traces, self.settings)
1547 for falsifying_test_case in self.falsifying_test_cases:
1548 fragments = []
1550 ran_test_case = runner.new_conjecture_data(
1551 falsifying_test_case.choices,
1552 max_choices=len(falsifying_test_case.choices),
1553 )
1554 ran_test_case.span_comments = falsifying_test_case.span_comments
1555 tb = None
1556 origin = None
1557 assert falsifying_test_case.expected_exception is not None
1558 expected_traceback = falsifying_test_case.expected_traceback
1559 if expected_traceback is None:
1560 exc = falsifying_test_case.expected_exception
1561 assert exc is not None
1562 expected_traceback = format_exception(exc, get_trimmed_traceback(exc))
1563 try:
1564 with with_reporter(fragments.append):
1565 self.execute_once(
1566 ran_test_case,
1567 print_test_case=True,
1568 is_final=True,
1569 expected_failure=(
1570 falsifying_test_case.expected_exception,
1571 expected_traceback,
1572 ),
1573 )
1574 except StopTest as e:
1575 # Link the expected exception from the first run. Not sure
1576 # how to access the current exception, if it failed
1577 # differently on this run. In fact, in the only known
1578 # reproducer, the StopTest is caused by OVERRUN before the
1579 # test is even executed. Possibly because all initial test cases
1580 # failed until the final non-traced replay, and something was
1581 # exhausted? Possibly a FIXME, but sufficiently weird to
1582 # ignore for now.
1583 err = FlakyFailure(
1584 "Inconsistent results: A test case failed on the "
1585 "first run but now succeeds (or fails with another "
1586 "error, or is for some reason not runnable).",
1587 # (note: e is a BaseException)
1588 [falsifying_test_case.expected_exception or e],
1589 )
1590 errors_to_report.append(ReportableError(fragments, err))
1591 except UnsatisfiedAssumption as e: # pragma: no cover # ironically flaky
1592 err = FlakyFailure(
1593 "Unreliable assumption: A test case which satisfied "
1594 "assumptions on the first run now fails it.",
1595 [e],
1596 )
1597 errors_to_report.append(ReportableError(fragments, err))
1598 except BaseException as e:
1599 # If we have anything for explain-mode, this is the time to report.
1600 fragments.extend(explanations[falsifying_test_case.interesting_origin])
1601 error_with_tb = e.with_traceback(get_trimmed_traceback())
1602 errors_to_report.append(ReportableError(fragments, error_with_tb))
1603 tb = format_exception(e, get_trimmed_traceback(e))
1604 origin = InterestingOrigin.from_exception(e)
1605 else:
1606 # execute_once() will always raise either the expected error, or Flaky.
1607 raise NotImplementedError("This should be unreachable")
1608 finally:
1609 ran_test_case.freeze()
1610 if observability_enabled():
1611 # log our observability line for the final failing test case
1612 tc = make_testcase(
1613 run_start=self._start_timestamp,
1614 property=self.test_identifier,
1615 data=ran_test_case,
1616 how_generated="minimal failing test case",
1617 representation=self._string_repr,
1618 arguments=ran_test_case._observability_args,
1619 timing=self._timing_features,
1620 coverage=None, # Not recorded when we're replaying the MFE
1621 status="passed" if sys.exc_info()[0] else "failed",
1622 status_reason=str(origin or "unexpected/flaky pass"),
1623 metadata={
1624 "traceback": tb,
1625 "status_reason_location": (
1626 f"{origin.filename}:{origin.lineno}"
1627 if origin and origin.filename
1628 else None
1629 ),
1630 },
1631 )
1632 deliver_observation(tc)
1634 # Whether or not replay actually raised the exception again, we want
1635 # to print the reproduce_failure decorator for the failing test case.
1636 if self.settings.print_blob:
1637 fragments.append(
1638 "\nYou can reproduce this test case by temporarily adding "
1639 f"{reproduction_decorator(falsifying_test_case.choices)} "
1640 "as a decorator on your test function"
1641 )
1643 _raise_to_user(
1644 errors_to_report,
1645 self.settings,
1646 report_lines,
1647 # A backend might report a failure and then report verified afterwards,
1648 # which is to be interpreted as "there are no more failures *other
1649 # than what we already reported*". Do not report this as unsound.
1650 unsound_backend=(
1651 runner._verified_by_backend
1652 if runner._verified_by_backend and not runner._backend_found_failure
1653 else None
1654 ),
1655 )
1658def _simplify_explicit_errors(errors: list[ReportableError]) -> list[ReportableError]:
1659 """
1660 Group explicit example errors by their InterestingOrigin, keeping only the
1661 simplest one, and adding a note of how many other examples failed with the same
1662 error.
1663 """
1664 by_origin: dict[InterestingOrigin, list[ReportableError]] = defaultdict(list)
1665 for error in errors:
1666 origin = InterestingOrigin.from_exception(error.exception)
1667 by_origin[origin].append(error)
1669 result = []
1670 for group in by_origin.values():
1671 if len(group) == 1:
1672 result.append(group[0])
1673 else:
1674 # Sort by shortlex of representation (first fragment)
1675 def shortlex_key(error):
1676 repr_str = error.fragments[0] if error.fragments else ""
1677 return (len(repr_str), repr_str)
1679 sorted_group = sorted(group, key=shortlex_key)
1680 simplest = sorted_group[0]
1681 other_count = len(group) - 1
1682 add_note(
1683 simplest.exception,
1684 f"(note: {other_count} other explicit example{'s' * (other_count > 1)} "
1685 "also failed with this error; use Verbosity.verbose to view)",
1686 )
1687 result.append(simplest)
1689 return result
1692def _raise_to_user(
1693 errors_to_report, settings, target_lines, trailer="", *, unsound_backend=None
1694):
1695 """Helper function for attaching notes and grouping multiple errors."""
1696 failing_prefix = "Failing test case: "
1697 ls = []
1698 for error in errors_to_report:
1699 for note in error.fragments:
1700 add_note(error.exception, note)
1701 if note.startswith(failing_prefix):
1702 ls.append(note.removeprefix(failing_prefix))
1703 if current_pytest_item.value:
1704 current_pytest_item.value._hypothesis_failing_examples = ls
1706 if len(errors_to_report) == 1:
1707 the_error_hypothesis_found = errors_to_report[0].exception
1708 else:
1709 assert errors_to_report
1710 the_error_hypothesis_found = BaseExceptionGroup(
1711 f"Hypothesis found {len(errors_to_report)} distinct failures{trailer}.",
1712 [error.exception for error in errors_to_report],
1713 )
1715 if settings.verbosity >= Verbosity.normal:
1716 for line in target_lines:
1717 add_note(the_error_hypothesis_found, line)
1719 if unsound_backend:
1720 add_note(
1721 the_error_hypothesis_found,
1722 f"backend={unsound_backend!r} claimed to verify this test passes - "
1723 "please send them a bug report!",
1724 )
1726 raise the_error_hypothesis_found
1729@contextlib.contextmanager
1730def fake_subTest(self, msg=None, **__):
1731 """Monkeypatch for `unittest.TestCase.subTest` during `@given`.
1733 If we don't patch this out, each failing test case is reported as a
1734 separate failing test by the unittest test runner, which is
1735 obviously incorrect. We therefore replace it for the duration with
1736 this version.
1737 """
1738 warnings.warn(
1739 "subTest per-test-case reporting interacts badly with Hypothesis "
1740 "trying hundreds of test cases, so we disable it for the duration of "
1741 "any test that uses `@given`.",
1742 HypothesisWarning,
1743 stacklevel=2,
1744 )
1745 yield
1748@dataclass(slots=False, frozen=False)
1749class HypothesisHandle:
1750 """This object is provided as the .hypothesis attribute on @given tests.
1752 Downstream users can reassign its attributes to insert custom logic into
1753 the execution of each case, for example by converting an async into a
1754 sync function.
1756 This must be an attribute of an attribute, because reassignment of a
1757 first-level attribute would not be visible to Hypothesis if the function
1758 had been decorated before the assignment.
1760 See https://github.com/HypothesisWorks/hypothesis/issues/1257 for more
1761 information.
1762 """
1764 inner_test: Any
1765 _get_fuzz_target: Any
1766 _given_kwargs: Any
1768 @property
1769 def fuzz_one_input(
1770 self,
1771 ) -> Callable[[bytes | bytearray | memoryview | BinaryIO], bytes | None]:
1772 """
1773 Run the test as a fuzz target, driven with the ``buffer`` of bytes.
1775 Depending on the passed ``buffer`` one of three things will happen:
1777 * If the bytestring was invalid, for example because it was too short or was
1778 filtered out by |assume| or |.filter|, |fuzz_one_input| returns ``None``.
1779 * If the bytestring was valid and the test passed, |fuzz_one_input| returns
1780 a canonicalised and pruned bytestring which will replay that |test case|.
1781 This is provided as an option to improve the performance of mutating
1782 fuzzers, but can safely be ignored.
1783 * If the test *failed*, i.e. raised an exception, |fuzz_one_input| will
1784 add the pruned buffer to :ref:`the Hypothesis example database <database>`
1785 and then re-raise that exception. All you need to do to reproduce,
1786 minimize, and de-duplicate all the failures found via fuzzing is run
1787 your test suite!
1789 To reduce the performance impact of database writes, |fuzz_one_input| only
1790 records failing inputs which would be valid shrinks for a known failure -
1791 meaning writes are somewhere between constant and log(N) rather than linear
1792 in runtime. However, this tracking only works within a persistent fuzzing
1793 process; for forkserver fuzzers we recommend ``database=None`` for the main
1794 run, and then replaying with a database enabled if you need to analyse
1795 failures.
1797 Note that the interpretation of both input and output bytestrings is
1798 specific to the exact version of Hypothesis you are using and the strategies
1799 given to the test, just like the :ref:`database <database>` and
1800 |@reproduce_failure|.
1802 Interaction with |@settings|
1803 ----------------------------
1805 |fuzz_one_input| uses just enough of Hypothesis' internals to drive your
1806 test function with a bytestring, and most settings therefore have no effect
1807 in this mode. We recommend running your tests the usual way before fuzzing
1808 to get the benefits of health checks, as well as afterwards to replay,
1809 shrink, deduplicate, and report whatever errors were discovered.
1811 * |settings.database| *is* used by |fuzz_one_input| - adding failures to
1812 the database to be replayed when
1813 you next run your tests is our preferred reporting mechanism and response
1814 to `the 'fuzzer taming' problem <https://blog.regehr.org/archives/925>`__.
1815 * |settings.verbosity| and |settings.stateful_step_count| work as usual.
1816 * The |~settings.deadline|, |~settings.derandomize|, |~settings.max_examples|,
1817 |~settings.phases|, |~settings.print_blob|, |~settings.report_multiple_bugs|,
1818 and |~settings.suppress_health_check| settings do not affect |fuzz_one_input|.
1820 Example Usage
1821 -------------
1823 .. code-block:: python
1825 @given(st.text())
1826 def test_foo(s): ...
1828 # This is a traditional fuzz target - call it with a bytestring,
1829 # or a binary IO object, and it runs the test once.
1830 fuzz_target = test_foo.hypothesis.fuzz_one_input
1832 # For example:
1833 fuzz_target(b"\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00")
1834 fuzz_target(io.BytesIO(b"\\x01"))
1836 .. tip::
1838 If you expect to discover many failures while using |fuzz_one_input|,
1839 consider wrapping your database with |BackgroundWriteDatabase|, for
1840 low-overhead writes of failures.
1842 .. tip::
1844 | Want an integrated workflow for your team's local tests, CI, and continuous fuzzing?
1845 | Use `HypoFuzz <https://hypofuzz.com/>`__ to fuzz your whole test suite, and find more bugs with the same tests!
1847 .. seealso::
1849 See also the :doc:`/how-to/external-fuzzers` how-to.
1850 """
1851 # Note: most users, if they care about fuzzer performance, will access the
1852 # property and assign it to a local variable to move the attribute lookup
1853 # outside their fuzzing loop / before the fork point. We cache it anyway,
1854 # so that naive or unusual use-cases get the best possible performance too.
1855 try:
1856 return self.__cached_target # type: ignore
1857 except AttributeError:
1858 self.__cached_target = self._get_fuzz_target()
1859 return self.__cached_target
1862@overload
1863def given(
1864 _: EllipsisType, /
1865) -> Callable[
1866 [Callable[..., Coroutine[Any, Any, None] | None]], Callable[[], None]
1867]: ...
1870@overload
1871def given(
1872 *_given_arguments: SearchStrategy[Any],
1873) -> Callable[
1874 [Callable[..., Coroutine[Any, Any, None] | None]], Callable[..., None]
1875]: ...
1878@overload
1879def given(
1880 **_given_kwargs: SearchStrategy[Any] | EllipsisType,
1881) -> Callable[
1882 [Callable[..., Coroutine[Any, Any, None] | None]], Callable[..., None]
1883]: ...
1886def given(
1887 *_given_arguments: SearchStrategy[Any] | EllipsisType,
1888 **_given_kwargs: SearchStrategy[Any] | EllipsisType,
1889) -> Callable[[Callable[..., Coroutine[Any, Any, None] | None]], Callable[..., None]]:
1890 """
1891 The |@given| decorator turns a function into a Hypothesis test. This is the
1892 main entry point to Hypothesis.
1894 .. seealso::
1896 See also the :doc:`/tutorial/introduction` tutorial, which introduces
1897 defining Hypothesis tests with |@given|.
1899 .. _given-arguments:
1901 Arguments to ``@given``
1902 -----------------------
1904 Arguments to |@given| may be either positional or keyword arguments:
1906 .. code-block:: python
1908 @given(st.integers(), st.floats())
1909 def test_one(x, y):
1910 pass
1912 @given(x=st.integers(), y=st.floats())
1913 def test_two(x, y):
1914 pass
1916 If using keyword arguments, the arguments may appear in any order, as with
1917 standard Python functions:
1919 .. code-block:: python
1921 # different order, but still equivalent to before
1922 @given(y=st.floats(), x=st.integers())
1923 def test(x, y):
1924 assert isinstance(x, int)
1925 assert isinstance(y, float)
1927 If |@given| is provided fewer positional arguments than the decorated test,
1928 the test arguments are filled in on the right side, leaving the leftmost
1929 positional arguments unfilled:
1931 .. code-block:: python
1933 @given(st.integers(), st.floats())
1934 def test(manual_string, y, z):
1935 assert manual_string == "x"
1936 assert isinstance(y, int)
1937 assert isinstance(z, float)
1939 # `test` is now a callable which takes one argument `manual_string`
1941 test("x")
1942 # or equivalently:
1943 test(manual_string="x")
1945 The reason for this "from the right" behavior is to support using |@given|
1946 with instance methods, by automatically passing through ``self``:
1948 .. code-block:: python
1950 class MyTest(TestCase):
1951 @given(st.integers())
1952 def test(self, x):
1953 assert isinstance(self, MyTest)
1954 assert isinstance(x, int)
1956 If (and only if) using keyword arguments, |@given| may be combined with
1957 ``**kwargs`` or ``*args``:
1959 .. code-block:: python
1961 @given(x=integers(), y=integers())
1962 def test(x, **kwargs):
1963 assert "y" in kwargs
1965 @given(x=integers(), y=integers())
1966 def test(x, *args, **kwargs):
1967 assert args == ()
1968 assert "x" not in kwargs
1969 assert "y" in kwargs
1971 It is an error to:
1973 * Mix positional and keyword arguments to |@given|.
1974 * Use |@given| with a function that has a default value for an argument.
1975 * Use |@given| with positional arguments with a function that uses ``*args``,
1976 ``**kwargs``, or keyword-only arguments.
1978 The function returned by given has all the same arguments as the original
1979 test, minus those that are filled in by |@given|. See the :ref:`notes on
1980 framework compatibility <framework-compatibility>` for how this interacts
1981 with features of other testing libraries, such as :pypi:`pytest` fixtures.
1982 """
1984 if currently_in_test_context():
1985 fail_health_check(
1986 Settings(),
1987 "Nesting @given tests results in quadratic generation and shrinking "
1988 "behavior, and can usually be more cleanly expressed by replacing the "
1989 "inner function with an st.data() parameter on the outer @given."
1990 "\n\n"
1991 "If it is difficult or impossible to refactor this test to remove the "
1992 "nested @given, you can disable this health check with "
1993 "@settings(suppress_health_check=[HealthCheck.nested_given]) on the "
1994 "outer @given. See "
1995 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
1996 "for details.",
1997 HealthCheck.nested_given,
1998 )
2000 def run_test_as_given(test):
2001 if inspect.isclass(test):
2002 # Provide a meaningful error to users, instead of exceptions from
2003 # internals that assume we're dealing with a function.
2004 raise InvalidArgument("@given cannot be applied to a class")
2006 if (
2007 "_pytest" in sys.modules
2008 and "_pytest.fixtures" in sys.modules
2009 and (
2010 tuple(map(int, sys.modules["_pytest"].__version__.split(".")[:2]))
2011 >= (8, 4)
2012 )
2013 and isinstance(
2014 test, sys.modules["_pytest.fixtures"].FixtureFunctionDefinition
2015 )
2016 ):
2017 raise InvalidArgument("@given cannot be applied to a pytest fixture")
2019 given_arguments = tuple(_given_arguments)
2020 given_kwargs = dict(_given_kwargs)
2022 original_sig = get_signature(test)
2023 if given_arguments == (Ellipsis,) and not given_kwargs:
2024 # user indicated that they want to infer all arguments
2025 given_kwargs = {
2026 p.name: Ellipsis
2027 for p in original_sig.parameters.values()
2028 if p.kind in (p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY)
2029 }
2030 given_arguments = ()
2032 check_invalid = is_invalid_test(
2033 test, original_sig, given_arguments, given_kwargs
2034 )
2036 # If the argument check found problems, return a dummy test function
2037 # that will raise an error if it is actually called.
2038 if check_invalid is not None:
2039 return check_invalid
2041 # Because the argument check succeeded, we can convert @given's
2042 # positional arguments into keyword arguments for simplicity.
2043 if given_arguments:
2044 assert not given_kwargs
2045 posargs = [
2046 p.name
2047 for p in original_sig.parameters.values()
2048 if p.kind is p.POSITIONAL_OR_KEYWORD
2049 ]
2050 given_kwargs = dict(
2051 list(zip(posargs[::-1], given_arguments[::-1], strict=False))[::-1]
2052 )
2053 # These have been converted, so delete them to prevent accidental use.
2054 del given_arguments
2056 new_signature = new_given_signature(original_sig, given_kwargs)
2058 # Use type information to convert "infer" arguments into appropriate strategies.
2059 if ... in given_kwargs.values():
2060 hints = get_type_hints(test)
2061 for name in [name for name, value in given_kwargs.items() if value is ...]:
2062 if name not in hints:
2063 return _invalid(
2064 f"passed {name}=... for {test.__name__}, but {name} has "
2065 "no type annotation",
2066 test=test,
2067 given_kwargs=given_kwargs,
2068 )
2069 given_kwargs[name] = st.from_type(hints[name])
2071 # only raise if the same thread uses two different executors, not if two
2072 # different threads use different executors.
2073 thread_local = ThreadLocal(prev_self=lambda: not_set)
2074 # maps thread_id to whether that thread overlaps in execution with any
2075 # other thread in this @given. We use this to detect whether an @given is
2076 # being run from multiple different threads at once, which informs
2077 # decisions like whether to raise DeadlineExceeded or HealthCheck.too_slow.
2078 thread_overlap: dict[int, bool] = {}
2079 thread_overlap_lock = Lock()
2081 @impersonate(test)
2082 @define_function_signature(test.__name__, test.__doc__, new_signature)
2083 def wrapped_test(*arguments, **kwargs):
2084 # Tell pytest to omit the body of this function from tracebacks
2085 __tracebackhide__ = True
2086 with thread_overlap_lock:
2087 for overlap_thread_id in thread_overlap:
2088 thread_overlap[overlap_thread_id] = True
2090 threadid = threading.get_ident()
2091 # if there are existing threads when this thread starts, then
2092 # this thread starts at an overlapped state.
2093 has_existing_threads = len(thread_overlap) > 0
2094 thread_overlap[threadid] = has_existing_threads
2096 try:
2097 test = wrapped_test.hypothesis.inner_test
2098 if getattr(test, "is_hypothesis_test", False):
2099 raise InvalidArgument(
2100 f"You have applied @given to the test {test.__name__} more than "
2101 "once, which wraps the test several times and is extremely slow. "
2102 "A similar effect can be gained by combining the arguments "
2103 "of the two calls to given. For example, instead of "
2104 "@given(booleans()) @given(integers()), you could write "
2105 "@given(booleans(), integers())"
2106 )
2108 settings = wrapped_test._hypothesis_internal_use_settings
2109 random = get_random_for_wrapped_test(test, wrapped_test)
2110 arguments, kwargs, stuff = process_arguments_to_given(
2111 wrapped_test,
2112 arguments,
2113 kwargs,
2114 given_kwargs,
2115 new_signature.parameters,
2116 )
2118 if (
2119 inspect.iscoroutinefunction(test)
2120 and get_executor(stuff.selfy) is default_executor
2121 ):
2122 # See https://github.com/HypothesisWorks/hypothesis/issues/3054
2123 # If our custom executor doesn't handle coroutines, or we return an
2124 # awaitable from a non-async-def function, we just rely on the
2125 # return_value health check. This catches most user errors though.
2126 raise InvalidArgument(
2127 "Hypothesis doesn't know how to run async test functions like "
2128 f"{test.__name__}. You'll need to write a custom executor, "
2129 "or use a library like pytest-asyncio or pytest-trio which can "
2130 "handle the translation for you.\n See https://hypothesis."
2131 "readthedocs.io/en/latest/details.html#custom-function-execution"
2132 )
2134 runner = stuff.selfy
2135 if isinstance(stuff.selfy, TestCase) and test.__name__ in dir(TestCase):
2136 fail_health_check(
2137 settings,
2138 f"You have applied @given to the method {test.__name__}, which is "
2139 "used by the unittest runner but is not itself a test. "
2140 "This is not useful in any way.",
2141 HealthCheck.not_a_test_method,
2142 )
2143 if bad_django_TestCase(runner): # pragma: no cover
2144 # Covered by the Django tests, but not the pytest coverage task
2145 raise InvalidArgument(
2146 "You have applied @given to a method on "
2147 f"{type(runner).__qualname__}, but this "
2148 "class does not inherit from the supported versions in "
2149 "`hypothesis.extra.django`. Use the Hypothesis variants "
2150 "to ensure that each test case is run in a separate "
2151 "database transaction."
2152 )
2154 nonlocal thread_local
2155 # Check selfy really is self (not e.g. a mock) before we health-check
2156 cur_self = (
2157 stuff.selfy
2158 if getattr(type(stuff.selfy), test.__name__, None) is wrapped_test
2159 else None
2160 )
2161 if thread_local.prev_self is not_set:
2162 thread_local.prev_self = cur_self
2163 elif cur_self is not thread_local.prev_self:
2164 fail_health_check(
2165 settings,
2166 f"The method {test.__qualname__} was called from multiple "
2167 "different executors. This may lead to flaky tests and "
2168 "nonreproducible errors when replaying from database."
2169 "\n\n"
2170 "Unlike most health checks, HealthCheck.differing_executors "
2171 "warns about a correctness issue with your test. We "
2172 "therefore recommend fixing the underlying issue, rather "
2173 "than suppressing this health check. However, if you are "
2174 "confident this health check can be safely disabled, you can "
2175 "do so with "
2176 "@settings(suppress_health_check=[HealthCheck.differing_executors]). "
2177 "See "
2178 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
2179 "for details.",
2180 HealthCheck.differing_executors,
2181 )
2183 state = StateForActualGivenExecution(
2184 stuff,
2185 test,
2186 settings,
2187 random,
2188 wrapped_test,
2189 thread_overlap=thread_overlap,
2190 )
2192 # If there was a @reproduce_failure decorator, use it to reproduce
2193 # the error (or complain that we couldn't). Either way, this will
2194 # always raise some kind of error.
2195 if (
2196 reproduce_failure := wrapped_test._hypothesis_internal_use_reproduce_failure
2197 ) is not None:
2198 expected_version, failure = reproduce_failure
2199 if expected_version != __version__:
2200 raise InvalidArgument(
2201 "Attempting to reproduce a failure from a different "
2202 f"version of Hypothesis. This failure is from {expected_version}, but "
2203 f"you are currently running {__version__!r}. Please change your "
2204 "Hypothesis version to a matching one."
2205 )
2206 try:
2207 state.execute_once(
2208 ConjectureData.for_choices(decode_failure(failure)),
2209 print_test_case=True,
2210 is_final=True,
2211 )
2212 raise DidNotReproduce(
2213 "Expected the test to raise an error, but it "
2214 "completed successfully."
2215 )
2216 except StopTest:
2217 raise DidNotReproduce(
2218 "The shape of the test data has changed in some way "
2219 "from where this blob was defined. Are you sure "
2220 "you're running the same test?"
2221 ) from None
2222 except UnsatisfiedAssumption:
2223 raise DidNotReproduce(
2224 "The test data failed to satisfy an assumption in the "
2225 "test. Have you added it since this blob was generated?"
2226 ) from None
2228 # There was no @reproduce_failure, so start by running any explicit
2229 # examples from @example decorators.
2230 if errors := list(
2231 execute_explicit_examples(
2232 state, wrapped_test, arguments, kwargs, original_sig
2233 )
2234 ):
2235 # If we're not going to report multiple bugs, we would have
2236 # stopped running explicit examples at the first failure.
2237 assert len(errors) == 1 or state.settings.report_multiple_bugs
2239 # If an explicit example raised a 'skip' exception, ensure it's never
2240 # wrapped up in an exception group. Because we break out of the loop
2241 # immediately on finding a skip, if present it's always the last error.
2242 if isinstance(errors[-1].exception, skip_exceptions_to_reraise()):
2243 del errors[:-1]
2245 if state.settings.verbosity < Verbosity.verbose:
2246 # keep only one error per interesting origin, unless
2247 # verbosity is high
2248 errors = _simplify_explicit_errors(errors)
2250 _raise_to_user(errors, state.settings, [], " in explicit examples")
2252 # If there were any explicit examples, they all ran successfully.
2253 # The next step is to use the Conjecture engine to run the test on
2254 # many different inputs.
2255 ran_explicit_examples = (
2256 Phase.explicit in state.settings.phases
2257 and getattr(wrapped_test, "hypothesis_explicit_examples", ())
2258 )
2259 SKIP_BECAUSE_NO_TEST_CASES = unittest.SkipTest(
2260 "Hypothesis has been told to run no test cases for this test."
2261 )
2262 if not (
2263 Phase.reuse in settings.phases or Phase.generate in settings.phases
2264 ):
2265 if not ran_explicit_examples:
2266 raise SKIP_BECAUSE_NO_TEST_CASES
2267 return
2269 try:
2270 if isinstance(runner, TestCase) and hasattr(runner, "subTest"):
2271 subTest = runner.subTest
2272 try:
2273 runner.subTest = types.MethodType(fake_subTest, runner)
2274 state.run_engine()
2275 finally:
2276 runner.subTest = subTest
2277 else:
2278 state.run_engine()
2279 except BaseException as e:
2280 # The exception caught here should either be an actual test
2281 # failure (or BaseExceptionGroup), or some kind of fatal error
2282 # that caused the engine to stop.
2283 generated_seed = (
2284 wrapped_test._hypothesis_internal_use_generated_seed
2285 )
2286 assert state._runner is not None
2287 stopped_because_slow_shrinking = (
2288 state._runner.statistics.get("stopped-because")
2289 == "shrinking was very slow"
2290 )
2291 with local_settings(settings):
2292 if generated_seed is not None and (
2293 not state.failed_normally or stopped_because_slow_shrinking
2294 ):
2295 pytest_extra_msg = (
2296 (
2297 ", or by running pytest with "
2298 f"--hypothesis-seed={generated_seed}"
2299 )
2300 if running_under_pytest
2301 else ""
2302 )
2303 if stopped_because_slow_shrinking:
2304 msg = (
2305 "\nThis test function exited early because"
2306 " it took too long to shrink. If desired for debugging, "
2307 f"you can reproduce this by adding @seed({generated_seed}) "
2308 f"to this test{pytest_extra_msg}."
2309 )
2310 else:
2311 msg = (
2312 "You can reproduce this failure by adding "
2313 f"@seed({generated_seed}) to this test"
2314 f"{pytest_extra_msg}."
2315 )
2316 report(msg)
2317 # The dance here is to avoid showing users long tracebacks
2318 # full of Hypothesis internals they don't care about.
2319 # We have to do this inline, to avoid adding another
2320 # internal stack frame just when we've removed the rest.
2321 #
2322 # Using a variable for our trimmed error ensures that the line
2323 # which will actually appear in tracebacks is as clear as
2324 # possible - "raise the_error_hypothesis_found".
2325 the_error_hypothesis_found = e.with_traceback(
2326 None
2327 if isinstance(e, BaseExceptionGroup)
2328 else get_trimmed_traceback()
2329 )
2330 raise the_error_hypothesis_found
2332 if not (ran_explicit_examples or state.ever_executed):
2333 raise SKIP_BECAUSE_NO_TEST_CASES
2334 finally:
2335 with thread_overlap_lock:
2336 del thread_overlap[threadid]
2338 def _get_fuzz_target() -> (
2339 Callable[[bytes | bytearray | memoryview | BinaryIO], bytes | None]
2340 ):
2341 # Because fuzzing interfaces are very performance-sensitive, we use a
2342 # somewhat more complicated structure here. `_get_fuzz_target()` is
2343 # called by the `HypothesisHandle.fuzz_one_input` property, allowing
2344 # us to defer our collection of the settings, random instance, and
2345 # reassignable `inner_test` (etc) until `fuzz_one_input` is accessed.
2346 #
2347 # We then share the performance cost of setting up `state` between
2348 # many invocations of the target. We explicitly force `deadline=None`
2349 # for performance reasons, saving ~40% the runtime of an empty test.
2350 test = wrapped_test.hypothesis.inner_test
2351 settings = Settings(
2352 parent=wrapped_test._hypothesis_internal_use_settings, deadline=None
2353 )
2354 random = get_random_for_wrapped_test(test, wrapped_test)
2355 _args, _kwargs, stuff = process_arguments_to_given(
2356 wrapped_test, (), {}, given_kwargs, new_signature.parameters
2357 )
2358 assert not _args
2359 assert not _kwargs
2360 state = StateForActualGivenExecution(
2361 stuff,
2362 test,
2363 settings,
2364 random,
2365 wrapped_test,
2366 thread_overlap=thread_overlap,
2367 )
2368 database_key = function_digest(test) + b".secondary"
2369 # We track the minimal-so-far test case for each distinct origin, so
2370 # that we track log-n instead of n test cases for long runs. In particular
2371 # it means that we saturate for common errors in long runs instead of
2372 # storing huge volumes of low-value data.
2373 minimal_failures: dict = {}
2375 def fuzz_one_input(
2376 buffer: bytes | bytearray | memoryview | BinaryIO,
2377 ) -> bytes | None:
2378 # This inner part is all that the fuzzer will actually run,
2379 # so we keep it as small and as fast as possible.
2380 if isinstance(buffer, io.IOBase):
2381 buffer = buffer.read(BUFFER_SIZE)
2382 assert isinstance(buffer, (bytes, bytearray, memoryview))
2383 data = ConjectureData(
2384 random=None,
2385 provider=BytestringProvider,
2386 provider_kw={"bytestring": buffer},
2387 )
2388 try:
2389 state.execute_once(data)
2390 status = Status.VALID
2391 except StopTest:
2392 status = data.status
2393 return None
2394 except UnsatisfiedAssumption as e:
2395 status = Status.INVALID
2396 data.events["gave up because"] = e.reason or ""
2397 data.invalid_location = e.location
2398 return None
2399 except BaseException as e:
2400 # The engine sets data.interesting_origin in
2401 # _execute_once_for_engine, but fuzz_one_input calls
2402 # execute_once directly, so we replicate it here.
2403 data.interesting_origin = InterestingOrigin.from_exception(e)
2404 known = minimal_failures.get(data.interesting_origin)
2405 if settings.database is not None and (
2406 known is None or sort_key(data.nodes) <= sort_key(known)
2407 ):
2408 settings.database.save(
2409 database_key, choices_to_bytes(data.choices)
2410 )
2411 minimal_failures[data.interesting_origin] = data.nodes
2412 status = Status.INTERESTING
2413 raise
2414 finally:
2415 if observability_enabled():
2416 data.freeze()
2417 tc = make_testcase(
2418 run_start=state._start_timestamp,
2419 property=state.test_identifier,
2420 data=data,
2421 how_generated="fuzz_one_input",
2422 representation=state._string_repr,
2423 arguments=data._observability_args,
2424 timing=state._timing_features,
2425 coverage=None,
2426 status=status,
2427 backend_metadata=data.provider.observe_test_case(),
2428 )
2429 deliver_observation(tc)
2430 state._timing_features = {}
2432 assert isinstance(data.provider, BytestringProvider)
2433 return bytes(data.provider.drawn)
2435 fuzz_one_input.__doc__ = HypothesisHandle.fuzz_one_input.__doc__
2436 return fuzz_one_input
2438 # After having created the decorated test function, we need to copy
2439 # over some attributes to make the switch as seamless as possible.
2441 for attrib in dir(test):
2442 if not (attrib.startswith("_") or hasattr(wrapped_test, attrib)):
2443 setattr(wrapped_test, attrib, getattr(test, attrib))
2444 wrapped_test.is_hypothesis_test = True
2445 if hasattr(test, "_hypothesis_internal_settings_applied"):
2446 # Used to check if @settings is applied twice.
2447 wrapped_test._hypothesis_internal_settings_applied = True
2448 wrapped_test._hypothesis_internal_use_seed = getattr(
2449 test, "_hypothesis_internal_use_seed", None
2450 )
2451 wrapped_test._hypothesis_internal_use_settings = (
2452 getattr(test, "_hypothesis_internal_use_settings", None) or Settings.default
2453 )
2454 wrapped_test._hypothesis_internal_use_reproduce_failure = getattr(
2455 test, "_hypothesis_internal_use_reproduce_failure", None
2456 )
2457 wrapped_test.hypothesis = HypothesisHandle(test, _get_fuzz_target, given_kwargs)
2458 return wrapped_test
2460 return run_test_as_given
2463def find(
2464 specifier: SearchStrategy[Ex],
2465 condition: Callable[[Any], bool],
2466 *,
2467 settings: Settings | None = None,
2468 random: Random | None = None,
2469 database_key: bytes | None = None,
2470) -> Ex:
2471 """Returns the minimal value from the given strategy ``specifier`` that
2472 matches the predicate function ``condition``."""
2473 if settings is None:
2474 settings = Settings(max_examples=2000)
2475 settings = Settings(
2476 settings, suppress_health_check=list(HealthCheck), report_multiple_bugs=False
2477 )
2479 if database_key is None and settings.database is not None:
2480 # Note: The database key is not guaranteed to be unique. If not, replaying
2481 # of database test cases may fail to reproduce due to being replayed on the
2482 # wrong condition.
2483 database_key = function_digest(condition)
2485 if not isinstance(specifier, SearchStrategy):
2486 raise InvalidArgument(
2487 f"Expected SearchStrategy but got {specifier!r} of "
2488 f"type {type(specifier).__name__}"
2489 )
2490 specifier.validate()
2492 last: list[Ex] = []
2494 @settings
2495 @given(specifier)
2496 def test(v):
2497 if condition(v):
2498 last[:] = [v]
2499 raise Found
2501 if random is not None:
2502 test = seed(random.getrandbits(64))(test)
2504 test._hypothesis_internal_database_key = database_key # type: ignore
2506 try:
2507 test()
2508 except Found:
2509 return last[0]
2511 raise NoSuchExample(get_pretty_function_description(condition))