Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This file is part of Hypothesis, which may be found at
2# https://github.com/HypothesisWorks/hypothesis/
3#
4# Copyright the Hypothesis Authors.
5# Individual contributors are listed in AUTHORS.rst and the git log.
6#
7# This Source Code Form is subject to the terms of the Mozilla Public License,
8# v. 2.0. If a copy of the MPL was not distributed with this file, You can
9# obtain one at https://mozilla.org/MPL/2.0/.
11import importlib
12import math
13import threading
14import time
15from collections import defaultdict
16from collections.abc import Callable, Generator, Sequence
17from contextlib import AbstractContextManager, contextmanager, nullcontext
18from dataclasses import dataclass, field
19from datetime import timedelta
20from enum import Enum
21from random import Random
22from typing import Literal, NoReturn, NotRequired, TypedDict, cast
24from hypothesis import HealthCheck, Phase, Verbosity, settings as Settings
25from hypothesis._settings import local_settings
26from hypothesis.database import ExampleDatabase, choices_from_bytes, choices_to_bytes
27from hypothesis.errors import (
28 BackendCannotProceed,
29 FlakyBackendFailure,
30 FlakyStrategyDefinition,
31 HypothesisException,
32 InvalidArgument,
33 StopTest,
34)
35from hypothesis.internal.cache import LRUReusedCache
36from hypothesis.internal.compat import ceil, override
37from hypothesis.internal.conjecture.choice import (
38 ChoiceConstraintsT,
39 ChoiceKeyT,
40 ChoiceNode,
41 ChoiceT,
42 ChoiceTemplate,
43 ValueHole,
44 choices_key,
45)
46from hypothesis.internal.conjecture.data import (
47 ConjectureData,
48 ConjectureResult,
49 DataObserver,
50 Overrun,
51 Status,
52 _Overrun,
53)
54from hypothesis.internal.conjecture.datatree import (
55 DataTree,
56 PreviouslyUnseenBehaviour,
57 TreeRecordingObserver,
58)
59from hypothesis.internal.conjecture.junkdrawer import (
60 ensure_free_stackframes,
61 startswith,
62)
63from hypothesis.internal.conjecture.pareto import NO_SCORE, ParetoFront, ParetoOptimiser
64from hypothesis.internal.conjecture.providers import (
65 AVAILABLE_PROVIDERS,
66 HypothesisProvider,
67 PrimitiveProvider,
68)
69from hypothesis.internal.conjecture.shrinker import Shrinker, ShrinkPredicateT, sort_key
70from hypothesis.internal.escalation import InterestingOrigin
71from hypothesis.internal.healthcheck import fail_health_check
72from hypothesis.internal.observability import Observation, with_observability_callback
73from hypothesis.reporting import base_report, report, verbose_report
75# In most cases, the following constants are all Final. However, we do allow users
76# to monkeypatch all of these variables, which means we cannot annotate them as
77# Final or mypyc will inline them and render monkeypatching useless.
79#: The maximum number of times the shrinker will reduce the complexity of a failing
80#: input before giving up. This avoids falling down a trap of exponential (or worse)
81#: complexity, where the shrinker appears to be making progress but will take a
82#: substantially long time to finish completely.
83MAX_SHRINKS: int = 500
85# If the shrinking phase takes more than five minutes, abort it early and print
86# a warning. Many CI systems will kill a build after around ten minutes with
87# no output, and appearing to hang isn't great for interactive use either -
88# showing partially-shrunk test cases is better than quitting with no test cases!
89# (but make it monkeypatchable, for the rare users who need to keep on shrinking)
91#: The maximum total time in seconds that the shrinker will try to shrink a failure
92#: for before giving up. This is across all shrinks for the same failure, so even
93#: if the shrinker successfully reduces the complexity of a single failure several
94#: times, it will stop when it hits |MAX_SHRINKING_SECONDS| of total time taken.
95MAX_SHRINKING_SECONDS: int = 300
97#: The maximum amount of entropy a single test case can use before giving up
98#: while making random choices during input generation.
99#:
100#: The "unit" of one |BUFFER_SIZE| does not have any defined semantics, and you
101#: should not rely on it, except that a linear increase |BUFFER_SIZE| will linearly
102#: increase the amount of entropy a test case can use during generation.
103BUFFER_SIZE: int = 8 * 1024
104CACHE_SIZE: int = 10000
105MIN_TEST_CALLS: int = 10
107# we use this to isolate Hypothesis from interacting with the global random,
108# to make it easier to reason about our global random warning logic (see
109# deprecate_random_in_strategy).
110_random = Random()
113def shortlex(s):
114 return (len(s), s)
117@dataclass(slots=True, frozen=False)
118class HealthCheckState:
119 valid_test_cases: int = field(default=0)
120 invalid_test_cases: int = field(default=0)
121 overrun_test_cases: int = field(default=0)
122 draw_times: defaultdict[str, list[float]] = field(
123 default_factory=lambda: defaultdict(list)
124 )
126 @property
127 def total_draw_time(self) -> float:
128 return math.fsum(sum(self.draw_times.values(), start=[]))
130 def timing_report(self) -> str:
131 """Return a terminal report describing what was slow."""
132 if not self.draw_times:
133 return ""
134 width = max(
135 len(k.removeprefix("generate:").removesuffix(": ")) for k in self.draw_times
136 )
137 out = [f"\n {'':^{width}} count | fraction | slowest draws (seconds)"]
138 args_in_order = sorted(self.draw_times.items(), key=lambda kv: -sum(kv[1]))
139 for i, (argname, times) in enumerate(args_in_order): # pragma: no branch
140 # If we have very many unique keys, which can happen due to interactive
141 # draws with computed labels, we'll skip uninformative rows.
142 if (
143 5 <= i < (len(self.draw_times) - 2)
144 and math.fsum(times) * 20 < self.total_draw_time
145 ):
146 out.append(f" (skipped {len(self.draw_times) - i} rows of fast draws)")
147 break
148 # Compute the row to report, omitting times <1ms to focus on slow draws
149 reprs = [f"{t:>6.3f}," for t in sorted(times)[-5:] if t > 5e-4]
150 desc = " ".join(([" -- "] * 5 + reprs)[-5:]).rstrip(",")
151 arg = argname.removeprefix("generate:").removesuffix(": ")
152 out.append(
153 f" {arg:^{width}} | {len(times):>4} | "
154 f"{math.fsum(times)/self.total_draw_time:>7.0%} | {desc}"
155 )
156 return "\n".join(out)
159def _invalid_thresholds(*, r: float, c: float) -> tuple[int, int]:
160 base = math.ceil(math.log(1 - c) / math.log(1 - r)) - 1
161 per_p = math.ceil(1 / r)
162 return base, per_p
165# stop once we're 99% confident the true valid rate is below 1%. See
166# https://github.com/HypothesisWorks/hypothesis/issues/4623#issuecomment-3814681997
167# for how we derived this formula.
168INVALID_THRESHOLD_BASE, INVALID_PER_VALID = _invalid_thresholds(r=0.01, c=0.99)
171class ExitReason(Enum):
172 max_examples = "settings.max_examples={s.max_examples}"
173 max_iterations = (
174 "settings.max_examples={s.max_examples}, "
175 "but < 1% of test cases satisfied assumptions"
176 )
177 max_shrinks = f"shrunk test case {MAX_SHRINKS} times"
178 finished = "nothing left to do"
179 flaky = "test was flaky"
180 very_slow_shrinking = "shrinking was very slow"
182 def describe(self, settings: Settings) -> str:
183 return self.value.format(s=settings)
186class RunIsComplete(Exception):
187 pass
190def _get_provider(backend: str) -> PrimitiveProvider | type[PrimitiveProvider]:
191 provider_cls = AVAILABLE_PROVIDERS[backend]
192 if isinstance(provider_cls, str):
193 module_name, class_name = provider_cls.rsplit(".", 1)
194 provider_cls = getattr(importlib.import_module(module_name), class_name)
196 if provider_cls.lifetime == "test_function":
197 return provider_cls(None)
198 elif provider_cls.lifetime == "test_case":
199 return provider_cls
200 else:
201 raise InvalidArgument(
202 f"invalid lifetime {provider_cls.lifetime} for provider {provider_cls.__name__}. "
203 "Expected one of 'test_function', 'test_case'."
204 )
207class CallStats(TypedDict):
208 status: str
209 runtime: float
210 drawtime: float
211 gctime: float
212 events: list[str]
215PhaseStatistics = TypedDict(
216 "PhaseStatistics",
217 {
218 "duration-seconds": float,
219 "test-cases": list[CallStats],
220 "distinct-failures": int,
221 "shrinks-successful": int,
222 },
223)
224StatisticsDict = TypedDict(
225 "StatisticsDict",
226 {
227 "generate-phase": NotRequired[PhaseStatistics],
228 "reuse-phase": NotRequired[PhaseStatistics],
229 "shrink-phase": NotRequired[PhaseStatistics],
230 "explain-phase": NotRequired[PhaseStatistics],
231 "stopped-because": NotRequired[str],
232 "targets": NotRequired[dict[str, float]],
233 "nodeid": NotRequired[str],
234 },
235)
238def choice_count(
239 choices: Sequence[ChoiceT | ChoiceTemplate | ValueHole],
240) -> int | None:
241 count = 0
242 for choice in choices:
243 if isinstance(choice, ValueHole):
244 # expands to however many choices the inverting strategy emits
245 return None
246 elif isinstance(choice, ChoiceTemplate):
247 if choice.count is None:
248 return None
249 count += choice.count
250 else:
251 count += 1
252 return count
255class DiscardObserver(DataObserver):
256 @override
257 def kill_branch(self) -> NoReturn:
258 raise ContainsDiscard
261def realize_choices(data: ConjectureData, *, for_failure: bool) -> None:
262 for node in data.nodes:
263 value = data.provider.realize(node.value, for_failure=for_failure)
264 expected_type = {
265 "string": str,
266 "float": float,
267 "integer": int,
268 "boolean": bool,
269 "bytes": bytes,
270 }[node.type]
271 if type(value) is not expected_type:
272 raise HypothesisException(
273 f"expected {expected_type} from "
274 f"{data.provider.realize.__qualname__}, got {type(value)}"
275 )
277 constraints = cast(
278 ChoiceConstraintsT,
279 {
280 k: data.provider.realize(v, for_failure=for_failure)
281 for k, v in node.constraints.items()
282 },
283 )
284 node.value = value
285 node.constraints = constraints
288class ConjectureRunner:
289 def __init__(
290 self,
291 test_function: Callable[[ConjectureData], None],
292 *,
293 settings: Settings | None = None,
294 random: Random | None = None,
295 database_key: bytes | None = None,
296 ignore_limits: bool = False,
297 thread_overlap: dict[int, bool] | None = None,
298 ) -> None:
299 self._test_function: Callable[[ConjectureData], None] = test_function
300 self.settings: Settings = settings or Settings()
301 self.shrinks: int = 0
302 self.finish_shrinking_deadline: float | None = None
303 self.call_count: int = 0
304 self.misaligned_count: int = 0
305 self.valid_test_cases: int = 0
306 self.invalid_test_cases: int = 0
307 self.overrun_test_cases: int = 0
308 self.random: Random = random or Random(_random.getrandbits(128))
309 self.database_key: bytes | None = database_key
310 self.ignore_limits: bool = ignore_limits
311 self.thread_overlap = {} if thread_overlap is None else thread_overlap
313 # Global dict of per-phase statistics, and a list of per-call stats
314 # which transfer to the global dict at the end of each phase.
315 self._current_phase: str = "(not a phase)"
316 self.statistics: StatisticsDict = {}
317 self.stats_per_test_case: list[CallStats] = []
318 # Time spent in any nested phase, so the enclosing phase can exclude it.
319 self._nested_phase_seconds: float = 0.0
321 self.interesting_test_cases: dict[InterestingOrigin, ConjectureResult] = {}
322 # We use call_count because there may be few possible valid_test_cases.
323 self.first_bug_found_at: int | None = None
324 self.last_bug_found_at: int | None = None
325 self.first_bug_found_time: float = math.inf
327 self.shrunk_test_cases: set[InterestingOrigin] = set()
328 self.health_check_state: HealthCheckState | None = None
329 self.tree: DataTree = DataTree()
330 self.provider: PrimitiveProvider | type[PrimitiveProvider] = _get_provider(
331 self.settings.backend
332 )
334 self.best_observed_targets: defaultdict[str, float] = defaultdict(
335 lambda: NO_SCORE
336 )
337 self.best_test_cases_of_observed_targets: dict[str, ConjectureResult] = {}
339 # Scheduling state for the target phase. For large max_examples we
340 # interleave repeated optimisation passes with generation, aiming to
341 # spend up to half the budget on optimisation in total - see
342 # _should_optimise_now.
343 self._target_valid_spent: int = 0
344 self._next_optimise_at: float = math.inf
345 self._best_scores_at_last_pass: dict[str, float] | None = None
347 # We keep the pareto front in the example database if we have one. This
348 # is only marginally useful at present, but speeds up local development
349 # because it means that large targets will be quickly surfaced in your
350 # testing.
351 self.pareto_front: ParetoFront | None = None
352 if self.database_key is not None and self.settings.database is not None:
353 self.pareto_front = ParetoFront(self.random)
354 self.pareto_front.on_evict(self.on_pareto_evict)
356 # We want to be able to get the ConjectureData object that results
357 # from running a choice sequence without recalculating, especially during
358 # shrinking where we need to know about the structure of the
359 # executed test case.
360 self.__data_cache = LRUReusedCache[
361 tuple[ChoiceKeyT, ...], ConjectureResult | _Overrun
362 ](CACHE_SIZE)
364 self.reused_previously_shrunk_test_case: bool = False
366 self.__pending_call_explanation: str | None = None
367 self._backend_found_failure: bool = False
368 self._backend_exceeded_deadline: bool = False
369 self._backend_discard_count: int = 0
370 # note unsound verification by alt backends
371 self._verified_by_backend: str | None = None
372 self._switch_to_hypothesis_provider: bool = False
374 @contextmanager
375 def _with_switch_to_hypothesis_provider(
376 self, value: bool
377 ) -> Generator[None, None, None]:
378 previous = self._switch_to_hypothesis_provider
379 try:
380 self._switch_to_hypothesis_provider = value
381 yield
382 finally:
383 self._switch_to_hypothesis_provider = previous
385 @property
386 def using_hypothesis_backend(self) -> bool:
387 return (
388 self.settings.backend == "hypothesis" or self._switch_to_hypothesis_provider
389 )
391 def explain_next_call_as(self, explanation: str) -> None:
392 self.__pending_call_explanation = explanation
394 def clear_call_explanation(self) -> None:
395 self.__pending_call_explanation = None
397 @contextmanager
398 def _log_phase_statistics(
399 self, phase: Literal["reuse", "generate", "shrink", "explain"]
400 ) -> Generator[None, None, None]:
401 # Phases may nest - the explain phase runs inside the shrink phase - so
402 # we save and restore the per-call stats and current phase, exclude the
403 # duration of any nested phase, and accumulate when a phase is entered
404 # more than once (the explain phase runs once per shrinking target).
405 saved_stats = self.stats_per_test_case
406 saved_phase = self._current_phase
407 saved_nested_seconds = self._nested_phase_seconds
408 self.stats_per_test_case = []
409 self._current_phase = phase
410 self._nested_phase_seconds = 0.0
411 start_time = time.perf_counter()
412 try:
413 yield
414 finally:
415 elapsed = time.perf_counter() - start_time
416 # A phase can be entered more than once (the explain phase runs once
417 # per shrinking target), so accumulate into any existing bucket.
418 stats = self.statistics.setdefault(
419 phase + "-phase", # type: ignore
420 {"duration-seconds": 0.0, "test-cases": []},
421 )
422 stats["duration-seconds"] += elapsed - self._nested_phase_seconds
423 stats["test-cases"] += self.stats_per_test_case
424 stats["distinct-failures"] = len(self.interesting_test_cases)
425 stats["shrinks-successful"] = self.shrinks
426 self.stats_per_test_case = saved_stats
427 self._current_phase = saved_phase
428 self._nested_phase_seconds = saved_nested_seconds + elapsed
430 @property
431 def should_optimise(self) -> bool:
432 return Phase.target in self.settings.phases
434 def __tree_is_exhausted(self) -> bool:
435 return self.tree.is_exhausted and self.using_hypothesis_backend
437 def __stoppable_test_function(self, data: ConjectureData) -> None:
438 """Run ``self._test_function``, but convert a ``StopTest`` exception
439 into a normal return and avoid raising anything flaky for RecursionErrors.
440 """
441 # We ensure that the test has this much stack space remaining, no
442 # matter the size of the stack when called, to de-flake RecursionErrors
443 # (#2494, #3671). Note, this covers the data generation part of the test;
444 # the actual test execution is additionally protected at the call site
445 # in hypothesis.core.execute_once.
446 with ensure_free_stackframes():
447 try:
448 self._test_function(data)
449 except StopTest as e:
450 if e.testcounter == data.testcounter:
451 # This StopTest has successfully stopped its test, and can now
452 # be discarded.
453 pass
454 else:
455 # This StopTest was raised by a different ConjectureData. We
456 # need to re-raise it so that it will eventually reach the
457 # correct engine.
458 raise
460 def _cache_key(self, choices: Sequence[ChoiceT]) -> tuple[ChoiceKeyT, ...]:
461 return choices_key(choices)
463 def _cache(self, data: ConjectureData) -> None:
464 result = data.as_result()
465 key = self._cache_key(data.choices)
466 self.__data_cache[key] = result
468 def cached_test_function(
469 self,
470 choices: Sequence[ChoiceT | ChoiceTemplate | ValueHole],
471 *,
472 error_on_discard: bool = False,
473 extend: int | Literal["full"] = 0,
474 ) -> ConjectureResult | _Overrun:
475 """
476 If ``error_on_discard`` is set to True this will raise ``ContainsDiscard``
477 in preference to running the actual test function. This is to allow us
478 to skip test cases we expect to be redundant in some cases. Note that
479 it may be the case that we don't raise ``ContainsDiscard`` even if the
480 result has discards if we cannot determine from previous runs whether
481 it will have a discard.
482 """
483 # node templates and value holes represent a not-yet-filled hole and
484 # therefore cannot be cached or retrieved from the cache.
485 has_value_hole = any(isinstance(choice, ValueHole) for choice in choices)
486 if not has_value_hole and not any(
487 isinstance(choice, ChoiceTemplate) for choice in choices
488 ):
489 # this type cast is validated by the isinstance checks above (ie,
490 # there are no ChoiceTemplate or ValueHole elements).
491 choices = cast(Sequence[ChoiceT], choices)
492 key = self._cache_key(choices)
493 try:
494 cached = self.__data_cache[key]
495 # if we have a cached overrun for this key, but we're allowing extensions
496 # of the nodes, it could in fact run to a valid data if we try.
497 if extend == 0 or cached.status is not Status.OVERRUN:
498 return cached
499 except KeyError:
500 pass
502 if extend == "full":
503 max_length = None
504 elif (count := choice_count(choices)) is None:
505 max_length = None
506 else:
507 max_length = count + extend
509 # explicitly use a no-op DataObserver here instead of a TreeRecordingObserver.
510 # The reason is we don't expect simulate_test_function to explore new choices
511 # and write back to the tree, so we don't want the overhead of the
512 # TreeRecordingObserver tracking those calls.
513 trial_observer: DataObserver | None = DataObserver()
514 if error_on_discard:
515 trial_observer = DiscardObserver()
517 try:
518 # A ValueHole is only resolvable by the strategy drawn at its
519 # position, so tree simulation - which replays at the choice level
520 # with no strategies involved - would resolve it incorrectly.
521 # (ChoiceTemplate is fine: its resolution is deterministic at the
522 # choice level, so simulation and the real run agree.)
523 if has_value_hole:
524 raise PreviouslyUnseenBehaviour
525 trial_data = self.new_conjecture_data(
526 choices, observer=trial_observer, max_choices=max_length
527 )
528 self.tree.simulate_test_function(trial_data)
529 except PreviouslyUnseenBehaviour:
530 pass
531 else:
532 trial_data.freeze()
533 key = self._cache_key(trial_data.choices)
534 if trial_data.status > Status.OVERRUN:
535 try:
536 return self.__data_cache[key]
537 except KeyError:
538 pass
539 else:
540 # if we simulated to an overrun, then we our result is certainly
541 # an overrun; no need to consult the cache. (and we store this result
542 # for simulation-less lookup later).
543 self.__data_cache[key] = Overrun
544 return Overrun
545 try:
546 return self.__data_cache[key]
547 except KeyError:
548 pass
550 data = self.new_conjecture_data(choices, max_choices=max_length)
551 # note that calling test_function caches `data` for us.
552 self.test_function(data)
553 return data.as_result()
555 def test_function(self, data: ConjectureData) -> None:
556 if self.__pending_call_explanation is not None:
557 self.debug(self.__pending_call_explanation)
558 self.__pending_call_explanation = None
560 self.call_count += 1
561 interrupted = False
563 def _backend_cannot_proceed(
564 exc: BackendCannotProceed, data: ConjectureData
565 ) -> None:
566 if exc.scope in ("verified", "exhausted"):
567 self._switch_to_hypothesis_provider = True
568 if exc.scope == "verified":
569 self._verified_by_backend = self.settings.backend
570 elif exc.scope == "discard_test_case":
571 self._backend_discard_count += 1
572 if (
573 self._backend_discard_count > 10
574 and (self._backend_discard_count / self.call_count) > 0.2
575 ):
576 verbose_report(
577 f"Switching away from backend {self.settings.backend!r} "
578 "to the Hypothesis backend, "
579 f"because {self._backend_discard_count} of {self.call_count} "
580 "attempted test cases "
581 f"({self._backend_discard_count / self.call_count * 100:0.1f}%) "
582 f"were discarded by backend {self.settings.backend!r}"
583 )
584 self._switch_to_hypothesis_provider = True
586 # treat all BackendCannotProceed exceptions as invalid. This isn't
587 # great; "verified" should really be counted as self.valid_test_cases += 1.
588 # But we check self.valid_test_cases == 0 to determine whether to raise
589 # Unsatisfiable, and that would throw this check off.
590 self.invalid_test_cases += 1
591 data.cannot_proceed_scope = exc.scope
593 # this fiddly bit of control flow is to work around `return` being
594 # disallowed in `finally` blocks as of python 3.14. Otherwise, we would
595 # just return in the _backend_cannot_proceed branch.
596 finally_early_return = False
598 try:
599 self.__stoppable_test_function(data)
600 except KeyboardInterrupt:
601 interrupted = True
602 raise
603 except BackendCannotProceed as exc:
604 _backend_cannot_proceed(exc, data)
605 # skip the post-test-case tracking; we're pretending this never happened
606 interrupted = True
607 data.freeze()
608 return
609 except BaseException as err:
610 data.freeze()
611 if isinstance(err, FlakyStrategyDefinition) and data._stateful_repr_parts:
612 # In a stateful test, surface the steps leading up to the
613 # inconsistency.
614 report(
615 "Steps leading up to this error:\n"
616 + "\n".join(f" {s}" for s in data._stateful_repr_parts)
617 )
618 if self.settings.backend != "hypothesis":
619 try:
620 realize_choices(data, for_failure=True)
621 except BackendCannotProceed as exc:
622 _backend_cannot_proceed(exc, data)
623 # skip the post-test-case tracking; we're pretending this
624 # never happened
625 interrupted = True
626 return
627 self.save_choices(data.choices)
628 raise
629 finally:
630 # No branch, because if we're interrupted we always raise
631 # the KeyboardInterrupt, never continue to the code below.
632 if not interrupted: # pragma: no branch
633 assert data.cannot_proceed_scope is None
634 data.freeze()
636 if self.settings.backend != "hypothesis":
637 try:
638 realize_choices(
639 data, for_failure=data.status is Status.INTERESTING
640 )
641 except BackendCannotProceed as exc:
642 _backend_cannot_proceed(exc, data)
643 finally_early_return = True
645 if not finally_early_return:
646 call_stats: CallStats = {
647 "status": data.status.name.lower(),
648 "runtime": data.finish_time - data.start_time,
649 "drawtime": math.fsum(data.draw_times.values()),
650 "gctime": data.gc_finish_time - data.gc_start_time,
651 "events": sorted(
652 k if v == "" else f"{k}: {v}"
653 for k, v in data.events.items()
654 ),
655 }
656 self.stats_per_test_case.append(call_stats)
658 self._cache(data)
659 if (
660 data.misaligned_at is not None
661 ): # pragma: no branch # coverage bug?
662 self.misaligned_count += 1
664 if finally_early_return:
665 return
667 self.debug_data(data)
669 if (
670 data.target_observations
671 and self.pareto_front is not None
672 and self.pareto_front.add(data.as_result())
673 ):
674 self.save_choices(data.choices, sub_key=b"pareto")
676 if data.status >= Status.VALID:
677 for k, v in data.target_observations.items():
678 self.best_observed_targets[k] = max(self.best_observed_targets[k], v)
680 if k not in self.best_test_cases_of_observed_targets:
681 data_as_result = data.as_result()
682 assert not isinstance(data_as_result, _Overrun)
683 self.best_test_cases_of_observed_targets[k] = data_as_result
684 continue
686 existing_test_case = self.best_test_cases_of_observed_targets[k]
687 existing_score = existing_test_case.target_observations[k]
689 if v < existing_score:
690 continue
692 if v > existing_score or sort_key(data.nodes) < sort_key(
693 existing_test_case.nodes
694 ):
695 data_as_result = data.as_result()
696 assert not isinstance(data_as_result, _Overrun)
697 self.best_test_cases_of_observed_targets[k] = data_as_result
699 if data.status is Status.VALID:
700 self.valid_test_cases += 1
701 if data.status is Status.INVALID:
702 self.invalid_test_cases += 1
703 if data.status is Status.OVERRUN:
704 self.overrun_test_cases += 1
706 if data.status == Status.INTERESTING:
707 if not self.using_hypothesis_backend:
708 # replay this failure on the hypothesis backend to ensure it still
709 # finds a failure. otherwise, it is flaky.
710 initial_exception = data.expected_exception
711 data = ConjectureData.for_choices(data.choices)
712 # we've already going to use the hypothesis provider for this
713 # data, so the verb "switch" is a bit misleading here. We're really
714 # setting this to inform our on_observation logic that the observation
715 # generated here was from a hypothesis backend, and shouldn't be
716 # sent to the on_observation of any alternative backend.
717 with self._with_switch_to_hypothesis_provider(True):
718 self.__stoppable_test_function(data)
719 data.freeze()
720 # TODO: Should same-origin also be checked? (discussion in
721 # https://github.com/HypothesisWorks/hypothesis/pull/4470#discussion_r2217055487)
722 if data.status != Status.INTERESTING:
723 desc_new_status = {
724 data.status.VALID: "passed",
725 data.status.INVALID: "failed filters",
726 data.status.OVERRUN: "overran",
727 }[data.status]
728 raise FlakyBackendFailure(
729 f"Inconsistent results from replaying a failing test case! "
730 f"Raised {type(initial_exception).__name__} on "
731 f"backend={self.settings.backend!r}, but "
732 f"{desc_new_status} under backend='hypothesis'.",
733 [initial_exception],
734 )
736 self._cache(data)
738 assert data.interesting_origin is not None
739 key = data.interesting_origin
740 changed = False
741 try:
742 existing = self.interesting_test_cases[key]
743 except KeyError:
744 changed = True
745 self.last_bug_found_at = self.call_count
746 if self.first_bug_found_at is None:
747 self.first_bug_found_at = self.call_count
748 self.first_bug_found_time = time.monotonic()
749 else:
750 if sort_key(data.nodes) < sort_key(existing.nodes):
751 self.shrinks += 1
752 self.downgrade_choices(existing.choices)
753 self.__data_cache.unpin(self._cache_key(existing.choices))
754 changed = True
756 if changed:
757 self.save_choices(data.choices)
758 self.interesting_test_cases[key] = data.as_result() # type: ignore
759 if not self.using_hypothesis_backend:
760 self._backend_found_failure = True
761 self.__data_cache.pin(self._cache_key(data.choices), data.as_result())
762 self.shrunk_test_cases.discard(key)
764 if self.shrinks >= MAX_SHRINKS:
765 self.exit_with(ExitReason.max_shrinks)
767 if (
768 not self.ignore_limits
769 and self.finish_shrinking_deadline is not None
770 and self.finish_shrinking_deadline < time.perf_counter()
771 ):
772 # See https://github.com/HypothesisWorks/hypothesis/issues/2340
773 report(
774 "WARNING: Hypothesis has spent more than five minutes working to shrink"
775 " a failing test case, and stopped because it is making very slow"
776 " progress. When you re-run your tests, shrinking will resume and may"
777 " take this long before aborting again.\n\nPLEASE REPORT THIS if you can"
778 " provide a reproducing example, so that we can improve shrinking"
779 " performance for everyone."
780 )
781 self.exit_with(ExitReason.very_slow_shrinking)
783 if not self.interesting_test_cases:
784 # Note that this logic is reproduced to end the generation phase when
785 # we have interesting test cases. Update that too if you change this!
786 # (The doubled implementation is because here we exit the engine entirely,
787 # while in the other case below we just want to move on to shrinking.)
788 if self.valid_test_cases >= self.settings.max_examples:
789 self.exit_with(ExitReason.max_examples)
790 if (self.invalid_test_cases + self.overrun_test_cases) > (
791 INVALID_THRESHOLD_BASE + INVALID_PER_VALID * self.valid_test_cases
792 ):
793 self.exit_with(ExitReason.max_iterations)
795 # With interesting test cases that made at least one choice,
796 # ``should_generate_more`` instead ends the generation phase on an
797 # exhausted tree, so that we still shrink (and explain) them.
798 if self.__tree_is_exhausted() and not any(
799 tc.choices for tc in self.interesting_test_cases.values()
800 ):
801 self.exit_with(ExitReason.finished)
803 self.record_for_health_check(data)
805 def on_pareto_evict(self, data: ConjectureResult) -> None:
806 db = self.settings.database
807 assert db is not None
809 db.delete(self.pareto_key, choices_to_bytes(data.choices))
811 def generate_novel_prefix(self) -> tuple[ChoiceT, ...]:
812 """Uses the tree to proactively generate a starting choice sequence
813 that we haven't explored yet for this test.
815 When this method is called, we assume that there must be at
816 least one novel prefix left to find. If there were not, then the
817 test run should have already stopped due to tree exhaustion.
818 """
819 return self.tree.generate_novel_prefix(self.random)
821 def record_for_health_check(self, data: ConjectureData) -> None:
822 # Once we've actually found a bug, there's no point in trying to run
823 # health checks - they'll just mask the actually important information.
824 if data.status == Status.INTERESTING:
825 self.health_check_state = None
827 state = self.health_check_state
829 if state is None:
830 return
832 for k, v in data.draw_times.items():
833 state.draw_times[k].append(v)
835 if data.status == Status.VALID:
836 state.valid_test_cases += 1
837 elif data.status == Status.INVALID:
838 state.invalid_test_cases += 1
839 else:
840 assert data.status == Status.OVERRUN
841 state.overrun_test_cases += 1
843 max_valid_draws = 10
844 max_invalid_draws = 50
845 max_overrun_draws = 20
847 assert state.valid_test_cases <= max_valid_draws
849 if state.valid_test_cases == max_valid_draws:
850 self.health_check_state = None
851 return
853 if state.overrun_test_cases == max_overrun_draws:
854 fail_health_check(
855 self.settings,
856 "Generated inputs routinely consumed more than the maximum "
857 f"allowed entropy: {state.valid_test_cases} inputs were generated "
858 f"successfully, while {state.overrun_test_cases} inputs exceeded the "
859 f"maximum allowed entropy during generation."
860 "\n\n"
861 f"Testing with inputs this large tends to be slow, and to produce "
862 "failures that are both difficult to shrink and difficult to understand. "
863 "Try decreasing the amount of data generated, for example by "
864 "decreasing the minimum size of collection strategies like "
865 "st.lists()."
866 "\n\n"
867 "If you expect the average size of your input to be this large, "
868 "you can disable this health check with "
869 "@settings(suppress_health_check=[HealthCheck.data_too_large]). "
870 "See "
871 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
872 "for details.",
873 HealthCheck.data_too_large,
874 )
875 if state.invalid_test_cases == max_invalid_draws:
876 fail_health_check(
877 self.settings,
878 "It looks like this test is filtering out a lot of inputs. "
879 f"{state.valid_test_cases} inputs were generated successfully, "
880 f"while {state.invalid_test_cases} inputs were filtered out. "
881 "\n\n"
882 "An input might be filtered out by calls to assume(), "
883 "strategy.filter(...), or occasionally by Hypothesis internals."
884 "\n\n"
885 "Applying this much filtering makes input generation slow, since "
886 "Hypothesis must discard inputs which are filtered out and try "
887 "generating it again. It is also possible that applying this much "
888 "filtering will distort the domain and/or distribution of the test, "
889 "leaving your testing less rigorous than expected."
890 "\n\n"
891 "If you expect this many inputs to be filtered out during generation, "
892 "you can disable this health check with "
893 "@settings(suppress_health_check=[HealthCheck.filter_too_much]). See "
894 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
895 "for details.",
896 HealthCheck.filter_too_much,
897 )
899 # Allow at least the greater of one second or 5x the deadline. If deadline
900 # is None, allow 30s - the user can disable the healthcheck too if desired.
901 draw_time = state.total_draw_time
902 draw_time_limit = 5 * (self.settings.deadline or timedelta(seconds=6))
903 if (
904 draw_time > max(1.0, draw_time_limit.total_seconds())
905 # we disable HealthCheck.too_slow under concurrent threads, since
906 # cpython may switch away from a thread for arbitrarily long.
907 and not self.thread_overlap.get(threading.get_ident(), False)
908 ):
909 extra_str = []
910 if state.invalid_test_cases:
911 extra_str.append(f"{state.invalid_test_cases} invalid inputs")
912 if state.overrun_test_cases:
913 extra_str.append(
914 f"{state.overrun_test_cases} inputs which exceeded the "
915 "maximum allowed entropy"
916 )
917 extra_str = ", and ".join(extra_str)
918 extra_str = f" ({extra_str})" if extra_str else ""
920 fail_health_check(
921 self.settings,
922 "Input generation is slow: Hypothesis only generated "
923 f"{state.valid_test_cases} valid inputs after {draw_time:.2f} "
924 f"seconds{extra_str}."
925 "\n" + state.timing_report() + "\n\n"
926 "This could be for a few reasons:"
927 "\n"
928 "1. This strategy could be generating too much data per input. "
929 "Try decreasing the amount of data generated, for example by "
930 "decreasing the minimum size of collection strategies like "
931 "st.lists()."
932 "\n"
933 "2. Some other expensive computation could be running during input "
934 "generation. For example, "
935 "if @st.composite or st.data() is interspersed with an expensive "
936 "computation, HealthCheck.too_slow is likely to trigger. If this "
937 "computation is unrelated to input generation, move it elsewhere. "
938 "Otherwise, try making it more efficient, or disable this health "
939 "check if that is not possible."
940 "\n\n"
941 "If you expect input generation to take this long, you can disable "
942 "this health check with "
943 "@settings(suppress_health_check=[HealthCheck.too_slow]). See "
944 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
945 "for details.",
946 HealthCheck.too_slow,
947 )
949 def save_choices(
950 self, choices: Sequence[ChoiceT], sub_key: bytes | None = None
951 ) -> None:
952 if self.database is not None:
953 self.database.save(self.sub_key(sub_key), choices_to_bytes(choices))
955 def downgrade_choices(self, choices: Sequence[ChoiceT]) -> None:
956 buffer = choices_to_bytes(choices)
957 if self.settings.database is not None and self.database_key is not None:
958 self.settings.database.move(self.database_key, self.secondary_key, buffer)
960 def sub_key(self, sub_key: bytes | None) -> bytes:
961 assert self.database_key is not None
962 if sub_key is None:
963 return self.database_key
964 return b".".join((self.database_key, sub_key))
966 @property
967 def secondary_key(self) -> bytes:
968 return self.sub_key(b"secondary")
970 @property
971 def pareto_key(self) -> bytes:
972 return self.sub_key(b"pareto")
974 def debug(self, message: str) -> None:
975 if self.settings.verbosity >= Verbosity.debug:
976 base_report(message)
978 @property
979 def report_debug_info(self) -> bool:
980 return self.settings.verbosity >= Verbosity.debug
982 def debug_data(self, data: ConjectureData | ConjectureResult) -> None:
983 if not self.report_debug_info:
984 return
986 status = repr(data.status)
987 if data.status == Status.INTERESTING:
988 status = f"{status} ({data.interesting_origin!r})"
989 elif data.status == Status.INVALID and isinstance(data, ConjectureData):
990 assert isinstance(data, ConjectureData) # mypy is silly
991 status = f"{status} ({data.events.get('gave up because', '?')})"
993 self.debug(f"{len(data.choices)} choices -> {status}\n\t{data.choices}")
995 def observe_for_provider(self) -> AbstractContextManager:
996 def on_observation(observation: Observation) -> None:
997 assert observation.type == "test_case"
998 # because lifetime == "test_function"
999 assert isinstance(self.provider, PrimitiveProvider)
1000 # only fire if we actually used that provider to generate this observation
1001 if not self._switch_to_hypothesis_provider:
1002 self.provider.on_observation(observation)
1004 if (
1005 self.settings.backend != "hypothesis"
1006 # only for lifetime = "test_function" providers (guaranteed
1007 # by this isinstance check)
1008 and isinstance(self.provider, PrimitiveProvider)
1009 # and the provider opted-in to observations
1010 and self.provider.add_observability_callback
1011 ):
1012 return with_observability_callback(on_observation)
1013 return nullcontext()
1015 def run(self) -> None:
1016 with local_settings(self.settings), self.observe_for_provider():
1017 try:
1018 self._run()
1019 except RunIsComplete:
1020 pass
1021 for v in self.interesting_test_cases.values():
1022 self.debug_data(v)
1023 self.debug(
1024 f"Run complete after {self.call_count} test cases "
1025 f"({self.valid_test_cases} valid) and {self.shrinks} shrinks"
1026 )
1028 @property
1029 def database(self) -> ExampleDatabase | None:
1030 if self.database_key is None:
1031 return None
1032 return self.settings.database
1034 def has_existing_test_cases(self) -> bool:
1035 return self.database is not None and Phase.reuse in self.settings.phases
1037 def reuse_existing_test_cases(self) -> None:
1038 """If appropriate (we have a database and have been told to use it),
1039 try to reload existing test cases from the database.
1041 If there are a lot we don't try all of them. We always try the
1042 smallest test case in the database (which is guaranteed to be the
1043 last failure) and the largest (which is usually the seed test case
1044 which the last failure came from but we don't enforce that). We
1045 then take a random sampling of the remainder and try those. Any
1046 test cases that are no longer interesting are cleared out.
1047 """
1048 if self.has_existing_test_cases():
1049 self.debug("Reusing test cases from database")
1050 # We have to do some careful juggling here. We have two database
1051 # corpora: The primary and secondary. The primary corpus is a
1052 # small set of minimized test cases each of which has at one point
1053 # demonstrated a distinct bug. We want to retry all of these.
1055 # We also have a secondary corpus of test cases that have at some
1056 # point demonstrated interestingness (currently only ones that
1057 # were previously non-minimal test cases for a bug, but this will
1058 # likely expand in future). These are a good source of potentially
1059 # interesting test cases, but there are a lot of them, so we down
1060 # sample the secondary corpus to a more manageable size.
1062 db = self.settings.database
1063 db_key = self.database_key
1064 assert db is not None
1065 assert db_key is not None
1067 corpus = sorted(db.fetch(db_key), key=shortlex)
1068 factor = 0.1 if (Phase.generate in self.settings.phases) else 1
1069 desired_size = max(2, ceil(factor * self.settings.max_examples))
1070 primary_corpus_size = len(corpus)
1072 if len(corpus) < desired_size:
1073 extra_corpus = list(db.fetch(self.secondary_key))
1075 shortfall = desired_size - len(corpus)
1077 if len(extra_corpus) <= shortfall:
1078 extra = extra_corpus
1079 else:
1080 extra = self.random.sample(extra_corpus, shortfall)
1081 extra.sort(key=shortlex)
1082 corpus.extend(extra)
1084 # We want a fast path where every primary entry in the database was
1085 # interesting.
1086 found_interesting_in_primary = False
1087 all_interesting_in_primary_were_exact = True
1089 for i, existing in enumerate(corpus):
1090 if i >= primary_corpus_size and found_interesting_in_primary:
1091 break
1092 choices = choices_from_bytes(existing)
1093 if choices is None:
1094 # clear out any keys which fail deserialization
1095 db.delete(db_key, existing)
1096 continue
1097 data = self.cached_test_function(choices, extend="full")
1098 if data.status != Status.INTERESTING:
1099 db.delete(db_key, existing)
1100 db.delete(self.secondary_key, existing)
1101 else:
1102 if i < primary_corpus_size:
1103 found_interesting_in_primary = True
1104 assert not isinstance(data, _Overrun)
1105 if choices_key(choices) != choices_key(data.choices):
1106 all_interesting_in_primary_were_exact = False
1107 if not self.settings.report_multiple_bugs:
1108 break
1109 if found_interesting_in_primary:
1110 if all_interesting_in_primary_were_exact:
1111 self.reused_previously_shrunk_test_case = True
1113 # Because self.database is not None (because self.has_existing_test_cases())
1114 # and self.database_key is not None (because we fetched using it above),
1115 # we can guarantee self.pareto_front is not None
1116 assert self.pareto_front is not None
1118 # If we've not found any interesting test cases so far we try some of
1119 # the pareto front from the last run.
1120 if len(corpus) < desired_size and not self.interesting_test_cases:
1121 desired_extra = desired_size - len(corpus)
1122 pareto_corpus = list(db.fetch(self.pareto_key))
1123 if len(pareto_corpus) > desired_extra:
1124 pareto_corpus = self.random.sample(pareto_corpus, desired_extra)
1125 pareto_corpus.sort(key=shortlex)
1127 for existing in pareto_corpus:
1128 choices = choices_from_bytes(existing)
1129 if choices is None:
1130 db.delete(self.pareto_key, existing)
1131 continue
1132 data = self.cached_test_function(choices, extend="full")
1133 if data not in self.pareto_front:
1134 db.delete(self.pareto_key, existing)
1135 if data.status == Status.INTERESTING:
1136 break
1138 def exit_with(self, reason: ExitReason) -> None:
1139 if self.ignore_limits:
1140 return
1141 self.statistics["stopped-because"] = reason.describe(self.settings)
1142 if self.best_observed_targets:
1143 self.statistics["targets"] = dict(self.best_observed_targets)
1144 self.debug(f"exit_with({reason.name})")
1145 self.exit_reason = reason
1146 raise RunIsComplete
1148 def should_generate_more(self) -> bool:
1149 # End the generation phase where we would have ended it if no bugs had
1150 # been found. This reproduces the exit logic in `self.test_function`,
1151 # but with the important distinction that this clause will move on to
1152 # the shrinking phase having found one or more bugs, while the other
1153 # will exit having found zero bugs.
1154 invalid_threshold = (
1155 INVALID_THRESHOLD_BASE + INVALID_PER_VALID * self.valid_test_cases
1156 )
1157 if (
1158 self.valid_test_cases >= self.settings.max_examples
1159 or (self.invalid_test_cases + self.overrun_test_cases) > invalid_threshold
1160 or self.__tree_is_exhausted()
1161 ):
1162 return False
1164 # If we haven't found a bug, keep looking - if we hit any limits on
1165 # the number of tests to run that will raise an exception and stop
1166 # the run.
1167 if not self.interesting_test_cases:
1168 return True
1169 # Users who disable shrinking probably want to exit as fast as possible.
1170 # If we've found a bug and won't report more than one, stop looking.
1171 # If we first saw a bug more than 10 seconds ago, stop looking.
1172 elif (
1173 Phase.shrink not in self.settings.phases
1174 or not self.settings.report_multiple_bugs
1175 or time.monotonic() - self.first_bug_found_time > 10
1176 ):
1177 return False
1178 assert isinstance(self.first_bug_found_at, int)
1179 assert isinstance(self.last_bug_found_at, int)
1180 assert self.first_bug_found_at <= self.last_bug_found_at <= self.call_count
1181 # Otherwise, keep searching for between ten and 'a heuristic' calls.
1182 # We cap 'calls after first bug' so errors are reported reasonably
1183 # soon even for tests that are allowed to run for a very long time,
1184 # or sooner if the latest half of our test effort has been fruitless.
1185 return self.call_count < MIN_TEST_CALLS or self.call_count < min(
1186 self.first_bug_found_at + 1000, self.last_bug_found_at * 2
1187 )
1189 def generate_new_test_cases(self) -> None:
1190 if Phase.generate not in self.settings.phases:
1191 return
1192 if self.interesting_test_cases:
1193 # The example database has failing test cases from a previous run,
1194 # so we'd rather report that they're still failing ASAP than take
1195 # the time to look for additional failures.
1196 return
1198 self.debug("Generating new test cases")
1200 assert self.should_generate_more()
1201 self._switch_to_hypothesis_provider = True
1202 zero_data = self.cached_test_function((ChoiceTemplate("simplest", count=None),))
1203 if zero_data.status > Status.OVERRUN:
1204 assert isinstance(zero_data, ConjectureResult)
1205 # if the crosshair backend cannot proceed, it does not (and cannot)
1206 # realize the symbolic values, with the intent that Hypothesis will
1207 # throw away this test case. We usually do, but if it's the zero data
1208 # then we try to pin it here, which requires realizing the symbolics.
1209 #
1210 # We don't (yet) rely on the zero data being pinned, and so
1211 # it's simply a very slight performance loss to simply not pin it
1212 # if doing so would error.
1213 if zero_data.cannot_proceed_scope is None: # pragma: no branch
1214 self.__data_cache.pin(
1215 self._cache_key(zero_data.choices), zero_data.as_result()
1216 ) # Pin forever
1218 if zero_data.status == Status.OVERRUN or (
1219 zero_data.status == Status.VALID
1220 and isinstance(zero_data, ConjectureResult)
1221 and zero_data.length * 2 > BUFFER_SIZE
1222 ):
1223 fail_health_check(
1224 self.settings,
1225 "The smallest natural input for this test is very "
1226 "large. This makes it difficult for Hypothesis to generate "
1227 "good inputs, especially when trying to shrink failing inputs."
1228 "\n\n"
1229 "Consider reducing the amount of data generated by the strategy. "
1230 "Also consider introducing small alternative values for some "
1231 "strategies. For example, could you "
1232 "mark some arguments as optional by replacing `some_complex_strategy`"
1233 "with `st.none() | some_complex_strategy`?"
1234 "\n\n"
1235 "If you are confident that the size of the smallest natural input "
1236 "to your test cannot be reduced, you can suppress this health check "
1237 "with @settings(suppress_health_check=[HealthCheck.large_base_example]). "
1238 "See "
1239 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
1240 "for details.",
1241 HealthCheck.large_base_example,
1242 )
1244 self.health_check_state = HealthCheckState()
1246 # We attempt to use the size of the minimal generated test case starting
1247 # from a given novel prefix as a guideline to generate smaller test
1248 # cases for an initial period, by restriscting ourselves to test cases
1249 # that are not much larger than it.
1250 #
1251 # Calculating the actual minimal generated test case is hard, so we
1252 # take a best guess that zero extending a prefix produces the minimal
1253 # test case starting with that prefix (this is true for our built in
1254 # strategies). This is only a reasonable thing to do if the resulting
1255 # test case is valid. If we regularly run into situations where it is
1256 # not valid then this strategy is a waste of time, so we want to
1257 # abandon it early. In order to do this we track how many times in a
1258 # row it has failed to work, and abort small test case generation when
1259 # it has failed too many times in a row.
1260 consecutive_zero_extend_is_invalid = 0
1262 # We control growth during initial test case generation, for two
1263 # reasons:
1264 #
1265 # * It gives us an opportunity to find small test cases early, which
1266 # gives us a fast path for easy to find bugs.
1267 # * It avoids low probability events where we might end up
1268 # generating very large test cases during health checks, which
1269 # on slower machines can trigger HealthCheck.too_slow.
1270 #
1271 # The heuristic we use is that we attempt to estimate the smallest
1272 # extension of this prefix, and limit the size to no more than
1273 # an order of magnitude larger than that. If we fail to estimate
1274 # the size accurately, we skip over this prefix and try again.
1275 #
1276 # We need to tune the test case size based on the initial prefix,
1277 # because any fixed size might be too small, and any size based
1278 # on the strategy in general can fall afoul of strategies that
1279 # have very different sizes for different prefixes.
1280 #
1281 # We previously set a minimum value of 10 on small_test_case_cap, with the
1282 # reasoning of avoiding flaky health checks. However, some users set a
1283 # low max_examples for performance. A hard lower bound in this case biases
1284 # the distribution towards small (and less powerful) test cases. Flaky
1285 # and loud health checks are better than silent performance degradation.
1286 small_test_case_cap = min(self.settings.max_examples // 10, 50)
1287 # For small budgets we run a single optimisation pass halfway through
1288 # the budget, which experience shows is roughly optimal. For larger
1289 # budgets we instead start optimising early and interleave repeated
1290 # passes with generation - see _should_optimise_now.
1291 if self.settings.max_examples < 1000:
1292 self._next_optimise_at = max(
1293 self.settings.max_examples // 2, small_test_case_cap + 1, 10
1294 )
1295 else:
1296 self._next_optimise_at = max(200, small_test_case_cap + 1)
1297 self._switch_to_hypothesis_provider = False
1299 while self.should_generate_more():
1300 # we don't yet integrate DataTree with backends. Instead of generating
1301 # a novel prefix, ask the backend for an input.
1302 if not self.using_hypothesis_backend:
1303 data = self.new_conjecture_data([])
1304 self.test_function(data)
1305 continue
1307 self._current_phase = "generate"
1308 prefix = self.generate_novel_prefix()
1309 if (
1310 self.valid_test_cases <= small_test_case_cap
1311 and self.call_count <= 5 * small_test_case_cap
1312 and not self.interesting_test_cases
1313 and consecutive_zero_extend_is_invalid < 5
1314 ):
1315 minimal_test_case = self.cached_test_function(
1316 prefix + (ChoiceTemplate("simplest", count=None),)
1317 )
1319 if minimal_test_case.status < Status.VALID:
1320 consecutive_zero_extend_is_invalid += 1
1321 continue
1322 # Because the Status code is greater than Status.VALID, it cannot be
1323 # Status.OVERRUN, which guarantees that the minimal_test_case is a
1324 # ConjectureResult object.
1325 assert isinstance(minimal_test_case, ConjectureResult)
1326 consecutive_zero_extend_is_invalid = 0
1327 minimal_extension = len(minimal_test_case.choices) - len(prefix)
1328 max_length = len(prefix) + minimal_extension * 5
1330 # We could end up in a situation where even though the prefix was
1331 # novel when we generated it, because we've now tried zero extending
1332 # it not all possible continuations of it will be novel. In order to
1333 # avoid making redundant test calls, we rerun it in simulation mode
1334 # first. If this has a predictable result, then we don't bother
1335 # running the test function for real here. If however we encounter
1336 # some novel behaviour, we try again with the real test function,
1337 # starting from the new novel prefix that has discovered.
1338 trial_data = self.new_conjecture_data(prefix, max_choices=max_length)
1339 try:
1340 self.tree.simulate_test_function(trial_data)
1341 continue
1342 except PreviouslyUnseenBehaviour:
1343 pass
1345 # If the simulation entered part of the tree that has been killed,
1346 # we don't want to run this.
1347 assert isinstance(trial_data.observer, TreeRecordingObserver)
1348 if trial_data.observer.killed:
1349 continue
1351 # We might have hit the cap on number of test cases we should
1352 # run when calculating the minimal test case.
1353 if not self.should_generate_more():
1354 break
1356 prefix = trial_data.choices
1357 else:
1358 max_length = None
1360 data = self.new_conjecture_data(prefix, max_choices=max_length)
1361 self.test_function(data)
1363 if (
1364 data.status is Status.OVERRUN
1365 and max_length is not None
1366 and "gave up because" not in data.events
1367 ):
1368 data.events["gave up because"] = (
1369 "reduced max size for early test cases (avoids flaky health checks)"
1370 )
1372 self.generate_mutations_from(data)
1374 # Although the optimisations are logically a distinct phase, we
1375 # actually normally run them as part of test case generation. The
1376 # reason for this is that we cannot guarantee that optimisation
1377 # actually exhausts our budget: It might finish running and we
1378 # discover that actually we still could run a bunch more test cases
1379 # if we want.
1380 if self._should_optimise_now():
1381 self._run_optimise_pass()
1383 def generate_mutations_from(self, data: ConjectureData | ConjectureResult) -> None:
1384 # A thing that is often useful but rarely happens by accident is
1385 # to generate the same value at multiple different points in the
1386 # test case.
1387 #
1388 # Rather than make this the responsibility of individual strategies
1389 # we implement a small mutator that just takes parts of the test
1390 # case with the same label and tries replacing one of them with a
1391 # copy of the other and tries running it. If we've made a good
1392 # guess about what to put where, this will run a similar generated
1393 # test case with more duplication.
1394 if (
1395 # An OVERRUN doesn't have enough information about the test
1396 # case to mutate, so we just skip those.
1397 data.status >= Status.INVALID
1398 # This has a tendency to trigger some weird edge cases during
1399 # generation so we don't let it run until we're done with the
1400 # health checks.
1401 and self.health_check_state is None
1402 ):
1403 initial_calls = self.call_count
1404 failed_mutations = 0
1406 while (
1407 self.should_generate_more()
1408 # We implement fairly conservative checks for how long we
1409 # we should run mutation for, as it's generally not obvious
1410 # how helpful it is for any given test case.
1411 and self.call_count <= initial_calls + 5
1412 and failed_mutations <= 5
1413 ):
1414 groups = data.spans.mutator_groups
1415 if not groups:
1416 break
1418 group = self.random.choice(groups)
1419 (start1, end1), (start2, end2) = self.random.sample(sorted(group), 2)
1420 if start1 > start2:
1421 (start1, end1), (start2, end2) = (start2, end2), (start1, end1)
1423 if (
1424 start1 <= start2 <= end2 <= end1
1425 ): # pragma: no cover # flaky on conjecture-cover tests
1426 # One span entirely contains the other. The strategy is very
1427 # likely some kind of tree. e.g. we might have
1428 #
1429 # ┌─────┐
1430 # ┌─────┤ a ├──────┐
1431 # │ └─────┘ │
1432 # ┌──┴──┐ ┌──┴──┐
1433 # ┌──┤ b ├──┐ ┌──┤ c ├──┐
1434 # │ └──┬──┘ │ │ └──┬──┘ │
1435 # ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐
1436 # │ d │ │ e │ │ f │ │ g │ │ h │ │ i │
1437 # └───┘ └───┘ └───┘ └───┘ └───┘ └───┘
1438 #
1439 # where each node is drawn from the same strategy and so
1440 # has the same span label. We might have selected the spans
1441 # corresponding to the a and c nodes, which is the entire
1442 # tree and the subtree of (and including) c respectively.
1443 #
1444 # There are two possible mutations we could apply in this case:
1445 # 1. replace a with c (replace child with parent)
1446 # 2. replace c with a (replace parent with child)
1447 #
1448 # (1) results in multiple partial copies of the
1449 # parent:
1450 # ┌─────┐
1451 # ┌─────┤ a ├────────────┐
1452 # │ └─────┘ │
1453 # ┌──┴──┐ ┌─┴───┐
1454 # ┌──┤ b ├──┐ ┌─────┤ a ├──────┐
1455 # │ └──┬──┘ │ │ └─────┘ │
1456 # ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌──┴──┐ ┌──┴──┐
1457 # │ d │ │ e │ │ f │ ┌──┤ b ├──┐ ┌──┤ c ├──┐
1458 # └───┘ └───┘ └───┘ │ └──┬──┘ │ │ └──┬──┘ │
1459 # ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐
1460 # │ d │ │ e │ │ f │ │ g │ │ h │ │ i │
1461 # └───┘ └───┘ └───┘ └───┘ └───┘ └───┘
1462 #
1463 # While (2) results in truncating part of the parent:
1464 #
1465 # ┌─────┐
1466 # ┌──┤ c ├──┐
1467 # │ └──┬──┘ │
1468 # ┌─┴─┐ ┌─┴─┐ ┌─┴─┐
1469 # │ g │ │ h │ │ i │
1470 # └───┘ └───┘ └───┘
1471 #
1472 # (1) is the same as Example IV.4. in Nautilus (NDSS '19)
1473 # (https://wcventure.github.io/FuzzingPaper/Paper/NDSS19_Nautilus.pdf),
1474 # except we do not repeat the replacement additional times
1475 # (the paper repeats it once for a total of two copies).
1476 #
1477 # We currently only apply mutation (1), and ignore mutation
1478 # (2). The reason is that the attempt generated from (2) is
1479 # always something that Hypothesis could easily have generated
1480 # itself, by simply not making various choices. Whereas
1481 # duplicating the exact value + structure of particular choices
1482 # in (1) would have been hard for Hypothesis to generate by
1483 # chance.
1484 #
1485 # TODO: an extension of this mutation might repeat (1) on
1486 # a geometric distribution between 0 and ~10 times. We would
1487 # need to find the corresponding span to recurse on in the new
1488 # choices, probably just by using the choices index.
1490 # case (1): duplicate the choices in start1:start2.
1491 attempt = data.choices[:start2] + data.choices[start1:]
1492 else:
1493 start, end = self.random.choice([(start1, end1), (start2, end2)])
1494 replacement = data.choices[start:end]
1495 # We attempt to replace both the spans with
1496 # whichever choice we made. Note that this might end
1497 # up messing up and getting the span boundaries
1498 # wrong - labels matching are only a best guess as to
1499 # whether the two are equivalent - but it doesn't
1500 # really matter. It may not achieve the desired result,
1501 # but it's still a perfectly acceptable choice sequence
1502 # to try.
1503 attempt = (
1504 data.choices[:start1]
1505 + replacement
1506 + data.choices[end1:start2]
1507 + replacement
1508 + data.choices[end2:]
1509 )
1511 try:
1512 new_data = self.cached_test_function(
1513 attempt,
1514 # We set error_on_discard so that we don't end up
1515 # entering parts of the tree we consider redundant
1516 # and not worth exploring.
1517 error_on_discard=True,
1518 )
1519 except ContainsDiscard:
1520 failed_mutations += 1
1521 continue
1523 if new_data is Overrun:
1524 failed_mutations += 1 # pragma: no cover # annoying case
1525 else:
1526 assert isinstance(new_data, ConjectureResult)
1527 if (
1528 new_data.status >= data.status
1529 and choices_key(data.choices) != choices_key(new_data.choices)
1530 and all(
1531 k in new_data.target_observations
1532 and new_data.target_observations[k] >= v
1533 for k, v in data.target_observations.items()
1534 )
1535 ):
1536 data = new_data
1537 failed_mutations = 0
1538 else:
1539 failed_mutations += 1
1541 def _should_optimise_now(self) -> bool:
1542 """Decide whether to run an optimisation pass at this point in generation."""
1543 if not self.should_optimise:
1544 return False
1545 if self.valid_test_cases >= self._next_optimise_at:
1546 return True
1547 # After the first pass, we should optimize if we have found a new best score from
1548 # generation and we have the budget to continue optimizing.
1549 return (
1550 self._best_scores_at_last_pass is not None
1551 and self.settings.max_examples >= 1000
1552 and self._target_valid_spent < self.settings.max_examples // 2
1553 and any(
1554 score > self._best_scores_at_last_pass.get(target, NO_SCORE)
1555 for target, score in self.best_observed_targets.items()
1556 )
1557 )
1559 def _run_optimise_pass(self) -> None:
1560 """Run one optimisation pass, then schedule the next one."""
1561 if self.settings.max_examples < 1000:
1562 # We'll only ever run a single optimization pass in this case.
1563 max_valid = None
1564 else:
1565 remaining = self.settings.max_examples // 2 - self._target_valid_spent
1566 max_valid = self.valid_test_cases + min(max(200, remaining // 4), remaining)
1567 self._current_phase = "target"
1568 start_valid = self.valid_test_cases
1569 improved = False
1570 try:
1571 improved = self.optimise_targets(max_valid=max_valid)
1572 finally:
1573 spent = self.valid_test_cases - start_valid
1574 self._target_valid_spent += spent
1575 # Alternate with generation on a fair-share schedule for as long
1576 # as passes keep finding improvements and budget remains.
1577 if (
1578 max_valid is not None
1579 and improved
1580 and self._target_valid_spent < self.settings.max_examples // 2
1581 ):
1582 self._next_optimise_at = self.valid_test_cases + spent
1583 else:
1584 self._next_optimise_at = math.inf
1585 self._best_scores_at_last_pass = dict(self.best_observed_targets)
1586 self._current_phase = "generate"
1588 def optimise_targets(self, *, max_valid: int | None = None) -> bool:
1589 """If any target observations have been made, attempt to optimise them
1590 all, stopping early if ``self.valid_test_cases`` reaches ``max_valid``.
1591 Returns whether this made any improvements."""
1592 if not self.should_optimise:
1593 return False
1594 from hypothesis.internal.conjecture.optimiser import Optimiser
1596 # We want to avoid running the optimiser for too long in case we hit
1597 # an unbounded target score. We start this off fairly conservatively
1598 # in case interesting test cases are easy to find and then ramp it up
1599 # on an exponential schedule so we don't hamper the optimiser too much
1600 # if it needs a long time to find good enough improvements.
1601 improved = False
1602 max_improvements = 10
1603 while True:
1604 prev_calls = self.call_count
1606 any_improvements = False
1608 for target, data in list(self.best_test_cases_of_observed_targets.items()):
1609 if max_valid is not None and self.valid_test_cases >= max_valid:
1610 return improved
1611 optimiser = Optimiser(
1612 self, data, target, max_improvements=max_improvements
1613 )
1614 optimiser.run()
1615 if optimiser.improvements > 0:
1616 any_improvements = True
1617 improved = True
1619 if self.interesting_test_cases:
1620 break
1622 max_improvements *= 2
1624 if any_improvements:
1625 continue
1627 if self.best_observed_targets:
1628 self.pareto_optimise()
1630 if prev_calls == self.call_count:
1631 break
1632 return improved
1634 def pareto_optimise(self) -> None:
1635 if self.pareto_front is not None:
1636 ParetoOptimiser(self).run()
1638 def _run(self) -> None:
1639 # have to use the primitive provider to interpret database bits...
1640 self._switch_to_hypothesis_provider = True
1641 with self._log_phase_statistics("reuse"):
1642 self.reuse_existing_test_cases()
1643 # Fast path for development: If the database gave us interesting
1644 # test cases from the previously stored primary key, don't try
1645 # shrinking it again as it's unlikely to work.
1646 if self.reused_previously_shrunk_test_case:
1647 self.exit_with(ExitReason.finished)
1648 # ...but we should use the supplied provider when generating...
1649 self._switch_to_hypothesis_provider = False
1650 with self._log_phase_statistics("generate"):
1651 self.generate_new_test_cases()
1652 # We normally run the target phase mixed in with the generate phase,
1653 # but if we've been asked to run it but not generation then we have to
1654 # run it explicitly on its own here.
1655 if Phase.generate not in self.settings.phases:
1656 self._current_phase = "target"
1657 self.optimise_targets()
1658 # ...and back to the primitive provider when shrinking.
1659 self._switch_to_hypothesis_provider = True
1660 with self._log_phase_statistics("shrink"):
1661 self.shrink_interesting_test_cases()
1662 self.exit_with(ExitReason.finished)
1664 def new_conjecture_data(
1665 self,
1666 prefix: Sequence[ChoiceT | ChoiceTemplate | ValueHole],
1667 *,
1668 observer: DataObserver | None = None,
1669 max_choices: int | None = None,
1670 ) -> ConjectureData:
1671 provider = (
1672 HypothesisProvider if self._switch_to_hypothesis_provider else self.provider
1673 )
1674 observer = observer or self.tree.new_observer()
1675 if not self.using_hypothesis_backend:
1676 observer = DataObserver()
1678 return ConjectureData(
1679 prefix=prefix,
1680 observer=observer,
1681 provider=provider,
1682 max_choices=max_choices,
1683 random=self.random,
1684 )
1686 def shrink_interesting_test_cases(self) -> None:
1687 """If we've found interesting test cases, try to replace each of them
1688 with a minimal interesting test case with the same interesting_origin.
1690 We may find one or more test cases with a new interesting_origin
1691 during the shrink process. If so we shrink these too.
1692 """
1693 if Phase.shrink not in self.settings.phases or not self.interesting_test_cases:
1694 return
1696 self.debug("Shrinking failing test cases")
1697 self.finish_shrinking_deadline = time.perf_counter() + MAX_SHRINKING_SECONDS
1699 for prev_data in sorted(
1700 self.interesting_test_cases.values(), key=lambda d: sort_key(d.nodes)
1701 ):
1702 assert prev_data.status == Status.INTERESTING
1703 data = self.new_conjecture_data(prev_data.choices)
1704 self.test_function(data)
1705 if data.status != Status.INTERESTING:
1706 self.exit_with(ExitReason.flaky)
1708 self.clear_secondary_key()
1710 while len(self.shrunk_test_cases) < len(self.interesting_test_cases):
1711 target, result = min(
1712 (
1713 (k, v)
1714 for k, v in self.interesting_test_cases.items()
1715 if k not in self.shrunk_test_cases
1716 ),
1717 key=lambda kv: (sort_key(kv[1].nodes), shortlex(repr(kv[0]))),
1718 )
1719 self.debug(f"Shrinking {target!r}: {result.choices}")
1721 if not self.settings.report_multiple_bugs:
1722 # If multi-bug reporting is disabled, we shrink our currently-minimal
1723 # failure, allowing 'slips' to any bug with a smaller minimal test case.
1724 self.shrink(result, lambda d: d.status == Status.INTERESTING)
1725 return
1727 def predicate(d: ConjectureResult | _Overrun) -> bool:
1728 if d.status < Status.INTERESTING:
1729 return False
1730 d = cast(ConjectureResult, d)
1731 return d.interesting_origin == target
1733 self.shrink(result, predicate)
1735 self.shrunk_test_cases.add(target)
1737 def clear_secondary_key(self) -> None:
1738 if self.has_existing_test_cases():
1739 # If we have any smaller test cases in the secondary corpus, now is
1740 # a good time to try them to see if they work as shrinks. They
1741 # probably won't, but it's worth a shot and gives us a good
1742 # opportunity to clear out the database.
1744 db = self.settings.database
1745 assert db is not None
1747 # It's not worth trying the primary corpus because we already
1748 # tried all of those in the initial phase.
1749 corpus = sorted(db.fetch(self.secondary_key), key=shortlex)
1750 for c in corpus:
1751 choices = choices_from_bytes(c)
1752 if choices is None:
1753 db.delete(self.secondary_key, c)
1754 continue
1755 primary = {
1756 choices_to_bytes(v.choices)
1757 for v in self.interesting_test_cases.values()
1758 }
1759 if shortlex(c) > max(map(shortlex, primary)):
1760 break
1762 self.cached_test_function(choices)
1763 # We unconditionally remove c from the secondary key as it
1764 # is either now primary or worse than our primary test case
1765 # for this reason for interestingness.
1766 db.delete(self.secondary_key, c)
1768 def shrink(
1769 self,
1770 initial: ConjectureData | ConjectureResult,
1771 predicate: ShrinkPredicateT | None = None,
1772 allow_transition: (
1773 Callable[[ConjectureData | ConjectureResult, ConjectureData], bool] | None
1774 ) = None,
1775 ) -> ConjectureData | ConjectureResult:
1776 s = self.new_shrinker(initial, predicate, allow_transition)
1777 s.shrink()
1778 return s.shrink_target
1780 def new_shrinker(
1781 self,
1782 initial: ConjectureData | ConjectureResult,
1783 predicate: ShrinkPredicateT | None = None,
1784 allow_transition: (
1785 Callable[[ConjectureData | ConjectureResult, ConjectureData], bool] | None
1786 ) = None,
1787 ) -> Shrinker:
1788 return Shrinker(
1789 self,
1790 initial,
1791 predicate,
1792 allow_transition=allow_transition,
1793 explain=Phase.explain in self.settings.phases,
1794 in_target_phase=self._current_phase == "target",
1795 )
1797 def passing_choice_sequences(
1798 self, prefix: Sequence[ChoiceNode] = ()
1799 ) -> frozenset[tuple[ChoiceNode, ...]]:
1800 """Return a collection of choice sequence nodes which cause the test to pass.
1801 Optionally restrict this by a certain prefix, which is useful for explain mode.
1802 """
1803 return frozenset(
1804 cast(ConjectureResult, result).nodes
1805 for key in self.__data_cache
1806 if (result := self.__data_cache[key]).status is Status.VALID
1807 and startswith(cast(ConjectureResult, result).nodes, prefix)
1808 )
1811class ContainsDiscard(Exception):
1812 pass