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