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