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 math
12import unicodedata
13from collections import defaultdict
14from collections.abc import Callable, Iterator, Sequence
15from dataclasses import dataclass
16from functools import lru_cache
17from typing import (
18 TYPE_CHECKING,
19 Any,
20 Literal,
21 TypeAlias,
22 cast,
23)
25from hypothesis.internal.conjecture.choice import (
26 ChoiceNode,
27 ChoiceT,
28 choice_equal,
29 choice_from_index,
30 choice_key,
31 choice_permitted,
32 choice_to_index,
33 choices_key,
34)
35from hypothesis.internal.conjecture.data import (
36 ConjectureData,
37 ConjectureResult,
38 Spans,
39 Status,
40 _Overrun,
41 draw_choice,
42)
43from hypothesis.internal.conjecture.junkdrawer import (
44 endswith,
45 find_integer,
46 replace_all,
47 startswith,
48)
49from hypothesis.internal.conjecture.shrinking import (
50 Bytes,
51 Float,
52 Integer,
53 Ordering,
54 String,
55)
56from hypothesis.internal.conjecture.shrinking.choicetree import (
57 ChoiceTree,
58 prefix_selection_order,
59 random_selection_order,
60)
61from hypothesis.internal.floats import MAX_PRECISE_INTEGER
63if TYPE_CHECKING:
64 from random import Random
66 from hypothesis.internal.conjecture.engine import ConjectureRunner
68ShrinkPredicateT: TypeAlias = Callable[[ConjectureResult | _Overrun], bool]
71def sort_key(nodes: Sequence[ChoiceNode]) -> tuple[int, tuple[int, ...]]:
72 """Returns a sort key such that "simpler" choice sequences are smaller than
73 "more complicated" ones.
75 We define sort_key so that x is simpler than y if x is shorter than y or if
76 they have the same length and map(choice_to_index, x) < map(choice_to_index, y).
78 The reason for using this ordering is:
80 1. If x is shorter than y then that means we had to make fewer decisions
81 in constructing the test case when we ran x than we did when we ran y.
82 2. If x is the same length as y then replacing a choice with a lower index
83 choice corresponds to replacing it with a simpler/smaller choice.
84 3. Because choices drawn early in generation potentially get used in more
85 places they potentially have a more significant impact on the final
86 result, so it makes sense to prioritise reducing earlier choices over
87 later ones.
88 """
89 return (
90 len(nodes),
91 tuple(choice_to_index(node.value, node.constraints) for node in nodes),
92 )
95@lru_cache(maxsize=4096)
96def _natural_simpler_chars(c, intervals):
97 """Return single-char replacements for ``c`` derived from natural text
98 transformations - case mapping (upper, lower, casefold) and unicode
99 decomposition (NFD, NFKD). We take each individual character of the
100 transformed form so that e.g. ``ß`` can shrink to ``s`` via casefold
101 even though the full case-folded form is two characters.
103 Only candidates which are in ``intervals`` and which have a strictly
104 smaller index in shrink order than ``c`` are returned, sorted by that
105 shrink-order index. Callers must pass a single character that is itself
106 in ``intervals``.
107 """
108 candidates: set[str] = set()
109 for form in ("NFKD", "NFD"):
110 candidates.update(unicodedata.normalize(form, c))
111 for transformed in (c.upper(), c.lower(), c.casefold()):
112 candidates.update(transformed)
113 candidates.discard(c)
114 original_idx = intervals.index_from_char_in_shrink_order(c)
115 result = sorted(
116 (intervals.index_from_char_in_shrink_order(cand), cand)
117 for cand in candidates
118 if cand in intervals
119 )
120 return [cand for idx, cand in result if idx < original_idx]
123@dataclass(slots=True, frozen=False)
124class ShrinkPass:
125 function: Any
126 name: str | None = None
127 last_prefix: Any = ()
129 # some execution statistics
130 calls: int = 0
131 misaligned: int = 0
132 shrinks: int = 0
133 deletions: int = 0
135 def __post_init__(self):
136 if self.name is None:
137 self.name = self.function.__name__
139 def __hash__(self):
140 return hash(self.name)
143class StopShrinking(Exception):
144 pass
147class Shrinker:
148 """A shrinker is a child object of a ConjectureRunner which is designed to
149 manage the associated state of a particular shrink problem. That is, we
150 have some initial ConjectureData object and some property of interest
151 that it satisfies, and we want to find a ConjectureData object with a
152 shortlex (see sort_key above) smaller choice sequence that exhibits the same
153 property.
155 Currently the only property of interest we use is that the status is
156 INTERESTING and the interesting_origin takes on some fixed value, but we
157 may potentially be interested in other use cases later.
158 However we assume that data with a status < VALID never satisfies the predicate.
160 The shrinker keeps track of a value shrink_target which represents the
161 current best known ConjectureData object satisfying the predicate.
162 It refines this value by repeatedly running *shrink passes*, which are
163 methods that perform a series of transformations to the current shrink_target
164 and evaluate the underlying test function to find new ConjectureData
165 objects. If any of these satisfy the predicate, the shrink_target
166 is updated automatically. Shrinking runs until no shrink pass can
167 improve the shrink_target, at which point it stops. It may also be
168 terminated if the underlying engine throws RunIsComplete, but that
169 is handled by the calling code rather than the Shrinker.
171 =======================
172 Designing Shrink Passes
173 =======================
175 Generally a shrink pass is just any function that calls
176 cached_test_function and/or consider_new_nodes a number of times,
177 but there are a couple of useful things to bear in mind.
179 A shrink pass *makes progress* if running it changes self.shrink_target
180 (i.e. it tries a shortlex smaller ConjectureData object satisfying
181 the predicate). The desired end state of shrinking is to find a
182 value such that no shrink pass can make progress, i.e. that we
183 are at a local minimum for each shrink pass.
185 In aid of this goal, the main invariant that a shrink pass much
186 satisfy is that whether it makes progress must be deterministic.
187 It is fine (encouraged even) for the specific progress it makes
188 to be non-deterministic, but if you run a shrink pass, it makes
189 no progress, and then you immediately run it again, it should
190 never succeed on the second time. This allows us to stop as soon
191 as we have run each shrink pass and seen no progress on any of
192 them.
194 This means that e.g. it's fine to try each of N deletions
195 or replacements in a random order, but it's not OK to try N random
196 deletions (unless you have already shrunk at least once, though we
197 don't currently take advantage of this loophole).
199 Shrink passes need to be written so as to be robust against
200 change in the underlying shrink target. It is generally safe
201 to assume that the shrink target does not change prior to the
202 point of first modification - e.g. if you change no bytes at
203 index ``i``, all spans whose start is ``<= i`` still exist,
204 as do all blocks, and the data object is still of length
205 ``>= i + 1``. This can only be violated by bad user code which
206 relies on an external source of non-determinism.
208 When the underlying shrink_target changes, shrink
209 passes should not run substantially more test_function calls
210 on success than they do on failure. Say, no more than a constant
211 factor more. In particular shrink passes should not iterate to a
212 fixed point.
214 This means that shrink passes are often written with loops that
215 are carefully designed to do the right thing in the case that no
216 shrinks occurred and try to adapt to any changes to do a reasonable
217 job. e.g. say we wanted to write a shrink pass that tried deleting
218 each individual choice (this isn't an especially good pass,
219 but it leads to a simple illustrative example), we might do it
220 by iterating over the choice sequence like so:
222 .. code-block:: python
224 i = 0
225 while i < len(self.shrink_target.nodes):
226 if not self.consider_new_nodes(
227 self.shrink_target.nodes[:i] + self.shrink_target.nodes[i + 1 :]
228 ):
229 i += 1
231 The reason for writing the loop this way is that i is always a
232 valid index into the current choice sequence, even if the current sequence
233 changes as a result of our actions. When the choice sequence changes,
234 we leave the index where it is rather than restarting from the
235 beginning, and carry on. This means that the number of steps we
236 run in this case is always bounded above by the number of steps
237 we would run if nothing works.
239 Another thing to bear in mind about shrink pass design is that
240 they should prioritise *progress*. If you have N operations that
241 you need to run, you should try to order them in such a way as
242 to avoid stalling, where you have long periods of test function
243 invocations where no shrinks happen. This is bad because whenever
244 we shrink we reduce the amount of work the shrinker has to do
245 in future, and often speed up the test function, so we ideally
246 wanted those shrinks to happen much earlier in the process.
248 Sometimes stalls are inevitable of course - e.g. if the pass
249 makes no progress, then the entire thing is just one long stall,
250 but it's helpful to design it so that stalls are less likely
251 in typical behaviour.
253 The two easiest ways to do this are:
255 * Just run the N steps in random order. As long as a
256 reasonably large proportion of the operations succeed, this
257 guarantees the expected stall length is quite short. The
258 book keeping for making sure this does the right thing when
259 it succeeds can be quite annoying.
260 * When you have any sort of nested loop, loop in such a way
261 that both loop variables change each time. This prevents
262 stalls which occur when one particular value for the outer
263 loop is impossible to make progress on, rendering the entire
264 inner loop into a stall.
266 However, although progress is good, too much progress can be
267 a bad sign! If you're *only* seeing successful reductions,
268 that's probably a sign that you are making changes that are
269 too timid. Two useful things to offset this:
271 * It's worth writing shrink passes which are *adaptive*, in
272 the sense that when operations seem to be working really
273 well we try to bundle multiple of them together. This can
274 often be used to turn what would be O(m) successful calls
275 into O(log(m)).
276 * It's often worth trying one or two special minimal values
277 before trying anything more fine grained (e.g. replacing
278 the whole thing with zero).
280 """
282 def derived_value(fn):
283 """It's useful during shrinking to have access to derived values of
284 the current shrink target.
286 This decorator allows you to define these as cached properties. They
287 are calculated once, then cached until the shrink target changes, then
288 recalculated the next time they are used."""
290 def accept(self):
291 try:
292 return self.__derived_values[fn.__name__]
293 except KeyError:
294 return self.__derived_values.setdefault(fn.__name__, fn(self))
296 accept.__name__ = fn.__name__
297 return property(accept)
299 def __init__(
300 self,
301 engine: "ConjectureRunner",
302 initial: ConjectureData | ConjectureResult,
303 predicate: ShrinkPredicateT | None,
304 *,
305 allow_transition: (
306 Callable[[ConjectureData | ConjectureResult, ConjectureData], bool] | None
307 ),
308 explain: bool,
309 in_target_phase: bool = False,
310 ):
311 """Create a shrinker for a particular engine, with a given starting
312 point and predicate. When shrink() is called it will attempt to find an
313 example for which predicate is True and which is strictly smaller than
314 initial.
316 Note that initial is a ConjectureData object, and predicate
317 takes ConjectureData objects.
318 """
319 assert predicate is not None or allow_transition is not None
320 self.engine = engine
321 self.__predicate = predicate or (lambda data: True)
322 self.__allow_transition = allow_transition or (lambda source, destination: True)
323 self.__derived_values: dict = {}
325 self.initial_size = len(initial.choices)
326 # We keep track of the current best example on the shrink_target
327 # attribute.
328 self.shrink_target = initial
329 self.clear_change_tracking()
330 self.shrinks = 0
332 # We terminate shrinks that seem to have reached their logical
333 # conclusion: If we've called the underlying test function at
334 # least self.max_stall times since the last time we shrunk,
335 # it's time to stop shrinking.
336 self.max_stall = 200
337 self.initial_calls = self.engine.call_count
338 self.initial_misaligned = self.engine.misaligned_count
339 self.calls_at_last_shrink = self.initial_calls
341 self.shrink_passes: list[ShrinkPass] = [
342 ShrinkPass(self.try_trivial_spans),
343 self.node_program("X" * 5),
344 self.node_program("X" * 4),
345 self.node_program("X" * 3),
346 self.node_program("X" * 2),
347 self.node_program("X" * 1),
348 ShrinkPass(self.pass_to_descendant),
349 ShrinkPass(self.reorder_spans),
350 ShrinkPass(self.minimize_duplicated_choices),
351 ShrinkPass(self.minimize_individual_choices),
352 ShrinkPass(self.redistribute_numeric_pairs),
353 ShrinkPass(self.lower_integers_together),
354 ShrinkPass(self.lower_duplicated_characters),
355 ShrinkPass(self.normalize_unicode_chars),
356 ]
358 # Because the shrinker is also used to `pareto_optimise` in the target phase,
359 # we sometimes want to allow extending buffers instead of aborting at the end.
360 self.__extend: Literal["full"] | int = "full" if in_target_phase else 0
361 self.should_explain = explain
363 @derived_value # type: ignore
364 def cached_calculations(self):
365 return {}
367 def cached(self, *keys):
368 def accept(f):
369 cache_key = (f.__name__, *keys)
370 try:
371 return self.cached_calculations[cache_key]
372 except KeyError:
373 return self.cached_calculations.setdefault(cache_key, f())
375 return accept
377 @property
378 def calls(self) -> int:
379 """Return the number of calls that have been made to the underlying
380 test function."""
381 return self.engine.call_count
383 @property
384 def misaligned(self) -> int:
385 return self.engine.misaligned_count
387 def check_calls(self) -> None:
388 if self.calls - self.calls_at_last_shrink >= self.max_stall:
389 raise StopShrinking
391 def cached_test_function(
392 self, nodes: Sequence[ChoiceNode]
393 ) -> tuple[bool, ConjectureResult | _Overrun | None]:
394 nodes = nodes[: len(self.nodes)]
396 if startswith(nodes, self.nodes):
397 return (True, None)
399 if sort_key(self.nodes) < sort_key(nodes):
400 return (False, None)
402 # sometimes our shrinking passes try obviously invalid things. We handle
403 # discarding them in one place here.
404 if any(not choice_permitted(node.value, node.constraints) for node in nodes):
405 return (False, None)
407 result = self.engine.cached_test_function(
408 [n.value for n in nodes], extend=self.__extend
409 )
410 previous = self.shrink_target
411 self.incorporate_test_data(result)
412 self.check_calls()
413 return (previous is not self.shrink_target, result)
415 def consider_new_nodes(self, nodes: Sequence[ChoiceNode]) -> bool:
416 return self.cached_test_function(nodes)[0]
418 def incorporate_test_data(self, data):
419 """Takes a ConjectureData or Overrun object updates the current
420 shrink_target if this data represents an improvement over it."""
421 if data.status < Status.VALID or data is self.shrink_target:
422 return
423 if (
424 self.__predicate(data)
425 and sort_key(data.nodes) < sort_key(self.shrink_target.nodes)
426 and self.__allow_transition(self.shrink_target, data)
427 ):
428 self.update_shrink_target(data)
430 def debug(self, msg: str) -> None:
431 self.engine.debug(msg)
433 @property
434 def random(self) -> "Random":
435 return self.engine.random
437 def shrink(self) -> None:
438 """Run the full set of shrinks and update shrink_target.
440 This method is "mostly idempotent" - calling it twice is unlikely to
441 have any effect, though it has a non-zero probability of doing so.
442 """
444 try:
445 self.initial_coarse_reduction()
446 self.greedy_shrink()
447 except StopShrinking:
448 # If we stopped shrinking because we're making slow progress (instead of
449 # reaching a local optimum), don't run the explain-phase logic.
450 self.should_explain = False
451 finally:
452 if self.engine.report_debug_info:
454 def s(n):
455 return "s" if n != 1 else ""
457 total_deleted = self.initial_size - len(self.shrink_target.choices)
458 calls = self.engine.call_count - self.initial_calls
459 misaligned = self.engine.misaligned_count - self.initial_misaligned
461 self.debug(
462 "---------------------\n"
463 "Shrink pass profiling\n"
464 "---------------------\n\n"
465 f"Shrinking made a total of {calls} call{s(calls)} of which "
466 f"{self.shrinks} shrank and {misaligned} were misaligned. This "
467 f"deleted {total_deleted} choices out of {self.initial_size}."
468 )
469 for useful in [True, False]:
470 self.debug("")
471 if useful:
472 self.debug("Useful passes:")
473 else:
474 self.debug("Useless passes:")
475 self.debug("")
476 for pass_ in sorted(
477 self.shrink_passes,
478 key=lambda t: (-t.calls, t.deletions, t.shrinks),
479 ):
480 if pass_.calls == 0:
481 continue
482 if (pass_.shrinks != 0) != useful:
483 continue
485 self.debug(
486 f" * {pass_.name} made {pass_.calls} call{s(pass_.calls)} of which "
487 f"{pass_.shrinks} shrank and {pass_.misaligned} were misaligned, "
488 f"deleting {pass_.deletions} choice{s(pass_.deletions)}."
489 )
490 self.debug("")
491 self.explain()
493 def explain(self) -> None:
494 if not self.should_explain or not self.shrink_target.arg_spans:
495 return
496 with self.engine._log_phase_statistics("explain"):
497 self._explain()
499 def _explain(self) -> None:
500 self.max_stall = 2**100
501 shrink_target = self.shrink_target
502 nodes = self.nodes
503 choices = self.choices
504 spans = self.shrink_target.spans
505 chunks: dict[int, list[tuple[ChoiceT, ...]]] = defaultdict(list)
507 # The node ranges of the spans we may vary. Multiple arg spans can share
508 # a range (e.g. builds() around a single strategy), in which case only
509 # the first processed gets a comment and the rest are skipped below.
510 arg_ranges = {
511 span_index: (spans[span_index].start, spans[span_index].end)
512 for span_index in self.shrink_target.arg_spans
513 }
515 # Before we start running experiments, let's check for known inputs which would
516 # make them redundant. The shrinking process means that we've already tried many
517 # variations on the minimal test case, so this can save a lot of time.
518 seen_passing_seq = self.engine.passing_choice_sequences(
519 prefix=self.nodes[: min(start for start, _ in arg_ranges.values())]
520 )
522 # Now that we've shrunk to a minimal failing test case, it's time to try
523 # varying each part that we've noted will go in the final report. Consider
524 # spans in largest-first order
525 for span_index, (start, end) in sorted(
526 arg_ranges.items(), key=lambda kv: (-(kv[1][1] - kv[1][0]), kv[1], kv[0])
527 ):
528 # Skip spans where nothing can in fact vary: ranges that are empty
529 # (e.g. a just() draw) or consist entirely of forced choices. The
530 # "or any other generated value" comment would be false for them.
531 if all(nodes[i].was_forced for i in range(start, end)):
532 continue
534 # Check for any previous test cases that match the prefix and suffix,
535 # so we can skip if we found a passing example while shrinking.
536 if any(
537 startswith(seen, nodes[:start]) and endswith(seen, nodes[end:])
538 for seen in seen_passing_seq
539 ):
540 continue
542 # Skip spans whose ranges are subsets of already-explained ranges.
543 # If a larger range can vary freely, so can its sub-ranges.
544 if any(
545 spans[c].start <= start and end <= spans[c].end
546 for c in self.shrink_target.span_comments
547 if c is not None
548 ):
549 continue
551 # Try a few targeted candidates before falling back to random sampling,
552 # so that simple cases like ``assert n1 == n2`` -- where the only
553 # passing value of ``n1`` is exactly ``n2``'s value -- aren't reported
554 # as freely-variable just because random sampling missed it.
555 candidates = list(self._explain_candidates(start, end))
557 # Run our experiments
558 n_same_failures = 0
559 note = "or any other generated value"
560 # TODO: is 100 same-failures out of 500 attempts a good heuristic?
561 for n_attempt in range(500 + len(candidates)): # pragma: no branch
562 # no-branch here because we don't coverage-test the abort-at-500 logic.
564 if n_attempt - 10 - len(candidates) > n_same_failures * 5:
565 # stop early if we're seeing mostly invalid examples
566 break # pragma: no cover
568 if n_attempt < len(candidates):
569 replacement = list(candidates[n_attempt])
570 else:
571 # replace start:end with random values
572 replacement = []
573 for i in range(start, end):
574 node = nodes[i]
575 if not node.was_forced:
576 value = draw_choice(
577 node.type, node.constraints, random=self.random
578 )
579 node = node.copy(with_value=value)
580 replacement.append(node.value)
582 attempt = choices[:start] + tuple(replacement) + choices[end:]
583 result = self.engine.cached_test_function(attempt, extend="full")
585 if result.status is Status.OVERRUN:
586 continue # pragma: no cover # flakily covered
587 result = cast(ConjectureResult, result)
588 if not (
589 len(attempt) == len(result.choices)
590 and endswith(result.nodes, nodes[end:])
591 ):
592 # Turns out this was a variable-length part, so grab the infix...
593 for span1, span2 in zip(
594 shrink_target.spans, result.spans, strict=False
595 ):
596 assert span1.start == span2.start
597 assert span1.start <= start
598 if span1.start == start and span1.end == end:
599 result_end = span2.end
600 break
601 else:
602 raise NotImplementedError("Expected matching prefixes")
604 attempt = (
605 choices[:start]
606 + result.choices[start:result_end]
607 + choices[end:]
608 )
609 chunks[span_index].append(result.choices[start:result_end])
610 result = self.engine.cached_test_function(attempt)
612 if result.status is Status.OVERRUN:
613 continue # pragma: no cover # flakily covered
614 result = cast(ConjectureResult, result)
615 else:
616 chunks[span_index].append(result.choices[start:end])
618 if shrink_target is not self.shrink_target: # pragma: no cover
619 # If we've shrunk further without meaning to, bail out.
620 self.shrink_target.span_comments.clear()
621 return
622 if result.status is Status.VALID:
623 # The test passed, indicating that this param can't vary freely.
624 # However, it's really hard to write a simple and reliable covering
625 # test, because of our `seen_passing_buffers` check above.
626 break # pragma: no cover
627 if self.__predicate(result): # pragma: no branch
628 n_same_failures += 1
629 if n_same_failures >= 100:
630 self.shrink_target.span_comments[span_index] = note
631 break
633 # Finally, if we've found multiple independently-variable parts, check whether
634 # they can all be varied together.
635 if len(self.shrink_target.span_comments) <= 1:
636 return
637 n_same_failures_together = 0
638 # Only include spans that actually got a comment
639 chunks_by_start_index = sorted(
640 (arg_ranges[k], v)
641 for k, v in chunks.items()
642 if k in self.shrink_target.span_comments
643 )
644 for _ in range(500): # pragma: no branch
645 # no-branch here because we don't coverage-test the abort-at-500 logic.
646 new_choices: list[ChoiceT] = []
647 prev_end = 0
648 for (start, end), ls in chunks_by_start_index:
649 assert prev_end <= start < end, "these chunks must be nonoverlapping"
650 new_choices.extend(choices[prev_end:start])
651 new_choices.extend(self.random.choice(ls))
652 prev_end = end
654 result = self.engine.cached_test_function(new_choices)
656 # This *can't* be a shrink because none of the components were.
657 assert shrink_target is self.shrink_target
658 if result.status == Status.VALID:
659 self.shrink_target.span_comments[None] = (
660 "The test sometimes passed when commented parts were varied together."
661 )
662 break # Test passed, this param can't vary freely.
663 if self.__predicate(result): # pragma: no branch
664 n_same_failures_together += 1
665 if n_same_failures_together >= 100:
666 self.shrink_target.span_comments[None] = (
667 "The test always failed when commented parts were varied together."
668 )
669 break
671 def _explain_candidates(
672 self, start: int, end: int
673 ) -> "Iterator[tuple[ChoiceT, ...]]":
674 """Yield deterministic candidate replacements for ``nodes[start:end]``.
676 Random sampling alone misses cases like ``assert n1 == n2``, where the
677 only passing value of ``n1`` is exactly ``n2``'s value. We try
678 substituting values from each other arg slice with matching length and
679 types, which catches such comparisons. Invalid borrowed values just
680 produce an irrelevant test result the outer loop discards.
681 """
682 nodes = self.nodes
683 spans = self.shrink_target.spans
684 target_types = tuple(nodes[i].type for i in range(start, end))
685 current_key = choices_key(tuple(nodes[i].value for i in range(start, end)))
686 seen: set[tuple[Any, ...]] = {current_key}
687 arg_ranges = sorted(
688 {(spans[i].start, spans[i].end) for i in self.shrink_target.arg_spans}
689 )
690 for start2, end2 in arg_ranges:
691 if (start2, end2) == (start, end) or (end2 - start2) != (end - start):
692 continue
693 if (
694 tuple(nodes[start2 + j].type for j in range(end - start))
695 != target_types
696 ):
697 continue
698 borrowed = tuple(nodes[start2 + j].value for j in range(end - start))
699 key = choices_key(borrowed)
700 if key in seen:
701 continue
702 seen.add(key)
703 yield borrowed
705 def greedy_shrink(self) -> None:
706 """Run a full set of greedy shrinks (that is, ones that will only ever
707 move to a better target) and update shrink_target appropriately.
709 This method iterates to a fixed point and so is idempontent - calling
710 it twice will have exactly the same effect as calling it once.
711 """
712 self.fixate_shrink_passes(self.shrink_passes)
714 def initial_coarse_reduction(self):
715 """Performs some preliminary reductions that should not be
716 repeated as part of the main shrink passes.
718 The main reason why these can't be included as part of shrink
719 passes is that they have much more ability to make the test
720 case "worse". e.g. they might rerandomise part of it, significantly
721 increasing the value of individual nodes, which works in direct
722 opposition to the lexical shrinking and will frequently undo
723 its work.
724 """
725 self.reduce_each_alternative()
727 @derived_value # type: ignore
728 def spans_starting_at(self):
729 result = [[] for _ in self.shrink_target.nodes]
730 for i, ex in enumerate(self.spans):
731 # We can have zero-length spans that start at the end
732 if ex.start < len(result):
733 result[ex.start].append(i)
734 return tuple(map(tuple, result))
736 def reduce_each_alternative(self):
737 """This is a pass that is designed to rerandomise use of the
738 one_of strategy or things that look like it, in order to try
739 to move from later strategies to earlier ones in the branch
740 order.
742 It does this by trying to systematically lower each value it
743 finds that looks like it might be the branch decision for
744 one_of, and then attempts to repair any changes in shape that
745 this causes.
746 """
747 i = 0
748 while i < len(self.shrink_target.nodes):
749 nodes = self.shrink_target.nodes
750 node = nodes[i]
751 if (
752 node.type == "integer"
753 and not node.was_forced
754 and node.value <= 10
755 and node.constraints["min_value"] == 0
756 ):
757 assert isinstance(node.value, int)
759 # We've found a plausible candidate for a ``one_of`` choice.
760 # We now want to see if the shape of the test case actually depends
761 # on it. If it doesn't, then we don't need to do this (comparatively
762 # costly) pass, and can let much simpler lexicographic reduction
763 # handle it later.
764 #
765 # We test this by trying to set the value to zero and seeing if the
766 # shape changes, as measured by either changing the number of subsequent
767 # nodes, or changing the nodes in such a way as to cause one of the
768 # previous values to no longer be valid in its position.
769 zero_attempt = self.cached_test_function(
770 nodes[:i] + (nodes[i].copy(with_value=0),) + nodes[i + 1 :]
771 )[1]
772 if (
773 zero_attempt is not self.shrink_target
774 and zero_attempt is not None
775 and zero_attempt.status >= Status.VALID
776 ):
777 changed_shape = len(zero_attempt.nodes) != len(nodes)
779 if not changed_shape:
780 for j in range(i + 1, len(nodes)):
781 zero_node = zero_attempt.nodes[j]
782 orig_node = nodes[j]
783 if (
784 zero_node.type != orig_node.type
785 or not choice_permitted(
786 orig_node.value, zero_node.constraints
787 )
788 ):
789 changed_shape = True
790 break
791 if changed_shape:
792 for v in range(node.value):
793 if self.try_lower_node_as_alternative(i, v):
794 break
795 i += 1
797 def try_lower_node_as_alternative(self, i, v):
798 """Attempt to lower `self.shrink_target.nodes[i]` to `v`,
799 while rerandomising and attempting to repair any subsequent
800 changes to the shape of the test case that this causes."""
801 nodes = self.shrink_target.nodes
802 if self.consider_new_nodes(
803 nodes[:i] + (nodes[i].copy(with_value=v),) + nodes[i + 1 :]
804 ):
805 return True
807 prefix = nodes[:i] + (nodes[i].copy(with_value=v),)
808 initial = self.shrink_target
809 spans = self.spans_starting_at[i]
810 for _ in range(3):
811 random_attempt = self.engine.cached_test_function(
812 [n.value for n in prefix], extend=len(nodes)
813 )
814 if random_attempt.status < Status.VALID:
815 continue
816 self.incorporate_test_data(random_attempt)
817 for j in spans:
818 initial_span = initial.spans[j]
819 attempt_span = random_attempt.spans[j]
820 contents = random_attempt.nodes[attempt_span.start : attempt_span.end]
821 self.consider_new_nodes(
822 nodes[:i] + contents + nodes[initial_span.end :]
823 )
824 if initial is not self.shrink_target:
825 return True
826 return False
828 @derived_value # type: ignore
829 def shrink_pass_choice_trees(self) -> dict[Any, ChoiceTree]:
830 return defaultdict(ChoiceTree)
832 def step(self, shrink_pass: ShrinkPass, *, random_order: bool = False) -> bool:
833 tree = self.shrink_pass_choice_trees[shrink_pass]
834 if tree.exhausted:
835 return False
837 initial_shrinks = self.shrinks
838 initial_calls = self.calls
839 initial_misaligned = self.misaligned
840 size = len(self.shrink_target.choices)
841 assert shrink_pass.name is not None
842 self.engine.explain_next_call_as(shrink_pass.name)
844 if random_order:
845 selection_order = random_selection_order(self.random)
846 else:
847 selection_order = prefix_selection_order(shrink_pass.last_prefix)
849 try:
850 shrink_pass.last_prefix = tree.step(
851 selection_order,
852 lambda chooser: shrink_pass.function(chooser),
853 )
854 finally:
855 shrink_pass.calls += self.calls - initial_calls
856 shrink_pass.misaligned += self.misaligned - initial_misaligned
857 shrink_pass.shrinks += self.shrinks - initial_shrinks
858 shrink_pass.deletions += size - len(self.shrink_target.choices)
859 self.engine.clear_call_explanation()
860 return True
862 def fixate_shrink_passes(self, passes: list[ShrinkPass]) -> None:
863 """Run steps from each pass in ``passes`` until the current shrink target
864 is a fixed point of all of them."""
865 any_ran = True
866 while any_ran:
867 any_ran = False
869 reordering = {}
871 # We run remove_discarded after every pass to do cleanup
872 # keeping track of whether that actually works. Either there is
873 # no discarded data and it is basically free, or it reliably works
874 # and deletes data, or it doesn't work. In that latter case we turn
875 # it off for the rest of this loop through the passes, but will
876 # try again once all of the passes have been run.
877 can_discard = self.remove_discarded()
879 calls_at_loop_start = self.calls
881 # We keep track of how many calls can be made by a single step
882 # without making progress and use this to test how much to pad
883 # out self.max_stall by as we go along.
884 max_calls_per_failing_step = 1
886 for sp in passes:
887 if can_discard:
888 can_discard = self.remove_discarded()
890 before_sp = self.shrink_target
892 # Run the shrink pass until it fails to make any progress
893 # max_failures times in a row. This implicitly boosts shrink
894 # passes that are more likely to work.
895 failures = 0
896 max_failures = 20
897 while failures < max_failures:
898 # We don't allow more than max_stall consecutive failures
899 # to shrink, but this means that if we're unlucky and the
900 # shrink passes are in a bad order where only the ones at
901 # the end are useful, if we're not careful this heuristic
902 # might stop us before we've tried everything. In order to
903 # avoid that happening, we make sure that there's always
904 # plenty of breathing room to make it through a single
905 # iteration of the fixate_shrink_passes loop.
906 self.max_stall = max(
907 self.max_stall,
908 2 * max_calls_per_failing_step
909 + (self.calls - calls_at_loop_start),
910 )
912 prev = self.shrink_target
913 initial_calls = self.calls
914 # It's better for us to run shrink passes in a deterministic
915 # order, to avoid repeat work, but this can cause us to create
916 # long stalls when there are a lot of steps which fail to do
917 # anything useful. In order to avoid this, once we've noticed
918 # we're in a stall (i.e. half of max_failures calls have failed
919 # to do anything) we switch to randomly jumping around. If we
920 # find a success then we'll resume deterministic order from
921 # there which, with any luck, is in a new good region.
922 if not self.step(sp, random_order=failures >= max_failures // 2):
923 # step returns False when there is nothing to do because
924 # the entire choice tree is exhausted. If this happens
925 # we break because we literally can't run this pass any
926 # more than we already have until something else makes
927 # progress.
928 break
929 any_ran = True
931 # Don't count steps that didn't actually try to do
932 # anything as failures. Otherwise, this call is a failure
933 # if it failed to make any changes to the shrink target.
934 if initial_calls != self.calls:
935 if prev is not self.shrink_target:
936 failures = 0
937 else:
938 max_calls_per_failing_step = max(
939 max_calls_per_failing_step, self.calls - initial_calls
940 )
941 failures += 1
943 # We reorder the shrink passes so that on our next run through
944 # we try good ones first. The rule is that shrink passes that
945 # did nothing useful are the worst, shrink passes that reduced
946 # the length are the best.
947 if self.shrink_target is before_sp:
948 reordering[sp] = 1
949 elif len(self.choices) < len(before_sp.choices):
950 reordering[sp] = -1
951 else:
952 reordering[sp] = 0
954 passes.sort(key=reordering.__getitem__)
956 @property
957 def nodes(self) -> tuple[ChoiceNode, ...]:
958 return self.shrink_target.nodes
960 @property
961 def choices(self) -> tuple[ChoiceT, ...]:
962 return self.shrink_target.choices
964 @property
965 def spans(self) -> Spans:
966 return self.shrink_target.spans
968 @derived_value # type: ignore
969 def spans_by_label(self):
970 """
971 A mapping of labels to a list of spans with that label. Spans in the list
972 are ordered by their normal index order.
973 """
975 spans_by_label = defaultdict(list)
976 for ex in self.spans:
977 spans_by_label[ex.label].append(ex)
978 return dict(spans_by_label)
980 @derived_value # type: ignore
981 def distinct_labels(self):
982 return sorted(self.spans_by_label, key=str)
984 def pass_to_descendant(self, chooser):
985 """Attempt to replace each span with a descendant span.
987 This is designed to deal with strategies that call themselves
988 recursively. For example, suppose we had:
990 binary_tree = st.deferred(
991 lambda: st.one_of(
992 st.integers(), st.tuples(binary_tree, binary_tree)))
994 This pass guarantees that we can replace any binary tree with one of
995 its subtrees - each of those will create an interval that the parent
996 could validly be replaced with, and this pass will try doing that.
998 This is pretty expensive - it takes O(len(intervals)^2) - so we run it
999 late in the process when we've got the number of intervals as far down
1000 as possible.
1001 """
1003 label = chooser.choose(
1004 self.distinct_labels, lambda l: len(self.spans_by_label[l]) >= 2
1005 )
1007 spans = self.spans_by_label[label]
1008 i = chooser.choose(range(len(spans) - 1))
1009 ancestor = spans[i]
1011 if i + 1 == len(spans) or spans[i + 1].start >= ancestor.end:
1012 return
1014 @self.cached(label, i)
1015 def descendants():
1016 lo = i + 1
1017 hi = len(spans)
1018 while lo + 1 < hi:
1019 mid = (lo + hi) // 2
1020 if spans[mid].start >= ancestor.end:
1021 hi = mid
1022 else:
1023 lo = mid
1024 return [
1025 span
1026 for span in spans[i + 1 : hi]
1027 if span.choice_count < ancestor.choice_count
1028 ]
1030 descendant = chooser.choose(descendants, lambda ex: ex.choice_count > 0)
1032 assert ancestor.start <= descendant.start
1033 assert ancestor.end >= descendant.end
1034 assert descendant.choice_count < ancestor.choice_count
1036 self.consider_new_nodes(
1037 self.nodes[: ancestor.start]
1038 + self.nodes[descendant.start : descendant.end]
1039 + self.nodes[ancestor.end :]
1040 )
1042 def lower_common_node_offset(self):
1043 """Sometimes we find ourselves in a situation where changes to one part
1044 of the choice sequence unlock changes to other parts. Sometimes this is
1045 good, but sometimes this can cause us to exhibit exponential slow
1046 downs!
1048 e.g. suppose we had the following:
1050 m = draw(integers(min_value=0))
1051 n = draw(integers(min_value=0))
1052 assert abs(m - n) > 1
1054 If this fails then we'll end up with a loop where on each iteration we
1055 reduce each of m and n by 2 - m can't go lower because of n, then n
1056 can't go lower because of m.
1058 This will take us O(m) iterations to complete, which is exponential in
1059 the data size, as we gradually zig zag our way towards zero.
1061 This can only happen if we're failing to reduce the size of the choice
1062 sequence: The number of iterations that reduce the length of the choice
1063 sequence is bounded by that length.
1065 So what we do is this: We keep track of which nodes are changing, and
1066 then if there's some non-zero common offset to them we try and minimize
1067 them all at once by lowering that offset.
1069 This may not work, and it definitely won't get us out of all possible
1070 exponential slow downs (an example of where it doesn't is where the
1071 shape of the nodes changes as a result of this bouncing behaviour),
1072 but it fails fast when it doesn't work and gets us out of a really
1073 nastily slow case when it does.
1074 """
1075 if len(self.__changed_nodes) <= 1:
1076 return
1078 changed = []
1079 for i in sorted(self.__changed_nodes):
1080 node = self.nodes[i]
1081 if node.trivial or node.type != "integer":
1082 continue
1083 changed.append(node)
1085 if not changed:
1086 return
1088 ints = [
1089 abs(node.value - node.constraints["shrink_towards"]) for node in changed
1090 ]
1091 offset = min(ints)
1092 assert offset > 0
1094 for i in range(len(ints)):
1095 ints[i] -= offset
1097 st = self.shrink_target
1099 def offset_node(node, n):
1100 return (
1101 node.index,
1102 node.index + 1,
1103 [node.copy(with_value=node.constraints["shrink_towards"] + n)],
1104 )
1106 def consider(n, sign):
1107 return self.consider_new_nodes(
1108 replace_all(
1109 st.nodes,
1110 [
1111 offset_node(node, sign * (n + v))
1112 for node, v in zip(changed, ints, strict=False)
1113 ],
1114 )
1115 )
1117 # shrink from both sides
1118 Integer.shrink(offset, lambda n: consider(n, 1))
1119 Integer.shrink(offset, lambda n: consider(n, -1))
1120 self.clear_change_tracking()
1122 def clear_change_tracking(self):
1123 self.__last_checked_changed_at = self.shrink_target
1124 self.__all_changed_nodes = set()
1126 def mark_changed(self, i):
1127 self.__changed_nodes.add(i)
1129 @property
1130 def __changed_nodes(self) -> set[int]:
1131 if self.__last_checked_changed_at is self.shrink_target:
1132 return self.__all_changed_nodes
1134 prev_target = self.__last_checked_changed_at
1135 new_target = self.shrink_target
1136 assert prev_target is not new_target
1137 prev_nodes = prev_target.nodes
1138 new_nodes = new_target.nodes
1139 assert sort_key(new_target.nodes) < sort_key(prev_target.nodes)
1141 if len(prev_nodes) != len(new_nodes) or any(
1142 n1.type != n2.type for n1, n2 in zip(prev_nodes, new_nodes, strict=True)
1143 ):
1144 # should we check constraints are equal as well?
1145 self.__all_changed_nodes = set()
1146 else:
1147 assert len(prev_nodes) == len(new_nodes)
1148 for i, (n1, n2) in enumerate(zip(prev_nodes, new_nodes, strict=True)):
1149 assert n1.type == n2.type
1150 if not choice_equal(n1.value, n2.value):
1151 self.__all_changed_nodes.add(i)
1153 return self.__all_changed_nodes
1155 def update_shrink_target(self, new_target):
1156 assert isinstance(new_target, ConjectureResult)
1157 self.shrinks += 1
1158 # If we are just taking a long time to shrink we don't want to
1159 # trigger this heuristic, so whenever we shrink successfully
1160 # we give ourselves a bit of breathing room to make sure we
1161 # would find a shrink that took that long to find the next time.
1162 # The case where we're taking a long time but making steady
1163 # progress is handled by `finish_shrinking_deadline` in engine.py
1164 self.max_stall = max(
1165 self.max_stall, (self.calls - self.calls_at_last_shrink) * 2
1166 )
1167 self.calls_at_last_shrink = self.calls
1168 self.shrink_target = new_target
1169 self.__derived_values = {}
1171 def try_shrinking_nodes(self, nodes, n):
1172 """Attempts to replace each node in the nodes list with n. Returns
1173 True if it succeeded (which may include some additional modifications
1174 to shrink_target).
1176 In current usage it is expected that each of the nodes currently have
1177 the same value and choice_type, although this is not essential. Note that
1178 n must be < the node at min(nodes) or this is not a valid shrink.
1180 This method will attempt to do some small amount of work to delete data
1181 that occurs after the end of the nodes. This is useful for cases where
1182 there is some size dependency on the value of a node.
1183 """
1184 # If the length of the shrink target has changed from under us such that
1185 # the indices are out of bounds, give up on the replacement.
1186 # TODO_BETTER_SHRINK: we probably want to narrow down the root cause here at some point.
1187 if any(node.index >= len(self.nodes) for node in nodes):
1188 return # pragma: no cover
1190 initial_attempt = replace_all(
1191 self.nodes,
1192 [(node.index, node.index + 1, [node.copy(with_value=n)]) for node in nodes],
1193 )
1195 attempt = self.cached_test_function(initial_attempt)[1]
1197 if attempt is None:
1198 return False
1200 if attempt is self.shrink_target:
1201 # if the initial shrink was a success, try lowering offsets.
1202 self.lower_common_node_offset()
1203 return True
1205 # If this produced something completely invalid we ditch it
1206 # here rather than trying to persevere.
1207 if attempt.status is Status.OVERRUN:
1208 # Lowering a size-controlling choice can make the realigned (and
1209 # now boring) collection stop triggering the failure, so the test
1210 # draws further and overruns before we see the realignment -- this
1211 # is common in stateful tests, where a non-failing step is followed
1212 # by more steps. Re-run without the length limit to recover the
1213 # realigned tree, which the repair logic below can then act on.
1214 attempt = self.engine.cached_test_function(
1215 [n.value for n in initial_attempt], extend="full"
1216 )
1217 if attempt.status is Status.OVERRUN:
1218 return False
1220 if attempt.status is Status.INVALID:
1221 return False
1223 # When we lower a choice that controls the size of a later collection,
1224 # eg
1225 #
1226 # n = data.draw_integer()
1227 # s = data.draw_string(min_size=n, max_size=n)
1228 #
1229 # the recorded value for that collection no longer fits the constraints
1230 # the test function actually used, so the engine realigns the tree by
1231 # substituting a freshly-generated (simplest) value -- discarding
1232 # whatever made the collection interesting. (We can't rely on
1233 # ``attempt.misaligned_at`` to detect this, because the realigned choice
1234 # sequence is often independently cached as an ordinary, non-misaligned
1235 # result.) We detect a string/bytes node whose recorded value is now too
1236 # long, and retry with it truncated to fit. We try preserving content
1237 # from either end, since the interesting part may be at the start or the
1238 # end (see test_can_shrink_variable_string_draws).
1239 for i in range(min(len(initial_attempt), len(attempt.nodes))):
1240 node = initial_attempt[i]
1241 attempt_node = attempt.nodes[i]
1242 if (
1243 node.type == attempt_node.type
1244 and node.type in {"string", "bytes"}
1245 and not node.was_forced
1246 and len(node.value) > attempt_node.constraints["max_size"]
1247 ):
1248 max_size = attempt_node.constraints["max_size"]
1249 for truncated in (node.value[:max_size], node.value[-max_size:]):
1250 if self.consider_new_nodes(
1251 initial_attempt[:i]
1252 + [
1253 node.copy(
1254 with_constraints=attempt_node.constraints,
1255 with_value=truncated,
1256 )
1257 ]
1258 + initial_attempt[i + 1 :]
1259 ):
1260 return True
1262 lost_nodes = len(self.nodes) - len(attempt.nodes)
1263 if lost_nodes <= 0:
1264 return False
1266 start = nodes[0].index
1267 end = nodes[-1].index + 1
1268 # We now look for contiguous regions to delete that might help fix up
1269 # this failed shrink. We only look for contiguous regions of the right
1270 # lengths because doing anything more than that starts to get very
1271 # expensive. See minimize_individual_choices for where we
1272 # try to be more aggressive.
1273 regions_to_delete = {(end, end + lost_nodes)}
1275 for ex in self.spans:
1276 if ex.start > start:
1277 continue
1278 if ex.end <= end:
1279 continue
1281 if ex.index >= len(attempt.spans):
1282 continue # pragma: no cover
1284 replacement = attempt.spans[ex.index]
1285 in_original = [c for c in ex.children if c.start >= end]
1286 in_replaced = [c for c in replacement.children if c.start >= end]
1288 if len(in_replaced) >= len(in_original) or not in_replaced:
1289 continue
1291 # We've found a span where some of the children went missing
1292 # as a result of this change, and just replacing it with the data
1293 # it would have had and removing the spillover didn't work. This
1294 # means that some of its children towards the right must be
1295 # important, so we try to arrange it so that it retains its
1296 # rightmost children instead of its leftmost.
1297 regions_to_delete.add(
1298 (in_original[0].start, in_original[-len(in_replaced)].start)
1299 )
1301 for u, v in sorted(regions_to_delete, key=lambda x: x[1] - x[0], reverse=True):
1302 try_with_deleted = initial_attempt[:u] + initial_attempt[v:]
1303 if self.consider_new_nodes(try_with_deleted):
1304 return True
1306 return False
1308 def remove_discarded(self):
1309 """Try removing all nodes marked as discarded.
1311 This is primarily to deal with data that has been ignored while
1312 doing rejection sampling - e.g. as a result of an integer range, or a
1313 filtered strategy.
1315 Such data will also be handled by the ``node_program("X")`` deletion
1316 passes, but those are necessarily more conservative and will try
1317 deleting each contiguous run of nodes individually. The common case is
1318 that all data drawn and rejected can just be thrown away immediately in
1319 one block, so this pass will be much faster than trying each one
1320 individually when it works.
1322 returns False if there is discarded data and removing it does not work,
1323 otherwise returns True.
1324 """
1325 while self.shrink_target.has_discards:
1326 discarded = []
1328 for ex in self.shrink_target.spans:
1329 if (
1330 ex.choice_count > 0
1331 and ex.discarded
1332 and (not discarded or ex.start >= discarded[-1][-1])
1333 ):
1334 discarded.append((ex.start, ex.end))
1336 # This can happen if we have discards but they are all of
1337 # zero length. This shouldn't happen very often so it's
1338 # faster to check for it here than at the point of example
1339 # generation.
1340 if not discarded:
1341 break
1343 attempt = list(self.nodes)
1344 for u, v in reversed(discarded):
1345 del attempt[u:v]
1347 if not self.consider_new_nodes(tuple(attempt)):
1348 return False
1349 return True
1351 @derived_value # type: ignore
1352 def duplicated_nodes(self):
1353 """Returns a list of nodes grouped (choice_type, value)."""
1354 duplicates = defaultdict(list)
1355 for node in self.nodes:
1356 duplicates[(node.type, choice_key(node.value))].append(node)
1357 return list(duplicates.values())
1359 def node_program(self, program: str) -> ShrinkPass:
1360 return ShrinkPass(
1361 lambda chooser: self._node_program(chooser, program),
1362 name=f"node_program_{program}",
1363 )
1365 def _node_program(self, chooser, program):
1366 n = len(program)
1367 # Adaptively attempt to run the node program at the current
1368 # index. If this successfully applies the node program ``k`` times
1369 # then this runs in ``O(log(k))`` test function calls.
1370 i = chooser.choose(range(len(self.nodes) - n + 1))
1372 # First, run the node program at the chosen index. If this fails,
1373 # don't do any extra work, so that failure is as cheap as possible.
1374 if not self.run_node_program(i, program, original=self.shrink_target):
1375 return
1377 # Because we run in a random order we will often find ourselves in the middle
1378 # of a region where we could run the node program. We thus start by moving
1379 # left to the beginning of that region if possible in order to start from
1380 # the beginning of that region.
1381 def offset_left(k):
1382 return i - k * n
1384 i = offset_left(
1385 find_integer(
1386 lambda k: self.run_node_program(
1387 offset_left(k), program, original=self.shrink_target
1388 )
1389 )
1390 )
1392 original = self.shrink_target
1393 # Now try to run the node program multiple times here.
1394 find_integer(
1395 lambda k: self.run_node_program(i, program, original=original, repeats=k)
1396 )
1398 def minimize_duplicated_choices(self, chooser):
1399 """Find choices that have been duplicated in multiple places and attempt
1400 to minimize all of the duplicates simultaneously.
1402 This lets us handle cases where two values can't be shrunk
1403 independently of each other but can easily be shrunk together.
1404 For example if we had something like:
1406 ls = data.draw(lists(integers()))
1407 y = data.draw(integers())
1408 assert y not in ls
1410 Suppose we drew y = 3 and after shrinking we have ls = [3]. If we were
1411 to replace both 3s with 0, this would be a valid shrink, but if we were
1412 to replace either 3 with 0 on its own the test would start passing.
1414 It is also useful for when that duplication is accidental and the value
1415 of the choices don't matter very much because it allows us to replace
1416 more values at once.
1417 """
1418 nodes = chooser.choose(self.duplicated_nodes)
1419 # we can't lower any nodes which are trivial. try proceeding with the
1420 # remaining nodes.
1421 nodes = [node for node in nodes if not node.trivial]
1422 if len(nodes) <= 1:
1423 return
1425 self.minimize_nodes(nodes)
1427 def redistribute_numeric_pairs(self, chooser):
1428 """If there is a sum of generated numbers that we need their sum
1429 to exceed some bound, lowering one of them requires raising the
1430 other. This pass enables that."""
1432 # look for a pair of nodes (node1, node2) which are both numeric
1433 # and aren't separated by too many other nodes. We'll decrease node1 and
1434 # increase node2 (note that the other way around doesn't make sense as
1435 # it's strictly worse in the ordering).
1436 def can_choose_node(node):
1437 # don't choose nan, inf, or floats above the threshold where f + 1 > f
1438 # (which is not necessarily true for floats above MAX_PRECISE_INTEGER).
1439 # The motivation for the last condition is to avoid trying weird
1440 # non-shrinks where we raise one node and think we lowered another
1441 # (but didn't).
1442 return node.type in {"integer", "float"} and not (
1443 node.type == "float"
1444 and (math.isnan(node.value) or abs(node.value) >= MAX_PRECISE_INTEGER)
1445 )
1447 node1 = chooser.choose(
1448 self.nodes,
1449 lambda node: can_choose_node(node) and not node.trivial,
1450 )
1451 node2 = chooser.choose(
1452 self.nodes,
1453 lambda node: (
1454 can_choose_node(node)
1455 # Note that it's fine for node2 to be trivial, because we're going to
1456 # explicitly make it *not* trivial by adding to its value.
1457 and not node.was_forced
1458 # to avoid quadratic behavior, scan ahead only a small amount for
1459 # the related node.
1460 and node1.index < node.index <= node1.index + 4
1461 ),
1462 )
1464 m: int | float = node1.value
1465 n: int | float = node2.value
1467 def boost(k: int) -> bool:
1468 # floats always shrink towards 0
1469 shrink_towards = (
1470 node1.constraints["shrink_towards"] if node1.type == "integer" else 0
1471 )
1472 if k > abs(m - shrink_towards):
1473 return False
1475 # We are trying to move node1 (m) closer to shrink_towards, and node2
1476 # (n) farther away from shrink_towards. If m is below shrink_towards,
1477 # we want to add to m and subtract from n, and vice versa if above
1478 # shrink_towards.
1479 if m < shrink_towards:
1480 k = -k
1482 try:
1483 v1 = m - k
1484 v2 = n + k
1485 except OverflowError: # pragma: no cover
1486 # if n or m is a float and k is over sys.float_info.max, coercing
1487 # k to a float will overflow.
1488 return False
1490 # if we've increased node2 to the point that we're past max precision,
1491 # give up - things have become too unstable.
1492 if node1.type == "float" and abs(v2) >= MAX_PRECISE_INTEGER:
1493 return False
1495 return self.consider_new_nodes(
1496 self.nodes[: node1.index]
1497 + (node1.copy(with_value=v1),)
1498 + self.nodes[node1.index + 1 : node2.index]
1499 + (node2.copy(with_value=v2),)
1500 + self.nodes[node2.index + 1 :]
1501 )
1503 find_integer(boost)
1505 def lower_integers_together(self, chooser):
1506 node1 = chooser.choose(
1507 self.nodes, lambda n: n.type == "integer" and not n.trivial
1508 )
1509 # Search up to 3 nodes ahead, to avoid quadratic time.
1510 node2 = self.nodes[
1511 chooser.choose(
1512 range(node1.index + 1, min(len(self.nodes), node1.index + 3 + 1)),
1513 lambda i: (
1514 self.nodes[i].type == "integer" and not self.nodes[i].was_forced
1515 ),
1516 )
1517 ]
1519 # one might expect us to require node2 to be nontrivial, and to minimize
1520 # the node which is closer to its shrink_towards, rather than node1
1521 # unconditionally. In reality, it's acceptable for us to transition node2
1522 # from trivial to nontrivial, because the shrink ordering is dominated by
1523 # the complexity of the earlier node1. What matters is minimizing node1.
1524 shrink_towards = node1.constraints["shrink_towards"]
1526 def consider(n):
1527 return self.consider_new_nodes(
1528 self.nodes[: node1.index]
1529 + (node1.copy(with_value=node1.value - n),)
1530 + self.nodes[node1.index + 1 : node2.index]
1531 + (node2.copy(with_value=node2.value - n),)
1532 + self.nodes[node2.index + 1 :]
1533 )
1535 find_integer(lambda n: consider(shrink_towards - n))
1536 find_integer(lambda n: consider(n - shrink_towards))
1538 def lower_duplicated_characters(self, chooser):
1539 """
1540 Select two string choices no more than 4 choices apart and simultaneously
1541 lower characters which appear in both strings. This helps cases where the
1542 same character must appear in two strings, but the actual value of the
1543 character is not relevant.
1545 This shrinking pass currently only tries lowering *all* instances of the
1546 duplicated character in both strings. So for instance, given two choices:
1548 "bbac"
1549 "abbb"
1551 we would try lowering all five of the b characters simultaneously. This
1552 may fail to shrink some cases where only certain character indices are
1553 correlated, for instance if only the b at index 1 could be lowered
1554 simultaneously and the rest did in fact actually have to be a `b`.
1556 It would be nice to try shrinking that case as well, but we would need good
1557 safeguards because it could get very expensive to try all combinations.
1558 I expect lowering all duplicates to handle most cases in the meantime.
1559 """
1560 node1 = chooser.choose(
1561 self.nodes, lambda n: n.type == "string" and not n.trivial
1562 )
1564 # limit search to up to 4 choices ahead, to avoid quadratic behavior
1565 node2 = self.nodes[
1566 chooser.choose(
1567 range(node1.index + 1, min(len(self.nodes), node1.index + 1 + 4)),
1568 lambda i: (
1569 self.nodes[i].type == "string"
1570 and not self.nodes[i].trivial
1571 # select nodes which have at least one of the same character present
1572 and set(node1.value) & set(self.nodes[i].value)
1573 ),
1574 )
1575 ]
1577 duplicated_characters = set(node1.value) & set(node2.value)
1578 # deterministic ordering
1579 char = chooser.choose(sorted(duplicated_characters))
1580 intervals = node1.constraints["intervals"]
1582 def copy_node(node, n):
1583 # replace all duplicate characters in each string. This might miss
1584 # some shrinks compared to only replacing some, but trying all possible
1585 # combinations of indices could get expensive if done without some
1586 # thought.
1587 return node.copy(
1588 with_value=node.value.replace(char, intervals.char_in_shrink_order(n))
1589 )
1591 Integer.shrink(
1592 intervals.index_from_char_in_shrink_order(char),
1593 lambda n: self.consider_new_nodes(
1594 self.nodes[: node1.index]
1595 + (copy_node(node1, n),)
1596 + self.nodes[node1.index + 1 : node2.index]
1597 + (copy_node(node2, n),)
1598 + self.nodes[node2.index + 1 :]
1599 ),
1600 )
1602 def normalize_unicode_chars(self, chooser):
1603 """For string nodes, try replacing characters with simpler equivalents
1604 from natural text transformations: unicode decomposition (NFD, NFKD)
1605 and case mapping. For example, an accented latin letter is reduced
1606 to its base form, a ligature is reduced to its first base character,
1607 a mathematical alphanumeric symbol is reduced to its plain ascii
1608 counterpart, and a lowercase letter is replaced with its uppercase
1609 form (which has a smaller shrink-order index in the default
1610 alphabet).
1612 The codepoint shrinker is binary-search based, so it can get stuck on
1613 a high codepoint whose simpler equivalents aren't reached by halving
1614 / shifting / masking. This pass directly tries the natural simpler
1615 forms one character at a time.
1616 """
1617 node = chooser.choose(
1618 self.nodes,
1619 lambda n: n.type == "string"
1620 and any(
1621 _natural_simpler_chars(c, n.constraints["intervals"]) for c in n.value
1622 ),
1623 )
1624 intervals = node.constraints["intervals"]
1625 i = chooser.choose(
1626 range(len(node.value)),
1627 lambda j: bool(_natural_simpler_chars(node.value[j], intervals)),
1628 )
1629 for replacement in _natural_simpler_chars(node.value[i], intervals):
1630 new_value = node.value[:i] + replacement + node.value[i + 1 :]
1631 if self.consider_new_nodes(
1632 self.nodes[: node.index]
1633 + (node.copy(with_value=new_value),)
1634 + self.nodes[node.index + 1 :]
1635 ):
1636 return
1638 def minimize_nodes(self, nodes):
1639 choice_type = nodes[0].type
1640 value = nodes[0].value
1641 # unlike choice_type and value, constraints are *not* guaranteed to be equal among all
1642 # passed nodes. We arbitrarily use the constraints of the first node. I think
1643 # this is unsound (= leads to us trying shrinks that could not have been
1644 # generated), but those get discarded at test-time, and this enables useful
1645 # slips where constraints are not equal but are close enough that doing the
1646 # same operation on both basically just works.
1647 constraints = nodes[0].constraints
1648 assert all(
1649 node.type == choice_type and choice_equal(node.value, value)
1650 for node in nodes
1651 )
1653 if choice_type == "integer":
1654 shrink_towards = constraints["shrink_towards"]
1655 # try shrinking from both sides towards shrink_towards.
1656 # we're starting from n = abs(shrink_towards - value). Because the
1657 # shrinker will not check its starting value, we need to try
1658 # shrinking to n first.
1659 self.try_shrinking_nodes(nodes, abs(shrink_towards - value))
1660 Integer.shrink(
1661 abs(shrink_towards - value),
1662 lambda n: self.try_shrinking_nodes(nodes, shrink_towards + n),
1663 )
1664 Integer.shrink(
1665 abs(shrink_towards - value),
1666 lambda n: self.try_shrinking_nodes(nodes, shrink_towards - n),
1667 )
1668 elif choice_type == "float":
1669 self.try_shrinking_nodes(nodes, abs(value))
1670 Float.shrink(
1671 abs(value),
1672 lambda val: self.try_shrinking_nodes(nodes, val),
1673 )
1674 Float.shrink(
1675 abs(value),
1676 lambda val: self.try_shrinking_nodes(nodes, -val),
1677 )
1678 elif choice_type == "boolean":
1679 # must be True, otherwise would be trivial and not selected.
1680 assert value is True
1681 # only one thing to try: false!
1682 self.try_shrinking_nodes(nodes, False)
1683 elif choice_type == "bytes":
1684 Bytes.shrink(
1685 value,
1686 lambda val: self.try_shrinking_nodes(nodes, val),
1687 min_size=constraints["min_size"],
1688 )
1689 elif choice_type == "string":
1690 String.shrink(
1691 value,
1692 lambda val: self.try_shrinking_nodes(nodes, val),
1693 intervals=constraints["intervals"],
1694 min_size=constraints["min_size"],
1695 )
1696 else:
1697 raise NotImplementedError
1699 def try_trivial_spans(self, chooser):
1700 i = chooser.choose(range(len(self.spans)))
1702 prev = self.shrink_target
1703 nodes = self.shrink_target.nodes
1704 span = self.spans[i]
1705 prefix = nodes[: span.start]
1706 replacement = tuple(
1707 [
1708 (
1709 node
1710 if node.was_forced
1711 else node.copy(
1712 with_value=choice_from_index(0, node.type, node.constraints)
1713 )
1714 )
1715 for node in nodes[span.start : span.end]
1716 ]
1717 )
1718 suffix = nodes[span.end :]
1719 attempt = self.cached_test_function(prefix + replacement + suffix)[1]
1721 if self.shrink_target is not prev:
1722 return
1724 if isinstance(attempt, ConjectureResult):
1725 new_span = attempt.spans[i]
1726 new_replacement = attempt.nodes[new_span.start : new_span.end]
1727 self.consider_new_nodes(prefix + new_replacement + suffix)
1729 def minimize_individual_choices(self, chooser):
1730 """Attempt to minimize each choice in sequence.
1732 This is the pass that ensures that e.g. each integer we draw is a
1733 minimum value. So it's the part that guarantees that if we e.g. do
1735 x = data.draw(integers())
1736 assert x < 10
1738 then in our shrunk test case, x = 10 rather than say 97.
1740 If we are unsuccessful at minimizing a choice of interest we then
1741 check if that's because it's changing the size of the test case and,
1742 if so, we also make an attempt to delete parts of the test case to
1743 see if that fixes it.
1745 We handle most of the common cases in try_shrinking_nodes which is
1746 pretty good at clearing out large contiguous blocks of dead space,
1747 but it fails when there is data that has to stay in particular places
1748 in the list.
1749 """
1750 node = chooser.choose(self.nodes, lambda node: not node.trivial)
1751 initial_target = self.shrink_target
1753 self.minimize_nodes([node])
1754 if self.shrink_target is not initial_target:
1755 # the shrink target changed, so our shrink worked. Defer doing
1756 # anything more intelligent until this shrink fails.
1757 return
1759 # the shrink failed. One particularly common case where minimizing a
1760 # node can fail is the antipattern of drawing a size and then drawing a
1761 # collection of that size, or more generally when there is a size
1762 # dependency on some single node. We'll explicitly try and fix up this
1763 # common case here: if decreasing an integer node by one would reduce
1764 # the size of the generated input, we'll try deleting things after that
1765 # node and see if the resulting attempt works.
1767 if node.type != "integer":
1768 # Only try this fixup logic on integer draws. Almost all size
1769 # dependencies are on integer draws, and if it's not, it's doing
1770 # something convoluted enough that it is unlikely to shrink well anyway.
1771 # TODO: extent to floats? we probably currently fail on the following,
1772 # albeit convoluted example:
1773 # n = int(data.draw(st.floats()))
1774 # s = data.draw(st.lists(st.integers(), min_size=n, max_size=n))
1775 return
1777 lowered = (
1778 self.nodes[: node.index]
1779 + (node.copy(with_value=node.value - 1),)
1780 + self.nodes[node.index + 1 :]
1781 )
1782 attempt = self.cached_test_function(lowered)[1]
1783 if (
1784 attempt is None
1785 or attempt.status < Status.VALID
1786 or len(attempt.nodes) == len(self.nodes)
1787 or len(attempt.nodes) == node.index + 1
1788 ):
1789 # no point in trying our size-dependency-logic if our attempt at
1790 # lowering the node resulted in:
1791 # * an invalid conjecture data
1792 # * the same number of nodes as before
1793 # * no nodes beyond the lowered node (nothing to try to delete afterwards)
1794 return
1796 # If it were then the original shrink should have worked and we could
1797 # never have got here.
1798 assert attempt is not self.shrink_target
1800 @self.cached(node.index)
1801 def first_span_after_node():
1802 lo = 0
1803 hi = len(self.spans)
1804 while lo + 1 < hi:
1805 mid = (lo + hi) // 2
1806 span = self.spans[mid]
1807 if span.start >= node.index:
1808 hi = mid
1809 else:
1810 lo = mid
1811 return hi
1813 # we try deleting both entire spans, and single nodes.
1814 # If we wanted to get more aggressive, we could try deleting n
1815 # consecutive nodes (that don't cross a span boundary) for say
1816 # n <= 2 or n <= 3.
1817 if chooser.choose([True, False]):
1818 span = self.spans[
1819 chooser.choose(
1820 range(first_span_after_node, len(self.spans)),
1821 lambda i: self.spans[i].choice_count > 0,
1822 )
1823 ]
1824 self.consider_new_nodes(lowered[: span.start] + lowered[span.end :])
1825 else:
1826 node = self.nodes[chooser.choose(range(node.index + 1, len(self.nodes)))]
1827 self.consider_new_nodes(lowered[: node.index] + lowered[node.index + 1 :])
1829 def reorder_spans(self, chooser):
1830 """This pass allows us to reorder the children of each span.
1832 For example, consider the following:
1834 .. code-block:: python
1836 import hypothesis.strategies as st
1837 from hypothesis import given
1840 @given(st.text(), st.text())
1841 def test_not_equal(x, y):
1842 assert x != y
1844 Without the ability to reorder x and y this could fail either with
1845 ``x=""``, ``y="0"``, or the other way around. With reordering it will
1846 reliably fail with ``x=""``, ``y="0"``.
1847 """
1848 span = chooser.choose(self.spans)
1850 label = chooser.choose(span.children).label
1851 spans = [c for c in span.children if c.label == label]
1852 if len(spans) <= 1:
1853 return
1855 endpoints = [(span.start, span.end) for span in spans]
1856 st = self.shrink_target
1858 Ordering.shrink(
1859 range(len(spans)),
1860 lambda indices: self.consider_new_nodes(
1861 replace_all(
1862 st.nodes,
1863 [
1864 (
1865 u,
1866 v,
1867 st.nodes[spans[i].start : spans[i].end],
1868 )
1869 for (u, v), i in zip(endpoints, indices, strict=True)
1870 ],
1871 )
1872 ),
1873 key=lambda i: sort_key(st.nodes[spans[i].start : spans[i].end]),
1874 )
1876 def run_node_program(self, i, program, original, repeats=1):
1877 """Node programs are a mini-DSL for node rewriting, defined as a sequence
1878 of commands that can be run at some index into the nodes
1880 Commands are:
1882 * "X", delete this node
1884 This method runs the node program in ``program`` at node index
1885 ``i`` on the ConjectureData ``original``. If ``repeats > 1`` then it
1886 will attempt to approximate the results of running it that many times.
1888 Returns True if this successfully changes the underlying shrink target,
1889 else False.
1890 """
1891 if i + len(program) > len(original.nodes) or i < 0:
1892 return False
1893 attempt = list(original.nodes)
1894 for _ in range(repeats):
1895 for k, command in reversed(list(enumerate(program))):
1896 j = i + k
1897 if j >= len(attempt):
1898 return False
1900 if command == "X":
1901 del attempt[j]
1902 else:
1903 raise NotImplementedError(f"Unrecognised command {command!r}")
1905 return self.consider_new_nodes(attempt)