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