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, 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 NotRequired, TypedDict, 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 self.settings.database.delete(self.pareto_key, choices_to_bytes(data.choices))
808 def generate_novel_prefix(self) -> tuple[ChoiceT, ...]:
809 """Uses the tree to proactively generate a starting choice sequence
810 that we haven't explored yet for this test.
812 When this method is called, we assume that there must be at
813 least one novel prefix left to find. If there were not, then the
814 test run should have already stopped due to tree exhaustion.
815 """
816 return self.tree.generate_novel_prefix(self.random)
818 def record_for_health_check(self, data: ConjectureData) -> None:
819 # Once we've actually found a bug, there's no point in trying to run
820 # health checks - they'll just mask the actually important information.
821 if data.status == Status.INTERESTING:
822 self.health_check_state = None
824 state = self.health_check_state
826 if state is None:
827 return
829 for k, v in data.draw_times.items():
830 state.draw_times[k].append(v)
832 if data.status == Status.VALID:
833 state.valid_test_cases += 1
834 elif data.status == Status.INVALID:
835 state.invalid_test_cases += 1
836 else:
837 assert data.status == Status.OVERRUN
838 state.overrun_test_cases += 1
840 max_valid_draws = 10
841 max_invalid_draws = 50
842 max_overrun_draws = 20
844 assert state.valid_test_cases <= max_valid_draws
846 if state.valid_test_cases == max_valid_draws:
847 self.health_check_state = None
848 return
850 if state.overrun_test_cases == max_overrun_draws:
851 fail_health_check(
852 self.settings,
853 "Generated inputs routinely consumed more than the maximum "
854 f"allowed entropy: {state.valid_test_cases} inputs were generated "
855 f"successfully, while {state.overrun_test_cases} inputs exceeded the "
856 f"maximum allowed entropy during generation."
857 "\n\n"
858 f"Testing with inputs this large tends to be slow, and to produce "
859 "failures that are both difficult to shrink and difficult to understand. "
860 "Try decreasing the amount of data generated, for example by "
861 "decreasing the minimum size of collection strategies like "
862 "st.lists()."
863 "\n\n"
864 "If you expect the average size of your input to be this large, "
865 "you can disable this health check with "
866 "@settings(suppress_health_check=[HealthCheck.data_too_large]). "
867 "See "
868 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
869 "for details.",
870 HealthCheck.data_too_large,
871 )
872 if state.invalid_test_cases == max_invalid_draws:
873 fail_health_check(
874 self.settings,
875 "It looks like this test is filtering out a lot of inputs. "
876 f"{state.valid_test_cases} inputs were generated successfully, "
877 f"while {state.invalid_test_cases} inputs were filtered out. "
878 "\n\n"
879 "An input might be filtered out by calls to assume(), "
880 "strategy.filter(...), or occasionally by Hypothesis internals."
881 "\n\n"
882 "Applying this much filtering makes input generation slow, since "
883 "Hypothesis must discard inputs which are filtered out and try "
884 "generating it again. It is also possible that applying this much "
885 "filtering will distort the domain and/or distribution of the test, "
886 "leaving your testing less rigorous than expected."
887 "\n\n"
888 "If you expect this many inputs to be filtered out during generation, "
889 "you can disable this health check with "
890 "@settings(suppress_health_check=[HealthCheck.filter_too_much]). See "
891 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
892 "for details.",
893 HealthCheck.filter_too_much,
894 )
896 # Allow at least the greater of one second or 5x the deadline. If deadline
897 # is None, allow 30s - the user can disable the healthcheck too if desired.
898 draw_time = state.total_draw_time
899 draw_time_limit = 5 * (self.settings.deadline or timedelta(seconds=6))
900 if (
901 draw_time > max(1.0, draw_time_limit.total_seconds())
902 # we disable HealthCheck.too_slow under concurrent threads, since
903 # cpython may switch away from a thread for arbitrarily long.
904 and not self.thread_overlap.get(threading.get_ident(), False)
905 ):
906 extra_str = []
907 if state.invalid_test_cases:
908 extra_str.append(f"{state.invalid_test_cases} invalid inputs")
909 if state.overrun_test_cases:
910 extra_str.append(
911 f"{state.overrun_test_cases} inputs which exceeded the "
912 "maximum allowed entropy"
913 )
914 extra_str = ", and ".join(extra_str)
915 extra_str = f" ({extra_str})" if extra_str else ""
917 fail_health_check(
918 self.settings,
919 "Input generation is slow: Hypothesis only generated "
920 f"{state.valid_test_cases} valid inputs after {draw_time:.2f} "
921 f"seconds{extra_str}."
922 "\n" + state.timing_report() + "\n\n"
923 "This could be for a few reasons:"
924 "\n"
925 "1. This strategy could be generating too much data per input. "
926 "Try decreasing the amount of data generated, for example by "
927 "decreasing the minimum size of collection strategies like "
928 "st.lists()."
929 "\n"
930 "2. Some other expensive computation could be running during input "
931 "generation. For example, "
932 "if @st.composite or st.data() is interspersed with an expensive "
933 "computation, HealthCheck.too_slow is likely to trigger. If this "
934 "computation is unrelated to input generation, move it elsewhere. "
935 "Otherwise, try making it more efficient, or disable this health "
936 "check if that is not possible."
937 "\n\n"
938 "If you expect input generation to take this long, you can disable "
939 "this health check with "
940 "@settings(suppress_health_check=[HealthCheck.too_slow]). See "
941 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
942 "for details.",
943 HealthCheck.too_slow,
944 )
946 def save_choices(
947 self, choices: Sequence[ChoiceT], sub_key: bytes | None = None
948 ) -> None:
949 if self.settings.database is not None:
950 key = self.sub_key(sub_key)
951 if key is None:
952 return
953 self.settings.database.save(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 | None:
961 if self.database_key is None:
962 return None
963 if sub_key is None:
964 return self.database_key
965 return b".".join((self.database_key, sub_key))
967 @property
968 def secondary_key(self) -> bytes | None:
969 return self.sub_key(b"secondary")
971 @property
972 def pareto_key(self) -> bytes | None:
973 return self.sub_key(b"pareto")
975 def debug(self, message: str) -> None:
976 if self.settings.verbosity >= Verbosity.debug:
977 base_report(message)
979 @property
980 def report_debug_info(self) -> bool:
981 return self.settings.verbosity >= Verbosity.debug
983 def debug_data(self, data: ConjectureData | ConjectureResult) -> None:
984 if not self.report_debug_info:
985 return
987 status = repr(data.status)
988 if data.status == Status.INTERESTING:
989 status = f"{status} ({data.interesting_origin!r})"
990 elif data.status == Status.INVALID and isinstance(data, ConjectureData):
991 assert isinstance(data, ConjectureData) # mypy is silly
992 status = f"{status} ({data.events.get('gave up because', '?')})"
994 self.debug(f"{len(data.choices)} choices -> {status}\n\t{data.choices}")
996 def observe_for_provider(self) -> AbstractContextManager:
997 def on_observation(observation: Observation) -> None:
998 assert observation.type == "test_case"
999 # because lifetime == "test_function"
1000 assert isinstance(self.provider, PrimitiveProvider)
1001 # only fire if we actually used that provider to generate this observation
1002 if not self._switch_to_hypothesis_provider:
1003 self.provider.on_observation(observation)
1005 if (
1006 self.settings.backend != "hypothesis"
1007 # only for lifetime = "test_function" providers (guaranteed
1008 # by this isinstance check)
1009 and isinstance(self.provider, PrimitiveProvider)
1010 # and the provider opted-in to observations
1011 and self.provider.add_observability_callback
1012 ):
1013 return with_observability_callback(on_observation)
1014 return nullcontext()
1016 def run(self) -> None:
1017 with local_settings(self.settings), self.observe_for_provider():
1018 try:
1019 self._run()
1020 except RunIsComplete:
1021 pass
1022 for v in self.interesting_test_cases.values():
1023 self.debug_data(v)
1024 self.debug(
1025 f"Run complete after {self.call_count} test cases "
1026 f"({self.valid_test_cases} valid) and {self.shrinks} shrinks"
1027 )
1029 @property
1030 def database(self) -> ExampleDatabase | None:
1031 if self.database_key is None:
1032 return None
1033 return self.settings.database
1035 def has_existing_test_cases(self) -> bool:
1036 return self.database is not None and Phase.reuse in self.settings.phases
1038 def reuse_existing_test_cases(self) -> None:
1039 """If appropriate (we have a database and have been told to use it),
1040 try to reload existing test cases from the database.
1042 If there are a lot we don't try all of them. We always try the
1043 smallest test case in the database (which is guaranteed to be the
1044 last failure) and the largest (which is usually the seed test case
1045 which the last failure came from but we don't enforce that). We
1046 then take a random sampling of the remainder and try those. Any
1047 test cases that are no longer interesting are cleared out.
1048 """
1049 if self.has_existing_test_cases():
1050 self.debug("Reusing test cases from database")
1051 # We have to do some careful juggling here. We have two database
1052 # corpora: The primary and secondary. The primary corpus is a
1053 # small set of minimized test cases each of which has at one point
1054 # demonstrated a distinct bug. We want to retry all of these.
1056 # We also have a secondary corpus of test cases that have at some
1057 # point demonstrated interestingness (currently only ones that
1058 # were previously non-minimal test cases for a bug, but this will
1059 # likely expand in future). These are a good source of potentially
1060 # interesting test cases, but there are a lot of them, so we down
1061 # sample the secondary corpus to a more manageable size.
1063 corpus = sorted(
1064 self.settings.database.fetch(self.database_key), key=shortlex
1065 )
1066 factor = 0.1 if (Phase.generate in self.settings.phases) else 1
1067 desired_size = max(2, ceil(factor * self.settings.max_examples))
1068 primary_corpus_size = len(corpus)
1070 if len(corpus) < desired_size:
1071 extra_corpus = list(self.settings.database.fetch(self.secondary_key))
1073 shortfall = desired_size - len(corpus)
1075 if len(extra_corpus) <= shortfall:
1076 extra = extra_corpus
1077 else:
1078 extra = self.random.sample(extra_corpus, shortfall)
1079 extra.sort(key=shortlex)
1080 corpus.extend(extra)
1082 # We want a fast path where every primary entry in the database was
1083 # interesting.
1084 found_interesting_in_primary = False
1085 all_interesting_in_primary_were_exact = True
1087 for i, existing in enumerate(corpus):
1088 if i >= primary_corpus_size and found_interesting_in_primary:
1089 break
1090 choices = choices_from_bytes(existing)
1091 if choices is None:
1092 # clear out any keys which fail deserialization
1093 self.settings.database.delete(self.database_key, existing)
1094 continue
1095 data = self.cached_test_function(choices, extend="full")
1096 if data.status != Status.INTERESTING:
1097 self.settings.database.delete(self.database_key, existing)
1098 self.settings.database.delete(self.secondary_key, existing)
1099 else:
1100 if i < primary_corpus_size:
1101 found_interesting_in_primary = True
1102 assert not isinstance(data, _Overrun)
1103 if choices_key(choices) != choices_key(data.choices):
1104 all_interesting_in_primary_were_exact = False
1105 if not self.settings.report_multiple_bugs:
1106 break
1107 if found_interesting_in_primary:
1108 if all_interesting_in_primary_were_exact:
1109 self.reused_previously_shrunk_test_case = True
1111 # Because self.database is not None (because self.has_existing_test_cases())
1112 # and self.database_key is not None (because we fetched using it above),
1113 # we can guarantee self.pareto_front is not None
1114 assert self.pareto_front is not None
1116 # If we've not found any interesting test cases so far we try some of
1117 # the pareto front from the last run.
1118 if len(corpus) < desired_size and not self.interesting_test_cases:
1119 desired_extra = desired_size - len(corpus)
1120 pareto_corpus = list(self.settings.database.fetch(self.pareto_key))
1121 if len(pareto_corpus) > desired_extra:
1122 pareto_corpus = self.random.sample(pareto_corpus, desired_extra)
1123 pareto_corpus.sort(key=shortlex)
1125 for existing in pareto_corpus:
1126 choices = choices_from_bytes(existing)
1127 if choices is None:
1128 self.settings.database.delete(self.pareto_key, existing)
1129 continue
1130 data = self.cached_test_function(choices, extend="full")
1131 if data not in self.pareto_front:
1132 self.settings.database.delete(self.pareto_key, existing)
1133 if data.status == Status.INTERESTING:
1134 break
1136 def exit_with(self, reason: ExitReason) -> None:
1137 if self.ignore_limits:
1138 return
1139 self.statistics["stopped-because"] = reason.describe(self.settings)
1140 if self.best_observed_targets:
1141 self.statistics["targets"] = dict(self.best_observed_targets)
1142 self.debug(f"exit_with({reason.name})")
1143 self.exit_reason = reason
1144 raise RunIsComplete
1146 def should_generate_more(self) -> bool:
1147 # End the generation phase where we would have ended it if no bugs had
1148 # been found. This reproduces the exit logic in `self.test_function`,
1149 # but with the important distinction that this clause will move on to
1150 # the shrinking phase having found one or more bugs, while the other
1151 # will exit having found zero bugs.
1152 invalid_threshold = (
1153 INVALID_THRESHOLD_BASE + INVALID_PER_VALID * self.valid_test_cases
1154 )
1155 if (
1156 self.valid_test_cases >= self.settings.max_examples
1157 or (self.invalid_test_cases + self.overrun_test_cases) > invalid_threshold
1158 or self.__tree_is_exhausted()
1159 ):
1160 return False
1162 # If we haven't found a bug, keep looking - if we hit any limits on
1163 # the number of tests to run that will raise an exception and stop
1164 # the run.
1165 if not self.interesting_test_cases:
1166 return True
1167 # Users who disable shrinking probably want to exit as fast as possible.
1168 # If we've found a bug and won't report more than one, stop looking.
1169 # If we first saw a bug more than 10 seconds ago, stop looking.
1170 elif (
1171 Phase.shrink not in self.settings.phases
1172 or not self.settings.report_multiple_bugs
1173 or time.monotonic() - self.first_bug_found_time > 10
1174 ):
1175 return False
1176 assert isinstance(self.first_bug_found_at, int)
1177 assert isinstance(self.last_bug_found_at, int)
1178 assert self.first_bug_found_at <= self.last_bug_found_at <= self.call_count
1179 # Otherwise, keep searching for between ten and 'a heuristic' calls.
1180 # We cap 'calls after first bug' so errors are reported reasonably
1181 # soon even for tests that are allowed to run for a very long time,
1182 # or sooner if the latest half of our test effort has been fruitless.
1183 return self.call_count < MIN_TEST_CALLS or self.call_count < min(
1184 self.first_bug_found_at + 1000, self.last_bug_found_at * 2
1185 )
1187 def generate_new_test_cases(self) -> None:
1188 if Phase.generate not in self.settings.phases:
1189 return
1190 if self.interesting_test_cases:
1191 # The example database has failing test cases from a previous run,
1192 # so we'd rather report that they're still failing ASAP than take
1193 # the time to look for additional failures.
1194 return
1196 self.debug("Generating new test cases")
1198 assert self.should_generate_more()
1199 self._switch_to_hypothesis_provider = True
1200 zero_data = self.cached_test_function((ChoiceTemplate("simplest", count=None),))
1201 if zero_data.status > Status.OVERRUN:
1202 assert isinstance(zero_data, ConjectureResult)
1203 # if the crosshair backend cannot proceed, it does not (and cannot)
1204 # realize the symbolic values, with the intent that Hypothesis will
1205 # throw away this test case. We usually do, but if it's the zero data
1206 # then we try to pin it here, which requires realizing the symbolics.
1207 #
1208 # We don't (yet) rely on the zero data being pinned, and so
1209 # it's simply a very slight performance loss to simply not pin it
1210 # if doing so would error.
1211 if zero_data.cannot_proceed_scope is None: # pragma: no branch
1212 self.__data_cache.pin(
1213 self._cache_key(zero_data.choices), zero_data.as_result()
1214 ) # Pin forever
1216 if zero_data.status == Status.OVERRUN or (
1217 zero_data.status == Status.VALID
1218 and isinstance(zero_data, ConjectureResult)
1219 and zero_data.length * 2 > BUFFER_SIZE
1220 ):
1221 fail_health_check(
1222 self.settings,
1223 "The smallest natural input for this test is very "
1224 "large. This makes it difficult for Hypothesis to generate "
1225 "good inputs, especially when trying to shrink failing inputs."
1226 "\n\n"
1227 "Consider reducing the amount of data generated by the strategy. "
1228 "Also consider introducing small alternative values for some "
1229 "strategies. For example, could you "
1230 "mark some arguments as optional by replacing `some_complex_strategy`"
1231 "with `st.none() | some_complex_strategy`?"
1232 "\n\n"
1233 "If you are confident that the size of the smallest natural input "
1234 "to your test cannot be reduced, you can suppress this health check "
1235 "with @settings(suppress_health_check=[HealthCheck.large_base_example]). "
1236 "See "
1237 "https://hypothesis.readthedocs.io/en/latest/reference/api.html#hypothesis.HealthCheck "
1238 "for details.",
1239 HealthCheck.large_base_example,
1240 )
1242 self.health_check_state = HealthCheckState()
1244 # We attempt to use the size of the minimal generated test case starting
1245 # from a given novel prefix as a guideline to generate smaller test
1246 # cases for an initial period, by restriscting ourselves to test cases
1247 # that are not much larger than it.
1248 #
1249 # Calculating the actual minimal generated test case is hard, so we
1250 # take a best guess that zero extending a prefix produces the minimal
1251 # test case starting with that prefix (this is true for our built in
1252 # strategies). This is only a reasonable thing to do if the resulting
1253 # test case is valid. If we regularly run into situations where it is
1254 # not valid then this strategy is a waste of time, so we want to
1255 # abandon it early. In order to do this we track how many times in a
1256 # row it has failed to work, and abort small test case generation when
1257 # it has failed too many times in a row.
1258 consecutive_zero_extend_is_invalid = 0
1260 # We control growth during initial test case generation, for two
1261 # reasons:
1262 #
1263 # * It gives us an opportunity to find small test cases early, which
1264 # gives us a fast path for easy to find bugs.
1265 # * It avoids low probability events where we might end up
1266 # generating very large test cases during health checks, which
1267 # on slower machines can trigger HealthCheck.too_slow.
1268 #
1269 # The heuristic we use is that we attempt to estimate the smallest
1270 # extension of this prefix, and limit the size to no more than
1271 # an order of magnitude larger than that. If we fail to estimate
1272 # the size accurately, we skip over this prefix and try again.
1273 #
1274 # We need to tune the test case size based on the initial prefix,
1275 # because any fixed size might be too small, and any size based
1276 # on the strategy in general can fall afoul of strategies that
1277 # have very different sizes for different prefixes.
1278 #
1279 # We previously set a minimum value of 10 on small_test_case_cap, with the
1280 # reasoning of avoiding flaky health checks. However, some users set a
1281 # low max_examples for performance. A hard lower bound in this case biases
1282 # the distribution towards small (and less powerful) test cases. Flaky
1283 # and loud health checks are better than silent performance degradation.
1284 small_test_case_cap = min(self.settings.max_examples // 10, 50)
1285 # For small budgets we run a single optimisation pass halfway through
1286 # the budget, which experience shows is roughly optimal. For larger
1287 # budgets we instead start optimising early and interleave repeated
1288 # passes with generation - see _should_optimise_now.
1289 if self.settings.max_examples < 1000:
1290 self._next_optimise_at = max(
1291 self.settings.max_examples // 2, small_test_case_cap + 1, 10
1292 )
1293 else:
1294 self._next_optimise_at = max(200, small_test_case_cap + 1)
1295 self._switch_to_hypothesis_provider = False
1297 while self.should_generate_more():
1298 # we don't yet integrate DataTree with backends. Instead of generating
1299 # a novel prefix, ask the backend for an input.
1300 if not self.using_hypothesis_backend:
1301 data = self.new_conjecture_data([])
1302 self.test_function(data)
1303 continue
1305 self._current_phase = "generate"
1306 prefix = self.generate_novel_prefix()
1307 if (
1308 self.valid_test_cases <= small_test_case_cap
1309 and self.call_count <= 5 * small_test_case_cap
1310 and not self.interesting_test_cases
1311 and consecutive_zero_extend_is_invalid < 5
1312 ):
1313 minimal_test_case = self.cached_test_function(
1314 prefix + (ChoiceTemplate("simplest", count=None),)
1315 )
1317 if minimal_test_case.status < Status.VALID:
1318 consecutive_zero_extend_is_invalid += 1
1319 continue
1320 # Because the Status code is greater than Status.VALID, it cannot be
1321 # Status.OVERRUN, which guarantees that the minimal_test_case is a
1322 # ConjectureResult object.
1323 assert isinstance(minimal_test_case, ConjectureResult)
1324 consecutive_zero_extend_is_invalid = 0
1325 minimal_extension = len(minimal_test_case.choices) - len(prefix)
1326 max_length = len(prefix) + minimal_extension * 5
1328 # We could end up in a situation where even though the prefix was
1329 # novel when we generated it, because we've now tried zero extending
1330 # it not all possible continuations of it will be novel. In order to
1331 # avoid making redundant test calls, we rerun it in simulation mode
1332 # first. If this has a predictable result, then we don't bother
1333 # running the test function for real here. If however we encounter
1334 # some novel behaviour, we try again with the real test function,
1335 # starting from the new novel prefix that has discovered.
1336 trial_data = self.new_conjecture_data(prefix, max_choices=max_length)
1337 try:
1338 self.tree.simulate_test_function(trial_data)
1339 continue
1340 except PreviouslyUnseenBehaviour:
1341 pass
1343 # If the simulation entered part of the tree that has been killed,
1344 # we don't want to run this.
1345 assert isinstance(trial_data.observer, TreeRecordingObserver)
1346 if trial_data.observer.killed:
1347 continue
1349 # We might have hit the cap on number of test cases we should
1350 # run when calculating the minimal test case.
1351 if not self.should_generate_more():
1352 break
1354 prefix = trial_data.choices
1355 else:
1356 max_length = None
1358 data = self.new_conjecture_data(prefix, max_choices=max_length)
1359 self.test_function(data)
1361 if (
1362 data.status is Status.OVERRUN
1363 and max_length is not None
1364 and "gave up because" not in data.events
1365 ):
1366 data.events["gave up because"] = (
1367 "reduced max size for early test cases (avoids flaky health checks)"
1368 )
1370 self.generate_mutations_from(data)
1372 # Although the optimisations are logically a distinct phase, we
1373 # actually normally run them as part of test case generation. The
1374 # reason for this is that we cannot guarantee that optimisation
1375 # actually exhausts our budget: It might finish running and we
1376 # discover that actually we still could run a bunch more test cases
1377 # if we want.
1378 if self._should_optimise_now():
1379 self._run_optimise_pass()
1381 def generate_mutations_from(self, data: ConjectureData | ConjectureResult) -> None:
1382 # A thing that is often useful but rarely happens by accident is
1383 # to generate the same value at multiple different points in the
1384 # test case.
1385 #
1386 # Rather than make this the responsibility of individual strategies
1387 # we implement a small mutator that just takes parts of the test
1388 # case with the same label and tries replacing one of them with a
1389 # copy of the other and tries running it. If we've made a good
1390 # guess about what to put where, this will run a similar generated
1391 # test case with more duplication.
1392 if (
1393 # An OVERRUN doesn't have enough information about the test
1394 # case to mutate, so we just skip those.
1395 data.status >= Status.INVALID
1396 # This has a tendency to trigger some weird edge cases during
1397 # generation so we don't let it run until we're done with the
1398 # health checks.
1399 and self.health_check_state is None
1400 ):
1401 initial_calls = self.call_count
1402 failed_mutations = 0
1404 while (
1405 self.should_generate_more()
1406 # We implement fairly conservative checks for how long we
1407 # we should run mutation for, as it's generally not obvious
1408 # how helpful it is for any given test case.
1409 and self.call_count <= initial_calls + 5
1410 and failed_mutations <= 5
1411 ):
1412 groups = data.spans.mutator_groups
1413 if not groups:
1414 break
1416 group = self.random.choice(groups)
1417 (start1, end1), (start2, end2) = self.random.sample(sorted(group), 2)
1418 if start1 > start2:
1419 (start1, end1), (start2, end2) = (start2, end2), (start1, end1)
1421 if (
1422 start1 <= start2 <= end2 <= end1
1423 ): # pragma: no cover # flaky on conjecture-cover tests
1424 # One span entirely contains the other. The strategy is very
1425 # likely some kind of tree. e.g. we might have
1426 #
1427 # ┌─────┐
1428 # ┌─────┤ a ├──────┐
1429 # │ └─────┘ │
1430 # ┌──┴──┐ ┌──┴──┐
1431 # ┌──┤ b ├──┐ ┌──┤ c ├──┐
1432 # │ └──┬──┘ │ │ └──┬──┘ │
1433 # ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐
1434 # │ d │ │ e │ │ f │ │ g │ │ h │ │ i │
1435 # └───┘ └───┘ └───┘ └───┘ └───┘ └───┘
1436 #
1437 # where each node is drawn from the same strategy and so
1438 # has the same span label. We might have selected the spans
1439 # corresponding to the a and c nodes, which is the entire
1440 # tree and the subtree of (and including) c respectively.
1441 #
1442 # There are two possible mutations we could apply in this case:
1443 # 1. replace a with c (replace child with parent)
1444 # 2. replace c with a (replace parent with child)
1445 #
1446 # (1) results in multiple partial copies of the
1447 # parent:
1448 # ┌─────┐
1449 # ┌─────┤ a ├────────────┐
1450 # │ └─────┘ │
1451 # ┌──┴──┐ ┌─┴───┐
1452 # ┌──┤ b ├──┐ ┌─────┤ a ├──────┐
1453 # │ └──┬──┘ │ │ └─────┘ │
1454 # ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌──┴──┐ ┌──┴──┐
1455 # │ d │ │ e │ │ f │ ┌──┤ b ├──┐ ┌──┤ c ├──┐
1456 # └───┘ └───┘ └───┘ │ └──┬──┘ │ │ └──┬──┘ │
1457 # ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐
1458 # │ d │ │ e │ │ f │ │ g │ │ h │ │ i │
1459 # └───┘ └───┘ └───┘ └───┘ └───┘ └───┘
1460 #
1461 # While (2) results in truncating part of the parent:
1462 #
1463 # ┌─────┐
1464 # ┌──┤ c ├──┐
1465 # │ └──┬──┘ │
1466 # ┌─┴─┐ ┌─┴─┐ ┌─┴─┐
1467 # │ g │ │ h │ │ i │
1468 # └───┘ └───┘ └───┘
1469 #
1470 # (1) is the same as Example IV.4. in Nautilus (NDSS '19)
1471 # (https://wcventure.github.io/FuzzingPaper/Paper/NDSS19_Nautilus.pdf),
1472 # except we do not repeat the replacement additional times
1473 # (the paper repeats it once for a total of two copies).
1474 #
1475 # We currently only apply mutation (1), and ignore mutation
1476 # (2). The reason is that the attempt generated from (2) is
1477 # always something that Hypothesis could easily have generated
1478 # itself, by simply not making various choices. Whereas
1479 # duplicating the exact value + structure of particular choices
1480 # in (1) would have been hard for Hypothesis to generate by
1481 # chance.
1482 #
1483 # TODO: an extension of this mutation might repeat (1) on
1484 # a geometric distribution between 0 and ~10 times. We would
1485 # need to find the corresponding span to recurse on in the new
1486 # choices, probably just by using the choices index.
1488 # case (1): duplicate the choices in start1:start2.
1489 attempt = data.choices[:start2] + data.choices[start1:]
1490 else:
1491 start, end = self.random.choice([(start1, end1), (start2, end2)])
1492 replacement = data.choices[start:end]
1493 # We attempt to replace both the spans with
1494 # whichever choice we made. Note that this might end
1495 # up messing up and getting the span boundaries
1496 # wrong - labels matching are only a best guess as to
1497 # whether the two are equivalent - but it doesn't
1498 # really matter. It may not achieve the desired result,
1499 # but it's still a perfectly acceptable choice sequence
1500 # to try.
1501 attempt = (
1502 data.choices[:start1]
1503 + replacement
1504 + data.choices[end1:start2]
1505 + replacement
1506 + data.choices[end2:]
1507 )
1509 try:
1510 new_data = self.cached_test_function(
1511 attempt,
1512 # We set error_on_discard so that we don't end up
1513 # entering parts of the tree we consider redundant
1514 # and not worth exploring.
1515 error_on_discard=True,
1516 )
1517 except ContainsDiscard:
1518 failed_mutations += 1
1519 continue
1521 if new_data is Overrun:
1522 failed_mutations += 1 # pragma: no cover # annoying case
1523 else:
1524 assert isinstance(new_data, ConjectureResult)
1525 if (
1526 new_data.status >= data.status
1527 and choices_key(data.choices) != choices_key(new_data.choices)
1528 and all(
1529 k in new_data.target_observations
1530 and new_data.target_observations[k] >= v
1531 for k, v in data.target_observations.items()
1532 )
1533 ):
1534 data = new_data
1535 failed_mutations = 0
1536 else:
1537 failed_mutations += 1
1539 def _should_optimise_now(self) -> bool:
1540 """Decide whether to run an optimisation pass at this point in generation."""
1541 if not self.should_optimise:
1542 return False
1543 if self.valid_test_cases >= self._next_optimise_at:
1544 return True
1545 # After the first pass, we should optimize if we have found a new best score from
1546 # generation and we have the budget to continue optimizing.
1547 return (
1548 self._best_scores_at_last_pass is not None
1549 and self.settings.max_examples >= 1000
1550 and self._target_valid_spent < self.settings.max_examples // 2
1551 and any(
1552 score > self._best_scores_at_last_pass.get(target, NO_SCORE)
1553 for target, score in self.best_observed_targets.items()
1554 )
1555 )
1557 def _run_optimise_pass(self) -> None:
1558 """Run one optimisation pass, then schedule the next one."""
1559 if self.settings.max_examples < 1000:
1560 # We'll only ever run a single optimization pass in this case.
1561 max_valid = None
1562 else:
1563 remaining = self.settings.max_examples // 2 - self._target_valid_spent
1564 max_valid = self.valid_test_cases + min(max(200, remaining // 4), remaining)
1565 self._current_phase = "target"
1566 start_valid = self.valid_test_cases
1567 improved = False
1568 try:
1569 improved = self.optimise_targets(max_valid=max_valid)
1570 finally:
1571 spent = self.valid_test_cases - start_valid
1572 self._target_valid_spent += spent
1573 # Alternate with generation on a fair-share schedule for as long
1574 # as passes keep finding improvements and budget remains.
1575 if (
1576 max_valid is not None
1577 and improved
1578 and self._target_valid_spent < self.settings.max_examples // 2
1579 ):
1580 self._next_optimise_at = self.valid_test_cases + spent
1581 else:
1582 self._next_optimise_at = math.inf
1583 self._best_scores_at_last_pass = dict(self.best_observed_targets)
1584 self._current_phase = "generate"
1586 def optimise_targets(self, *, max_valid: int | None = None) -> bool:
1587 """If any target observations have been made, attempt to optimise them
1588 all, stopping early if ``self.valid_test_cases`` reaches ``max_valid``.
1589 Returns whether this made any improvements."""
1590 if not self.should_optimise:
1591 return False
1592 from hypothesis.internal.conjecture.optimiser import Optimiser
1594 # We want to avoid running the optimiser for too long in case we hit
1595 # an unbounded target score. We start this off fairly conservatively
1596 # in case interesting test cases are easy to find and then ramp it up
1597 # on an exponential schedule so we don't hamper the optimiser too much
1598 # if it needs a long time to find good enough improvements.
1599 improved = False
1600 max_improvements = 10
1601 while True:
1602 prev_calls = self.call_count
1604 any_improvements = False
1606 for target, data in list(self.best_test_cases_of_observed_targets.items()):
1607 if max_valid is not None and self.valid_test_cases >= max_valid:
1608 return improved
1609 optimiser = Optimiser(
1610 self, data, target, max_improvements=max_improvements
1611 )
1612 optimiser.run()
1613 if optimiser.improvements > 0:
1614 any_improvements = True
1615 improved = True
1617 if self.interesting_test_cases:
1618 break
1620 max_improvements *= 2
1622 if any_improvements:
1623 continue
1625 if self.best_observed_targets:
1626 self.pareto_optimise()
1628 if prev_calls == self.call_count:
1629 break
1630 return improved
1632 def pareto_optimise(self) -> None:
1633 if self.pareto_front is not None:
1634 ParetoOptimiser(self).run()
1636 def _run(self) -> None:
1637 # have to use the primitive provider to interpret database bits...
1638 self._switch_to_hypothesis_provider = True
1639 with self._log_phase_statistics("reuse"):
1640 self.reuse_existing_test_cases()
1641 # Fast path for development: If the database gave us interesting
1642 # test cases from the previously stored primary key, don't try
1643 # shrinking it again as it's unlikely to work.
1644 if self.reused_previously_shrunk_test_case:
1645 self.exit_with(ExitReason.finished)
1646 # ...but we should use the supplied provider when generating...
1647 self._switch_to_hypothesis_provider = False
1648 with self._log_phase_statistics("generate"):
1649 self.generate_new_test_cases()
1650 # We normally run the target phase mixed in with the generate phase,
1651 # but if we've been asked to run it but not generation then we have to
1652 # run it explicitly on its own here.
1653 if Phase.generate not in self.settings.phases:
1654 self._current_phase = "target"
1655 self.optimise_targets()
1656 # ...and back to the primitive provider when shrinking.
1657 self._switch_to_hypothesis_provider = True
1658 with self._log_phase_statistics("shrink"):
1659 self.shrink_interesting_test_cases()
1660 self.exit_with(ExitReason.finished)
1662 def new_conjecture_data(
1663 self,
1664 prefix: Sequence[ChoiceT | ChoiceTemplate | ValueHole],
1665 *,
1666 observer: DataObserver | None = None,
1667 max_choices: int | None = None,
1668 ) -> ConjectureData:
1669 provider = (
1670 HypothesisProvider if self._switch_to_hypothesis_provider else self.provider
1671 )
1672 observer = observer or self.tree.new_observer()
1673 if not self.using_hypothesis_backend:
1674 observer = DataObserver()
1676 return ConjectureData(
1677 prefix=prefix,
1678 observer=observer,
1679 provider=provider,
1680 max_choices=max_choices,
1681 random=self.random,
1682 )
1684 def shrink_interesting_test_cases(self) -> None:
1685 """If we've found interesting test cases, try to replace each of them
1686 with a minimal interesting test case with the same interesting_origin.
1688 We may find one or more test cases with a new interesting_origin
1689 during the shrink process. If so we shrink these too.
1690 """
1691 if Phase.shrink not in self.settings.phases or not self.interesting_test_cases:
1692 return
1694 self.debug("Shrinking failing test cases")
1695 self.finish_shrinking_deadline = time.perf_counter() + MAX_SHRINKING_SECONDS
1697 for prev_data in sorted(
1698 self.interesting_test_cases.values(), key=lambda d: sort_key(d.nodes)
1699 ):
1700 assert prev_data.status == Status.INTERESTING
1701 data = self.new_conjecture_data(prev_data.choices)
1702 self.test_function(data)
1703 if data.status != Status.INTERESTING:
1704 self.exit_with(ExitReason.flaky)
1706 self.clear_secondary_key()
1708 while len(self.shrunk_test_cases) < len(self.interesting_test_cases):
1709 target, result = min(
1710 (
1711 (k, v)
1712 for k, v in self.interesting_test_cases.items()
1713 if k not in self.shrunk_test_cases
1714 ),
1715 key=lambda kv: (sort_key(kv[1].nodes), shortlex(repr(kv[0]))),
1716 )
1717 self.debug(f"Shrinking {target!r}: {result.choices}")
1719 if not self.settings.report_multiple_bugs:
1720 # If multi-bug reporting is disabled, we shrink our currently-minimal
1721 # failure, allowing 'slips' to any bug with a smaller minimal test case.
1722 self.shrink(result, lambda d: d.status == Status.INTERESTING)
1723 return
1725 def predicate(d: ConjectureResult | _Overrun) -> bool:
1726 if d.status < Status.INTERESTING:
1727 return False
1728 d = cast(ConjectureResult, d)
1729 return d.interesting_origin == target
1731 self.shrink(result, predicate)
1733 self.shrunk_test_cases.add(target)
1735 def clear_secondary_key(self) -> None:
1736 if self.has_existing_test_cases():
1737 # If we have any smaller test cases in the secondary corpus, now is
1738 # a good time to try them to see if they work as shrinks. They
1739 # probably won't, but it's worth a shot and gives us a good
1740 # opportunity to clear out the database.
1742 # It's not worth trying the primary corpus because we already
1743 # tried all of those in the initial phase.
1744 corpus = sorted(
1745 self.settings.database.fetch(self.secondary_key), key=shortlex
1746 )
1747 for c in corpus:
1748 choices = choices_from_bytes(c)
1749 if choices is None:
1750 self.settings.database.delete(self.secondary_key, c)
1751 continue
1752 primary = {
1753 choices_to_bytes(v.choices)
1754 for v in self.interesting_test_cases.values()
1755 }
1756 if shortlex(c) > max(map(shortlex, primary)):
1757 break
1759 self.cached_test_function(choices)
1760 # We unconditionally remove c from the secondary key as it
1761 # is either now primary or worse than our primary test case
1762 # for this reason for interestingness.
1763 self.settings.database.delete(self.secondary_key, c)
1765 def shrink(
1766 self,
1767 initial: ConjectureData | ConjectureResult,
1768 predicate: ShrinkPredicateT | None = None,
1769 allow_transition: (
1770 Callable[[ConjectureData | ConjectureResult, ConjectureData], bool] | None
1771 ) = None,
1772 ) -> ConjectureData | ConjectureResult:
1773 s = self.new_shrinker(initial, predicate, allow_transition)
1774 s.shrink()
1775 return s.shrink_target
1777 def new_shrinker(
1778 self,
1779 initial: ConjectureData | ConjectureResult,
1780 predicate: ShrinkPredicateT | None = None,
1781 allow_transition: (
1782 Callable[[ConjectureData | ConjectureResult, ConjectureData], bool] | None
1783 ) = None,
1784 ) -> Shrinker:
1785 return Shrinker(
1786 self,
1787 initial,
1788 predicate,
1789 allow_transition=allow_transition,
1790 explain=Phase.explain in self.settings.phases,
1791 in_target_phase=self._current_phase == "target",
1792 )
1794 def passing_choice_sequences(
1795 self, prefix: Sequence[ChoiceNode] = ()
1796 ) -> frozenset[tuple[ChoiceNode, ...]]:
1797 """Return a collection of choice sequence nodes which cause the test to pass.
1798 Optionally restrict this by a certain prefix, which is useful for explain mode.
1799 """
1800 return frozenset(
1801 cast(ConjectureResult, result).nodes
1802 for key in self.__data_cache
1803 if (result := self.__data_cache[key]).status is Status.VALID
1804 and startswith(cast(ConjectureResult, result).nodes, prefix)
1805 )
1808class ContainsDiscard(Exception):
1809 pass