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 # 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 # note that sort_key can raise on invalid nodes, so we need to check
408 # choice_permitted first
409 if sort_key(self.nodes) < sort_key(nodes):
410 return (False, None)
412 result = self.engine.cached_test_function(
413 [n.value for n in nodes], extend=self.__extend
414 )
415 previous = self.shrink_target
416 self.incorporate_test_data(result)
417 self.check_calls()
418 return (previous is not self.shrink_target, result)
420 def consider_new_nodes(self, nodes: Sequence[ChoiceNode]) -> bool:
421 return self.cached_test_function(nodes)[0]
423 def incorporate_test_data(self, data):
424 """Takes a ConjectureData or Overrun object updates the current
425 shrink_target if this data represents an improvement over it."""
426 if data.status < Status.VALID or data is self.shrink_target:
427 return
428 if (
429 self.__predicate(data)
430 and sort_key(data.nodes) < sort_key(self.shrink_target.nodes)
431 and self.__allow_transition(self.shrink_target, data)
432 ):
433 self.update_shrink_target(data)
435 def debug(self, msg: str) -> None:
436 self.engine.debug(msg)
438 @property
439 def random(self) -> "Random":
440 return self.engine.random
442 def shrink(self) -> None:
443 """Run the full set of shrinks and update shrink_target.
445 This method is "mostly idempotent" - calling it twice is unlikely to
446 have any effect, though it has a non-zero probability of doing so.
447 """
449 try:
450 self.initial_coarse_reduction()
451 self.greedy_shrink()
452 except StopShrinking:
453 # If we stopped shrinking because we're making slow progress (instead of
454 # reaching a local optimum), don't run the explain-phase logic.
455 self.should_explain = False
456 finally:
457 if self.engine.report_debug_info:
459 def s(n):
460 return "s" if n != 1 else ""
462 total_deleted = self.initial_size - len(self.shrink_target.choices)
463 calls = self.engine.call_count - self.initial_calls
464 misaligned = self.engine.misaligned_count - self.initial_misaligned
466 self.debug(
467 "---------------------\n"
468 "Shrink pass profiling\n"
469 "---------------------\n\n"
470 f"Shrinking made a total of {calls} call{s(calls)} of which "
471 f"{self.shrinks} shrank and {misaligned} were misaligned. This "
472 f"deleted {total_deleted} choices out of {self.initial_size}."
473 )
474 for useful in [True, False]:
475 self.debug("")
476 if useful:
477 self.debug("Useful passes:")
478 else:
479 self.debug("Useless passes:")
480 self.debug("")
481 for pass_ in sorted(
482 self.shrink_passes,
483 key=lambda t: (-t.calls, t.deletions, t.shrinks),
484 ):
485 if pass_.calls == 0:
486 continue
487 if (pass_.shrinks != 0) != useful:
488 continue
490 self.debug(
491 f" * {pass_.name} made {pass_.calls} call{s(pass_.calls)} of which "
492 f"{pass_.shrinks} shrank and {pass_.misaligned} were misaligned, "
493 f"deleting {pass_.deletions} choice{s(pass_.deletions)}."
494 )
495 self.debug("")
496 self.explain()
498 def explain(self) -> None:
499 if not self.should_explain or not self.shrink_target.arg_spans:
500 return
501 with self.engine._log_phase_statistics("explain"):
502 self._explain()
504 def _explain(self) -> None:
505 self.max_stall = 2**100
506 shrink_target = self.shrink_target
507 nodes = self.nodes
508 choices = self.choices
509 spans = self.shrink_target.spans
510 chunks: dict[int, list[tuple[ChoiceT, ...]]] = defaultdict(list)
512 # The node ranges of the spans we may vary. Multiple arg spans can share
513 # a range (e.g. builds() around a single strategy), in which case only
514 # the first processed gets a comment and the rest are skipped below.
515 arg_ranges = {
516 span_index: (spans[span_index].start, spans[span_index].end)
517 for span_index in self.shrink_target.arg_spans
518 }
520 # Before we start running experiments, let's check for known inputs which would
521 # make them redundant. The shrinking process means that we've already tried many
522 # variations on the minimal test case, so this can save a lot of time.
523 seen_passing_seq = self.engine.passing_choice_sequences(
524 prefix=self.nodes[: min(start for start, _ in arg_ranges.values())]
525 )
527 # Now that we've shrunk to a minimal failing test case, it's time to try
528 # varying each part that we've noted will go in the final report. Consider
529 # spans in largest-first order
530 for span_index, (start, end) in sorted(
531 arg_ranges.items(), key=lambda kv: (-(kv[1][1] - kv[1][0]), kv[1], kv[0])
532 ):
533 # Skip spans where nothing can in fact vary: ranges that are empty
534 # (e.g. a just() draw) or consist entirely of forced choices. The
535 # "or any other generated value" comment would be false for them.
536 if all(nodes[i].was_forced for i in range(start, end)):
537 continue
539 # Check for any previous test cases that match the prefix and suffix,
540 # so we can skip if we found a passing test case while shrinking.
541 if any(
542 startswith(seen, nodes[:start]) and endswith(seen, nodes[end:])
543 for seen in seen_passing_seq
544 ):
545 continue
547 # Skip spans whose ranges are subsets of already-explained ranges.
548 # If a larger range can vary freely, so can its sub-ranges.
549 if any(
550 spans[c].start <= start and end <= spans[c].end
551 for c in self.shrink_target.span_comments
552 if c is not None
553 ):
554 continue
556 # Try a few targeted candidates before falling back to random sampling,
557 # so that simple cases like ``assert n1 == n2`` -- where the only
558 # passing value of ``n1`` is exactly ``n2``'s value -- aren't reported
559 # as freely-variable just because random sampling missed it.
560 candidates = list(self._explain_candidates(start, end))
562 # Run our experiments
563 n_same_failures = 0
564 note = "or any other generated value"
565 # TODO: is 100 same-failures out of 500 attempts a good heuristic?
566 for n_attempt in range(500 + len(candidates)): # pragma: no branch
567 # no-branch here because we don't coverage-test the abort-at-500 logic.
569 if n_attempt - 10 - len(candidates) > n_same_failures * 5:
570 # stop early if we're seeing mostly invalid test cases
571 break # pragma: no cover
573 if n_attempt < len(candidates):
574 replacement = list(candidates[n_attempt])
575 else:
576 # replace start:end with random values
577 replacement = []
578 for i in range(start, end):
579 node = nodes[i]
580 if not node.was_forced:
581 value = draw_choice(
582 node.type, node.constraints, random=self.random
583 )
584 node = node.copy(with_value=value)
585 replacement.append(node.value)
587 attempt = choices[:start] + tuple(replacement) + choices[end:]
588 result = self.engine.cached_test_function(attempt, extend="full")
590 if result.status is Status.OVERRUN:
591 continue # pragma: no cover # flakily covered
592 result = cast(ConjectureResult, result)
593 if not (
594 len(attempt) == len(result.choices)
595 and endswith(result.nodes, nodes[end:])
596 ):
597 # Turns out this was a variable-length part, so grab the infix...
598 for span1, span2 in zip(
599 shrink_target.spans, result.spans, strict=False
600 ):
601 assert span1.start == span2.start
602 assert span1.start <= start
603 if span1.start == start and span1.end == end:
604 result_end = span2.end
605 break
606 else:
607 raise NotImplementedError("Expected matching prefixes")
609 attempt = (
610 choices[:start]
611 + result.choices[start:result_end]
612 + choices[end:]
613 )
614 chunks[span_index].append(result.choices[start:result_end])
615 result = self.engine.cached_test_function(attempt)
617 if result.status is Status.OVERRUN:
618 continue # pragma: no cover # flakily covered
619 result = cast(ConjectureResult, result)
620 else:
621 chunks[span_index].append(result.choices[start:end])
623 if shrink_target is not self.shrink_target: # pragma: no cover
624 # If we've shrunk further without meaning to, bail out.
625 self.shrink_target.span_comments.clear()
626 return
627 if result.status is Status.VALID:
628 # The test passed, indicating that this param can't vary freely.
629 # However, it's really hard to write a simple and reliable covering
630 # test, because of our `seen_passing_buffers` check above.
631 break # pragma: no cover
632 if self.__predicate(result): # pragma: no branch
633 n_same_failures += 1
634 if n_same_failures >= 100:
635 self.shrink_target.span_comments[span_index] = note
636 break
638 # Finally, if we've found multiple independently-variable parts, check whether
639 # they can all be varied together.
640 if len(self.shrink_target.span_comments) <= 1:
641 return
642 n_same_failures_together = 0
643 # Only include spans that actually got a comment
644 chunks_by_start_index = sorted(
645 (arg_ranges[k], v)
646 for k, v in chunks.items()
647 if k in self.shrink_target.span_comments
648 )
649 for _ in range(500): # pragma: no branch
650 # no-branch here because we don't coverage-test the abort-at-500 logic.
651 new_choices: list[ChoiceT] = []
652 prev_end = 0
653 for (start, end), ls in chunks_by_start_index:
654 assert prev_end <= start < end, "these chunks must be nonoverlapping"
655 new_choices.extend(choices[prev_end:start])
656 new_choices.extend(self.random.choice(ls))
657 prev_end = end
658 new_choices.extend(choices[prev_end:])
660 result = self.engine.cached_test_function(new_choices)
662 # This *can't* be a shrink because none of the components were.
663 assert shrink_target is self.shrink_target
664 if result.status == Status.VALID:
665 self.shrink_target.span_comments[None] = (
666 "The test sometimes passed when commented parts were varied together."
667 )
668 break # Test passed, this param can't vary freely.
669 if self.__predicate(result): # pragma: no branch
670 n_same_failures_together += 1
671 if n_same_failures_together >= 100:
672 self.shrink_target.span_comments[None] = (
673 "The test always failed when commented parts were varied together."
674 )
675 break
677 def _explain_candidates(
678 self, start: int, end: int
679 ) -> "Iterator[tuple[ChoiceT, ...]]":
680 """Yield deterministic candidate replacements for ``nodes[start:end]``.
682 Random sampling alone misses cases like ``assert n1 == n2``, where the
683 only passing value of ``n1`` is exactly ``n2``'s value. We try
684 substituting values from each other arg slice with matching length and
685 types, which catches such comparisons. Invalid borrowed values just
686 produce an irrelevant test result the outer loop discards.
687 """
688 nodes = self.nodes
689 spans = self.shrink_target.spans
690 target_types = tuple(nodes[i].type for i in range(start, end))
691 current_key = choices_key(tuple(nodes[i].value for i in range(start, end)))
692 seen: set[tuple[Any, ...]] = {current_key}
693 arg_ranges = sorted(
694 {(spans[i].start, spans[i].end) for i in self.shrink_target.arg_spans}
695 )
696 for start2, end2 in arg_ranges:
697 if (start2, end2) == (start, end) or (end2 - start2) != (end - start):
698 continue
699 if (
700 tuple(nodes[start2 + j].type for j in range(end - start))
701 != target_types
702 ):
703 continue
704 borrowed = tuple(nodes[start2 + j].value for j in range(end - start))
705 key = choices_key(borrowed)
706 if key in seen:
707 continue
708 seen.add(key)
709 yield borrowed
711 def greedy_shrink(self) -> None:
712 """Run a full set of greedy shrinks (that is, ones that will only ever
713 move to a better target) and update shrink_target appropriately.
715 This method iterates to a fixed point and so is idempontent - calling
716 it twice will have exactly the same effect as calling it once.
717 """
718 self.fixate_shrink_passes(self.shrink_passes)
720 def initial_coarse_reduction(self):
721 """Performs some preliminary reductions that should not be
722 repeated as part of the main shrink passes.
724 The main reason why these can't be included as part of shrink
725 passes is that they have much more ability to make the test
726 case "worse". e.g. they might rerandomise part of it, significantly
727 increasing the value of individual nodes, which works in direct
728 opposition to the lexical shrinking and will frequently undo
729 its work.
730 """
731 self.reduce_each_alternative()
733 @derived_value # type: ignore
734 def spans_starting_at(self):
735 result = [[] for _ in self.shrink_target.nodes]
736 for i, ex in enumerate(self.spans):
737 # We can have zero-length spans that start at the end
738 if ex.start < len(result):
739 result[ex.start].append(i)
740 return tuple(map(tuple, result))
742 def reduce_each_alternative(self):
743 """This is a pass that is designed to rerandomise use of the
744 one_of strategy or things that look like it, in order to try
745 to move from later strategies to earlier ones in the branch
746 order.
748 It does this by trying to systematically lower each value it
749 finds that looks like it might be the branch decision for
750 one_of, and then attempts to repair any changes in shape that
751 this causes.
752 """
753 i = 0
754 while i < len(self.shrink_target.nodes):
755 nodes = self.shrink_target.nodes
756 node = nodes[i]
757 if (
758 node.type == "integer"
759 and not node.was_forced
760 and node.value <= 10
761 and node.constraints["min_value"] == 0
762 ):
763 assert isinstance(node.value, int)
765 # We've found a plausible candidate for a ``one_of`` choice.
766 # We now want to see if the shape of the test case actually depends
767 # on it. If it doesn't, then we don't need to do this (comparatively
768 # costly) pass, and can let much simpler lexicographic reduction
769 # handle it later.
770 #
771 # We test this by trying to set the value to zero and seeing if the
772 # shape changes, as measured by either changing the number of subsequent
773 # nodes, or changing the nodes in such a way as to cause one of the
774 # previous values to no longer be valid in its position.
775 zero_attempt = self.cached_test_function(
776 nodes[:i] + (nodes[i].copy(with_value=0),) + nodes[i + 1 :]
777 )[1]
778 if (
779 zero_attempt is not self.shrink_target
780 and zero_attempt is not None
781 and zero_attempt.status >= Status.VALID
782 ):
783 changed_shape = len(zero_attempt.nodes) != len(nodes)
785 if not changed_shape:
786 for j in range(i + 1, len(nodes)):
787 zero_node = zero_attempt.nodes[j]
788 orig_node = nodes[j]
789 if (
790 zero_node.type != orig_node.type
791 or not choice_permitted(
792 orig_node.value, zero_node.constraints
793 )
794 ):
795 changed_shape = True
796 break
797 if changed_shape:
798 for v in range(node.value):
799 if self.try_lower_node_as_alternative(i, v):
800 break
801 i += 1
803 def try_lower_node_as_alternative(self, i, v):
804 """Attempt to lower `self.shrink_target.nodes[i]` to `v`,
805 while rerandomising and attempting to repair any subsequent
806 changes to the shape of the test case that this causes."""
807 nodes = self.shrink_target.nodes
808 if self.consider_new_nodes(
809 nodes[:i] + (nodes[i].copy(with_value=v),) + nodes[i + 1 :]
810 ):
811 return True
813 prefix = nodes[:i] + (nodes[i].copy(with_value=v),)
814 initial = self.shrink_target
815 spans = self.spans_starting_at[i]
816 for _ in range(3):
817 random_attempt = self.engine.cached_test_function(
818 [n.value for n in prefix], extend=len(nodes)
819 )
820 if random_attempt.status < Status.VALID:
821 continue
822 self.incorporate_test_data(random_attempt)
823 for j in spans:
824 initial_span = initial.spans[j]
825 attempt_span = random_attempt.spans[j]
826 contents = random_attempt.nodes[attempt_span.start : attempt_span.end]
827 self.consider_new_nodes(
828 nodes[:i] + contents + nodes[initial_span.end :]
829 )
830 if initial is not self.shrink_target:
831 return True
832 return False
834 @derived_value # type: ignore
835 def shrink_pass_choice_trees(self) -> dict[Any, ChoiceTree]:
836 return defaultdict(ChoiceTree)
838 def step(self, shrink_pass: ShrinkPass, *, random_order: bool = False) -> bool:
839 tree = self.shrink_pass_choice_trees[shrink_pass]
840 if tree.exhausted:
841 return False
843 initial_shrinks = self.shrinks
844 initial_calls = self.calls
845 initial_misaligned = self.misaligned
846 size = len(self.shrink_target.choices)
847 assert shrink_pass.name is not None
848 self.engine.explain_next_call_as(shrink_pass.name)
850 if random_order:
851 selection_order = random_selection_order(self.random)
852 else:
853 selection_order = prefix_selection_order(shrink_pass.last_prefix)
855 try:
856 shrink_pass.last_prefix = tree.step(
857 selection_order,
858 lambda chooser: shrink_pass.function(chooser),
859 )
860 finally:
861 shrink_pass.calls += self.calls - initial_calls
862 shrink_pass.misaligned += self.misaligned - initial_misaligned
863 shrink_pass.shrinks += self.shrinks - initial_shrinks
864 shrink_pass.deletions += size - len(self.shrink_target.choices)
865 self.engine.clear_call_explanation()
866 return True
868 def fixate_shrink_passes(self, passes: list[ShrinkPass]) -> None:
869 """Run steps from each pass in ``passes`` until the current shrink target
870 is a fixed point of all of them."""
871 any_ran = True
872 while any_ran:
873 any_ran = False
875 reordering = {}
877 # We run remove_discarded after every pass to do cleanup
878 # keeping track of whether that actually works. Either there is
879 # no discarded data and it is basically free, or it reliably works
880 # and deletes data, or it doesn't work. In that latter case we turn
881 # it off for the rest of this loop through the passes, but will
882 # try again once all of the passes have been run.
883 can_discard = self.remove_discarded()
885 calls_at_loop_start = self.calls
887 # We keep track of how many calls can be made by a single step
888 # without making progress and use this to test how much to pad
889 # out self.max_stall by as we go along.
890 max_calls_per_failing_step = 1
892 for sp in passes:
893 if can_discard:
894 can_discard = self.remove_discarded()
896 before_sp = self.shrink_target
898 # Run the shrink pass until it fails to make any progress
899 # max_failures times in a row. This implicitly boosts shrink
900 # passes that are more likely to work.
901 failures = 0
902 max_failures = 20
903 while failures < max_failures:
904 # We don't allow more than max_stall consecutive failures
905 # to shrink, but this means that if we're unlucky and the
906 # shrink passes are in a bad order where only the ones at
907 # the end are useful, if we're not careful this heuristic
908 # might stop us before we've tried everything. In order to
909 # avoid that happening, we make sure that there's always
910 # plenty of breathing room to make it through a single
911 # iteration of the fixate_shrink_passes loop.
912 self.max_stall = max(
913 self.max_stall,
914 2 * max_calls_per_failing_step
915 + (self.calls - calls_at_loop_start),
916 )
918 prev = self.shrink_target
919 initial_calls = self.calls
920 # It's better for us to run shrink passes in a deterministic
921 # order, to avoid repeat work, but this can cause us to create
922 # long stalls when there are a lot of steps which fail to do
923 # anything useful. In order to avoid this, once we've noticed
924 # we're in a stall (i.e. half of max_failures calls have failed
925 # to do anything) we switch to randomly jumping around. If we
926 # find a success then we'll resume deterministic order from
927 # there which, with any luck, is in a new good region.
928 if not self.step(sp, random_order=failures >= max_failures // 2):
929 # step returns False when there is nothing to do because
930 # the entire choice tree is exhausted. If this happens
931 # we break because we literally can't run this pass any
932 # more than we already have until something else makes
933 # progress.
934 break
935 any_ran = True
937 # Don't count steps that didn't actually try to do
938 # anything as failures. Otherwise, this call is a failure
939 # if it failed to make any changes to the shrink target.
940 if initial_calls != self.calls:
941 if prev is not self.shrink_target:
942 failures = 0
943 else:
944 max_calls_per_failing_step = max(
945 max_calls_per_failing_step, self.calls - initial_calls
946 )
947 failures += 1
949 # We reorder the shrink passes so that on our next run through
950 # we try good ones first. The rule is that shrink passes that
951 # did nothing useful are the worst, shrink passes that reduced
952 # the length are the best.
953 if self.shrink_target is before_sp:
954 reordering[sp] = 1
955 elif len(self.choices) < len(before_sp.choices):
956 reordering[sp] = -1
957 else:
958 reordering[sp] = 0
960 passes.sort(key=reordering.__getitem__)
962 @property
963 def nodes(self) -> tuple[ChoiceNode, ...]:
964 return self.shrink_target.nodes
966 @property
967 def choices(self) -> tuple[ChoiceT, ...]:
968 return self.shrink_target.choices
970 @property
971 def spans(self) -> Spans:
972 return self.shrink_target.spans
974 @derived_value # type: ignore
975 def spans_by_label(self):
976 """
977 A mapping of labels to a list of spans with that label. Spans in the list
978 are ordered by their normal index order.
979 """
981 spans_by_label = defaultdict(list)
982 for ex in self.spans:
983 spans_by_label[ex.label].append(ex)
984 return dict(spans_by_label)
986 @derived_value # type: ignore
987 def distinct_labels(self):
988 return sorted(self.spans_by_label, key=str)
990 def pass_to_descendant(self, chooser):
991 """Attempt to replace each span with a descendant span.
993 This is designed to deal with strategies that call themselves
994 recursively. For example, suppose we had:
996 binary_tree = st.deferred(
997 lambda: st.one_of(
998 st.integers(), st.tuples(binary_tree, binary_tree)))
1000 This pass guarantees that we can replace any binary tree with one of
1001 its subtrees - each of those will create an interval that the parent
1002 could validly be replaced with, and this pass will try doing that.
1004 This is pretty expensive - it takes O(len(intervals)^2) - so we run it
1005 late in the process when we've got the number of intervals as far down
1006 as possible.
1007 """
1009 label = chooser.choose(
1010 self.distinct_labels, lambda l: len(self.spans_by_label[l]) >= 2
1011 )
1013 spans = self.spans_by_label[label]
1014 i = chooser.choose(range(len(spans) - 1))
1015 ancestor = spans[i]
1017 if i + 1 == len(spans) or spans[i + 1].start >= ancestor.end:
1018 return
1020 @self.cached(label, i)
1021 def descendants():
1022 lo = i + 1
1023 hi = len(spans)
1024 while lo + 1 < hi:
1025 mid = (lo + hi) // 2
1026 if spans[mid].start >= ancestor.end:
1027 hi = mid
1028 else:
1029 lo = mid
1030 return [
1031 span
1032 for span in spans[i + 1 : hi]
1033 if span.choice_count < ancestor.choice_count
1034 ]
1036 descendant = chooser.choose(descendants, lambda ex: ex.choice_count > 0)
1038 assert ancestor.start <= descendant.start
1039 assert ancestor.end >= descendant.end
1040 assert descendant.choice_count < ancestor.choice_count
1042 self.consider_new_nodes(
1043 self.nodes[: ancestor.start]
1044 + self.nodes[descendant.start : descendant.end]
1045 + self.nodes[ancestor.end :]
1046 )
1048 def lower_common_node_offset(self):
1049 """Sometimes we find ourselves in a situation where changes to one part
1050 of the choice sequence unlock changes to other parts. Sometimes this is
1051 good, but sometimes this can cause us to exhibit exponential slow
1052 downs!
1054 e.g. suppose we had the following:
1056 m = draw(integers(min_value=0))
1057 n = draw(integers(min_value=0))
1058 assert abs(m - n) > 1
1060 If this fails then we'll end up with a loop where on each iteration we
1061 reduce each of m and n by 2 - m can't go lower because of n, then n
1062 can't go lower because of m.
1064 This will take us O(m) iterations to complete, which is exponential in
1065 the data size, as we gradually zig zag our way towards zero.
1067 This can only happen if we're failing to reduce the size of the choice
1068 sequence: The number of iterations that reduce the length of the choice
1069 sequence is bounded by that length.
1071 So what we do is this: We keep track of which nodes are changing, and
1072 then if there's some non-zero common offset to them we try and minimize
1073 them all at once by lowering that offset.
1075 This may not work, and it definitely won't get us out of all possible
1076 exponential slow downs (an example of where it doesn't is where the
1077 shape of the nodes changes as a result of this bouncing behaviour),
1078 but it fails fast when it doesn't work and gets us out of a really
1079 nastily slow case when it does.
1080 """
1081 if len(self.__changed_nodes) <= 1:
1082 return
1084 changed = []
1085 for i in sorted(self.__changed_nodes):
1086 node = self.nodes[i]
1087 if node.trivial or node.type != "integer":
1088 continue
1089 changed.append(node)
1091 if not changed:
1092 return
1094 ints = [
1095 abs(node.value - node.constraints["shrink_towards"]) for node in changed
1096 ]
1097 offset = min(ints)
1098 assert offset > 0
1100 for i in range(len(ints)):
1101 ints[i] -= offset
1103 st = self.shrink_target
1105 def offset_node(node, n):
1106 return (
1107 node.index,
1108 node.index + 1,
1109 [node.copy(with_value=node.constraints["shrink_towards"] + n)],
1110 )
1112 def consider(n, sign):
1113 return self.consider_new_nodes(
1114 replace_all(
1115 st.nodes,
1116 [
1117 offset_node(node, sign * (n + v))
1118 for node, v in zip(changed, ints, strict=False)
1119 ],
1120 )
1121 )
1123 # shrink from both sides
1124 Integer.shrink(offset, lambda n: consider(n, 1))
1125 Integer.shrink(offset, lambda n: consider(n, -1))
1126 self.clear_change_tracking()
1128 def clear_change_tracking(self):
1129 self.__last_checked_changed_at = self.shrink_target
1130 self.__all_changed_nodes = set()
1132 def mark_changed(self, i):
1133 self.__changed_nodes.add(i)
1135 @property
1136 def __changed_nodes(self) -> set[int]:
1137 if self.__last_checked_changed_at is self.shrink_target:
1138 return self.__all_changed_nodes
1140 prev_target = self.__last_checked_changed_at
1141 new_target = self.shrink_target
1142 assert prev_target is not new_target
1143 prev_nodes = prev_target.nodes
1144 new_nodes = new_target.nodes
1145 assert sort_key(new_target.nodes) < sort_key(prev_target.nodes)
1147 if len(prev_nodes) != len(new_nodes) or any(
1148 n1.type != n2.type for n1, n2 in zip(prev_nodes, new_nodes, strict=True)
1149 ):
1150 # should we check constraints are equal as well?
1151 self.__all_changed_nodes = set()
1152 else:
1153 assert len(prev_nodes) == len(new_nodes)
1154 for i, (n1, n2) in enumerate(zip(prev_nodes, new_nodes, strict=True)):
1155 assert n1.type == n2.type
1156 if not choice_equal(n1.value, n2.value):
1157 self.__all_changed_nodes.add(i)
1159 return self.__all_changed_nodes
1161 def update_shrink_target(self, new_target):
1162 assert isinstance(new_target, ConjectureResult)
1163 self.shrinks += 1
1164 # If we are just taking a long time to shrink we don't want to
1165 # trigger this heuristic, so whenever we shrink successfully
1166 # we give ourselves a bit of breathing room to make sure we
1167 # would find a shrink that took that long to find the next time.
1168 # The case where we're taking a long time but making steady
1169 # progress is handled by `finish_shrinking_deadline` in engine.py
1170 self.max_stall = max(
1171 self.max_stall, (self.calls - self.calls_at_last_shrink) * 2
1172 )
1173 self.calls_at_last_shrink = self.calls
1174 self.shrink_target = new_target
1175 self.__derived_values = {}
1177 def try_shrinking_nodes(self, nodes, n):
1178 """Attempts to replace each node in the nodes list with n. Returns
1179 True if it succeeded (which may include some additional modifications
1180 to shrink_target).
1182 In current usage it is expected that each of the nodes currently have
1183 the same value and choice_type, although this is not essential. Note that
1184 n must be < the node at min(nodes) or this is not a valid shrink.
1186 This method will attempt to do some small amount of work to delete data
1187 that occurs after the end of the nodes. This is useful for cases where
1188 there is some size dependency on the value of a node.
1189 """
1190 # If the length of the shrink target has changed from under us such that
1191 # the indices are out of bounds, give up on the replacement.
1192 # TODO_BETTER_SHRINK: we probably want to narrow down the root cause here at some point.
1193 if any(node.index >= len(self.nodes) for node in nodes):
1194 return # pragma: no cover
1196 initial_attempt = replace_all(
1197 self.nodes,
1198 [(node.index, node.index + 1, [node.copy(with_value=n)]) for node in nodes],
1199 )
1201 attempt = self.cached_test_function(initial_attempt)[1]
1203 if attempt is None:
1204 return False
1206 if attempt is self.shrink_target:
1207 # if the initial shrink was a success, try lowering offsets.
1208 self.lower_common_node_offset()
1209 return True
1211 # If this produced something completely invalid we ditch it
1212 # here rather than trying to persevere.
1213 if attempt.status is Status.OVERRUN:
1214 # Lowering a size-controlling choice can make the realigned (and
1215 # now boring) collection stop triggering the failure, so the test
1216 # draws further and overruns before we see the realignment -- this
1217 # is common in stateful tests, where a non-failing step is followed
1218 # by more steps. Re-run without the length limit to recover the
1219 # realigned tree, which the repair logic below can then act on.
1220 attempt = self.engine.cached_test_function(
1221 [n.value for n in initial_attempt], extend="full"
1222 )
1223 if attempt.status is Status.OVERRUN:
1224 return False
1226 if attempt.status is Status.INVALID:
1227 return False
1229 # When we lower a choice that controls the size of a later collection,
1230 # eg
1231 #
1232 # n = data.draw_integer()
1233 # s = data.draw_string(min_size=n, max_size=n)
1234 #
1235 # the recorded value for that collection no longer fits the constraints
1236 # the test function actually used, so the engine realigns the tree by
1237 # substituting a freshly-generated (simplest) value -- discarding
1238 # whatever made the collection interesting. (We can't rely on
1239 # ``attempt.misaligned_at`` to detect this, because the realigned choice
1240 # sequence is often independently cached as an ordinary, non-misaligned
1241 # result.) We detect a string/bytes node whose recorded value is now too
1242 # long, and retry with it truncated to fit. We try preserving content
1243 # from either end, since the interesting part may be at the start or the
1244 # end (see test_can_shrink_variable_string_draws).
1245 for i in range(min(len(initial_attempt), len(attempt.nodes))):
1246 node = initial_attempt[i]
1247 attempt_node = attempt.nodes[i]
1248 if (
1249 node.type == attempt_node.type
1250 and node.type in {"string", "bytes"}
1251 and not node.was_forced
1252 and len(node.value) > attempt_node.constraints["max_size"]
1253 ):
1254 max_size = attempt_node.constraints["max_size"]
1255 for truncated in (node.value[:max_size], node.value[-max_size:]):
1256 if self.consider_new_nodes(
1257 initial_attempt[:i]
1258 + [
1259 node.copy(
1260 with_constraints=attempt_node.constraints,
1261 with_value=truncated,
1262 )
1263 ]
1264 + initial_attempt[i + 1 :]
1265 ):
1266 return True
1268 lost_nodes = len(self.nodes) - len(attempt.nodes)
1269 if lost_nodes <= 0:
1270 return False
1272 start = nodes[0].index
1273 end = nodes[-1].index + 1
1274 # We now look for contiguous regions to delete that might help fix up
1275 # this failed shrink. We only look for contiguous regions of the right
1276 # lengths because doing anything more than that starts to get very
1277 # expensive. See minimize_individual_choices for where we
1278 # try to be more aggressive.
1279 regions_to_delete = {(end, end + lost_nodes)}
1281 for ex in self.spans:
1282 if ex.start > start:
1283 continue
1284 if ex.end <= end:
1285 continue
1287 if ex.index >= len(attempt.spans):
1288 continue # pragma: no cover
1290 replacement = attempt.spans[ex.index]
1291 in_original = [c for c in ex.children if c.start >= end]
1292 in_replaced = [c for c in replacement.children if c.start >= end]
1294 if len(in_replaced) >= len(in_original) or not in_replaced:
1295 continue
1297 # We've found a span where some of the children went missing
1298 # as a result of this change, and just replacing it with the data
1299 # it would have had and removing the spillover didn't work. This
1300 # means that some of its children towards the right must be
1301 # important, so we try to arrange it so that it retains its
1302 # rightmost children instead of its leftmost.
1303 regions_to_delete.add(
1304 (in_original[0].start, in_original[-len(in_replaced)].start)
1305 )
1307 for u, v in sorted(regions_to_delete, key=lambda x: x[1] - x[0], reverse=True):
1308 try_with_deleted = initial_attempt[:u] + initial_attempt[v:]
1309 if self.consider_new_nodes(try_with_deleted):
1310 return True
1312 return False
1314 def remove_discarded(self):
1315 """Try removing all nodes marked as discarded.
1317 This is primarily to deal with data that has been ignored while
1318 doing rejection sampling - e.g. as a result of an integer range, or a
1319 filtered strategy.
1321 Such data will also be handled by the ``node_program("X")`` deletion
1322 passes, but those are necessarily more conservative and will try
1323 deleting each contiguous run of nodes individually. The common case is
1324 that all data drawn and rejected can just be thrown away immediately in
1325 one block, so this pass will be much faster than trying each one
1326 individually when it works.
1328 returns False if there is discarded data and removing it does not work,
1329 otherwise returns True.
1330 """
1331 while self.shrink_target.has_discards:
1332 discarded = []
1334 for ex in self.shrink_target.spans:
1335 if (
1336 ex.choice_count > 0
1337 and ex.discarded
1338 and (not discarded or ex.start >= discarded[-1][-1])
1339 ):
1340 discarded.append((ex.start, ex.end))
1342 # This can happen if we have discards but they are all of
1343 # zero length. This shouldn't happen very often so it's
1344 # faster to check for it here than at the point of span
1345 # generation.
1346 if not discarded:
1347 break
1349 attempt = list(self.nodes)
1350 for u, v in reversed(discarded):
1351 del attempt[u:v]
1353 if not self.consider_new_nodes(tuple(attempt)):
1354 return False
1355 return True
1357 @derived_value # type: ignore
1358 def duplicated_nodes(self):
1359 """Returns a list of nodes grouped (choice_type, value)."""
1360 duplicates = defaultdict(list)
1361 for node in self.nodes:
1362 duplicates[(node.type, choice_key(node.value))].append(node)
1363 return list(duplicates.values())
1365 def node_program(self, program: str) -> ShrinkPass:
1366 return ShrinkPass(
1367 lambda chooser: self._node_program(chooser, program),
1368 name=f"node_program_{program}",
1369 )
1371 def _node_program(self, chooser, program):
1372 n = len(program)
1373 # Adaptively attempt to run the node program at the current
1374 # index. If this successfully applies the node program ``k`` times
1375 # then this runs in ``O(log(k))`` test function calls.
1376 i = chooser.choose(range(len(self.nodes) - n + 1))
1378 # First, run the node program at the chosen index. If this fails,
1379 # don't do any extra work, so that failure is as cheap as possible.
1380 if not self.run_node_program(i, program, original=self.shrink_target):
1381 return
1383 # Because we run in a random order we will often find ourselves in the middle
1384 # of a region where we could run the node program. We thus start by moving
1385 # left to the beginning of that region if possible in order to start from
1386 # the beginning of that region.
1387 def offset_left(k):
1388 return i - k * n
1390 i = offset_left(
1391 find_integer(
1392 lambda k: self.run_node_program(
1393 offset_left(k), program, original=self.shrink_target
1394 )
1395 )
1396 )
1398 original = self.shrink_target
1399 # Now try to run the node program multiple times here.
1400 find_integer(
1401 lambda k: self.run_node_program(i, program, original=original, repeats=k)
1402 )
1404 def minimize_duplicated_choices(self, chooser):
1405 """Find choices that have been duplicated in multiple places and attempt
1406 to minimize all of the duplicates simultaneously.
1408 This lets us handle cases where two values can't be shrunk
1409 independently of each other but can easily be shrunk together.
1410 For example if we had something like:
1412 ls = data.draw(lists(integers()))
1413 y = data.draw(integers())
1414 assert y not in ls
1416 Suppose we drew y = 3 and after shrinking we have ls = [3]. If we were
1417 to replace both 3s with 0, this would be a valid shrink, but if we were
1418 to replace either 3 with 0 on its own the test would start passing.
1420 It is also useful for when that duplication is accidental and the value
1421 of the choices don't matter very much because it allows us to replace
1422 more values at once.
1423 """
1424 nodes = chooser.choose(self.duplicated_nodes)
1425 # we can't lower any nodes which are trivial. try proceeding with the
1426 # remaining nodes.
1427 nodes = [node for node in nodes if not node.trivial]
1428 if len(nodes) <= 1:
1429 return
1431 self.minimize_nodes(nodes)
1433 def redistribute_numeric_pairs(self, chooser):
1434 """If there is a sum of generated numbers that we need their sum
1435 to exceed some bound, lowering one of them requires raising the
1436 other. This pass enables that."""
1438 # look for a pair of nodes (node1, node2) which are both numeric
1439 # and aren't separated by too many other nodes. We'll decrease node1 and
1440 # increase node2 (note that the other way around doesn't make sense as
1441 # it's strictly worse in the ordering).
1442 def can_choose_node(node):
1443 # don't choose nan, inf, or floats above the threshold where f + 1 > f
1444 # (which is not necessarily true for floats above MAX_PRECISE_INTEGER).
1445 # The motivation for the last condition is to avoid trying weird
1446 # non-shrinks where we raise one node and think we lowered another
1447 # (but didn't).
1448 return node.type in {"integer", "float"} and not (
1449 node.type == "float"
1450 and (math.isnan(node.value) or abs(node.value) >= MAX_PRECISE_INTEGER)
1451 )
1453 node1 = chooser.choose(
1454 self.nodes,
1455 lambda node: can_choose_node(node) and not node.trivial,
1456 )
1457 node2 = chooser.choose(
1458 self.nodes,
1459 lambda node: (
1460 can_choose_node(node)
1461 # Note that it's fine for node2 to be trivial, because we're going to
1462 # explicitly make it *not* trivial by adding to its value.
1463 and not node.was_forced
1464 # to avoid quadratic behavior, scan ahead only a small amount for
1465 # the related node.
1466 and node1.index < node.index <= node1.index + 4
1467 ),
1468 )
1470 m: int | float = node1.value
1471 n: int | float = node2.value
1473 def boost(k: int) -> bool:
1474 # floats always shrink towards 0
1475 shrink_towards = (
1476 node1.constraints["shrink_towards"] if node1.type == "integer" else 0
1477 )
1478 if k > abs(m - shrink_towards):
1479 return False
1481 # We are trying to move node1 (m) closer to shrink_towards, and node2
1482 # (n) farther away from shrink_towards. If m is below shrink_towards,
1483 # we want to add to m and subtract from n, and vice versa if above
1484 # shrink_towards.
1485 if m < shrink_towards:
1486 k = -k
1488 try:
1489 v1 = m - k
1490 v2 = n + k
1491 except OverflowError: # pragma: no cover
1492 # if n or m is a float and k is over sys.float_info.max, coercing
1493 # k to a float will overflow.
1494 return False
1496 # if we've increased node2 to the point that we're past max precision,
1497 # give up - things have become too unstable.
1498 if node2.type == "float" and abs(v2) >= MAX_PRECISE_INTEGER:
1499 return False
1501 return self.consider_new_nodes(
1502 self.nodes[: node1.index]
1503 + (node1.copy(with_value=v1),)
1504 + self.nodes[node1.index + 1 : node2.index]
1505 + (node2.copy(with_value=v2),)
1506 + self.nodes[node2.index + 1 :]
1507 )
1509 find_integer(boost)
1511 def lower_integers_together(self, chooser):
1512 node1 = chooser.choose(
1513 self.nodes, lambda n: n.type == "integer" and not n.trivial
1514 )
1515 # Search up to 3 nodes ahead, to avoid quadratic time.
1516 node2 = self.nodes[
1517 chooser.choose(
1518 range(node1.index + 1, min(len(self.nodes), node1.index + 3 + 1)),
1519 lambda i: (
1520 self.nodes[i].type == "integer" and not self.nodes[i].was_forced
1521 ),
1522 )
1523 ]
1525 # one might expect us to require node2 to be nontrivial, and to minimize
1526 # the node which is closer to its shrink_towards, rather than node1
1527 # unconditionally. In reality, it's acceptable for us to transition node2
1528 # from trivial to nontrivial, because the shrink ordering is dominated by
1529 # the complexity of the earlier node1. What matters is minimizing node1.
1530 shrink_towards = node1.constraints["shrink_towards"]
1532 def consider(n):
1533 return self.consider_new_nodes(
1534 self.nodes[: node1.index]
1535 + (node1.copy(with_value=node1.value - n),)
1536 + self.nodes[node1.index + 1 : node2.index]
1537 + (node2.copy(with_value=node2.value - n),)
1538 + self.nodes[node2.index + 1 :]
1539 )
1541 find_integer(lambda n: consider(shrink_towards - n))
1542 find_integer(lambda n: consider(n - shrink_towards))
1544 def lower_duplicated_characters(self, chooser):
1545 """
1546 Select two string choices no more than 4 choices apart and simultaneously
1547 lower characters which appear in both strings. This helps cases where the
1548 same character must appear in two strings, but the actual value of the
1549 character is not relevant.
1551 This shrinking pass currently only tries lowering *all* instances of the
1552 duplicated character in both strings. So for instance, given two choices:
1554 "bbac"
1555 "abbb"
1557 we would try lowering all five of the b characters simultaneously. This
1558 may fail to shrink some cases where only certain character indices are
1559 correlated, for instance if only the b at index 1 could be lowered
1560 simultaneously and the rest did in fact actually have to be a `b`.
1562 It would be nice to try shrinking that case as well, but we would need good
1563 safeguards because it could get very expensive to try all combinations.
1564 I expect lowering all duplicates to handle most cases in the meantime.
1565 """
1566 node1 = chooser.choose(
1567 self.nodes, lambda n: n.type == "string" and not n.trivial
1568 )
1570 # limit search to up to 4 choices ahead, to avoid quadratic behavior
1571 node2 = self.nodes[
1572 chooser.choose(
1573 range(node1.index + 1, min(len(self.nodes), node1.index + 1 + 4)),
1574 lambda i: (
1575 self.nodes[i].type == "string"
1576 and not self.nodes[i].trivial
1577 # select nodes which have at least one of the same character present
1578 and set(node1.value) & set(self.nodes[i].value)
1579 ),
1580 )
1581 ]
1583 duplicated_characters = set(node1.value) & set(node2.value)
1584 # deterministic ordering
1585 char = chooser.choose(sorted(duplicated_characters))
1586 intervals = node1.constraints["intervals"]
1588 def copy_node(node, n):
1589 # replace all duplicate characters in each string. This might miss
1590 # some shrinks compared to only replacing some, but trying all possible
1591 # combinations of indices could get expensive if done without some
1592 # thought.
1593 return node.copy(
1594 with_value=node.value.replace(char, intervals.char_in_shrink_order(n))
1595 )
1597 Integer.shrink(
1598 intervals.index_from_char_in_shrink_order(char),
1599 lambda n: self.consider_new_nodes(
1600 self.nodes[: node1.index]
1601 + (copy_node(node1, n),)
1602 + self.nodes[node1.index + 1 : node2.index]
1603 + (copy_node(node2, n),)
1604 + self.nodes[node2.index + 1 :]
1605 ),
1606 )
1608 def normalize_unicode_chars(self, chooser):
1609 """For string nodes, try replacing characters with simpler equivalents
1610 from natural text transformations: unicode decomposition (NFD, NFKD)
1611 and case mapping. For example, an accented latin letter is reduced
1612 to its base form, a ligature is reduced to its first base character,
1613 a mathematical alphanumeric symbol is reduced to its plain ascii
1614 counterpart, and a lowercase letter is replaced with its uppercase
1615 form (which has a smaller shrink-order index in the default
1616 alphabet).
1618 The codepoint shrinker is binary-search based, so it can get stuck on
1619 a high codepoint whose simpler equivalents aren't reached by halving
1620 / shifting / masking. This pass directly tries the natural simpler
1621 forms one character at a time.
1622 """
1623 node = chooser.choose(
1624 self.nodes,
1625 lambda n: n.type == "string"
1626 and any(
1627 _natural_simpler_chars(c, n.constraints["intervals"]) for c in n.value
1628 ),
1629 )
1630 intervals = node.constraints["intervals"]
1631 i = chooser.choose(
1632 range(len(node.value)),
1633 lambda j: bool(_natural_simpler_chars(node.value[j], intervals)),
1634 )
1635 for replacement in _natural_simpler_chars(node.value[i], intervals):
1636 new_value = node.value[:i] + replacement + node.value[i + 1 :]
1637 if self.consider_new_nodes(
1638 self.nodes[: node.index]
1639 + (node.copy(with_value=new_value),)
1640 + self.nodes[node.index + 1 :]
1641 ):
1642 return
1644 def widen_to_span_with_recorded_value(self, chooser):
1645 """Replace a strategy's choices with a ValueHole carrying the value it
1646 produced, so that the replay re-encodes that value more simply.
1648 We look for a span which starts with a non-forced zero-based non-zero
1649 integer choice (heuristically: a one_of branch selector), makes more
1650 than one choice, and has a recorded value, and propose the same
1651 choice sequence with the span's choices replaced by a single
1652 ValueHole. At replay time the hole is claimed by whichever strategy
1653 draws at that position: a one_of's ``_invert`` tries every alternative
1654 in-process and returns the simplest encoding whose image contains the
1655 value, so one test execution covers all of its branches. If the
1656 result is a strictly simpler choice sequence which still fails, it is
1657 kept, and normal shrinking takes the re-encoded value the rest of the
1658 way. All knowledge of how to encode values lives with the strategies,
1659 not here.
1660 """
1661 node = chooser.choose(
1662 self.nodes,
1663 lambda n: (
1664 n.type == "integer"
1665 and not n.was_forced
1666 and n.constraints["min_value"] == 0
1667 and n.value != 0
1668 ),
1669 )
1671 span_idx = chooser.choose(
1672 self.spans_starting_at[node.index],
1673 # Single-choice spans are excluded: they are either not a one_of
1674 # (e.g. the selector draw's own span), or a one_of whose current
1675 # branch made no choices, like just() - and a re-encoding of the
1676 # latter would be longer, which incorporate_test_data rejects.
1677 lambda i: self.spans[i].recorded_value is not no_recorded_value
1678 and self.spans[i].choice_count > 1,
1679 )
1680 span = self.spans[span_idx]
1682 attempt = (
1683 self.choices[: span.start]
1684 + (ValueHole(span.recorded_value),)
1685 + self.choices[span.end :]
1686 )
1687 self.incorporate_test_data(self.engine.cached_test_function(attempt))
1689 def minimize_nodes(self, nodes):
1690 choice_type = nodes[0].type
1691 value = nodes[0].value
1692 # unlike choice_type and value, constraints are *not* guaranteed to be equal among all
1693 # passed nodes. We arbitrarily use the constraints of the first node. I think
1694 # this is unsound (= leads to us trying shrinks that could not have been
1695 # generated), but those get discarded at test-time, and this enables useful
1696 # slips where constraints are not equal but are close enough that doing the
1697 # same operation on both basically just works.
1698 constraints = nodes[0].constraints
1699 assert all(
1700 node.type == choice_type and choice_equal(node.value, value)
1701 for node in nodes
1702 )
1704 if choice_type == "integer":
1705 shrink_towards = constraints["shrink_towards"]
1706 # try shrinking from both sides towards shrink_towards.
1707 # we're starting from n = abs(shrink_towards - value). Because the
1708 # shrinker will not check its starting value, we need to try
1709 # shrinking to n first.
1710 self.try_shrinking_nodes(nodes, abs(shrink_towards - value))
1711 Integer.shrink(
1712 abs(shrink_towards - value),
1713 lambda n: self.try_shrinking_nodes(nodes, shrink_towards + n),
1714 )
1715 Integer.shrink(
1716 abs(shrink_towards - value),
1717 lambda n: self.try_shrinking_nodes(nodes, shrink_towards - n),
1718 )
1719 elif choice_type == "float":
1720 self.try_shrinking_nodes(nodes, abs(value))
1721 Float.shrink(
1722 abs(value),
1723 lambda val: self.try_shrinking_nodes(nodes, val),
1724 )
1725 Float.shrink(
1726 abs(value),
1727 lambda val: self.try_shrinking_nodes(nodes, -val),
1728 )
1729 elif choice_type == "boolean":
1730 # must be True, otherwise would be trivial and not selected.
1731 assert value is True
1732 # only one thing to try: false!
1733 self.try_shrinking_nodes(nodes, False)
1734 elif choice_type == "bytes":
1735 Bytes.shrink(
1736 value,
1737 lambda val: self.try_shrinking_nodes(nodes, val),
1738 min_size=constraints["min_size"],
1739 )
1740 elif choice_type == "string":
1741 String.shrink(
1742 value,
1743 lambda val: self.try_shrinking_nodes(nodes, val),
1744 intervals=constraints["intervals"],
1745 min_size=constraints["min_size"],
1746 )
1747 else:
1748 raise NotImplementedError
1750 def try_trivial_spans(self, chooser):
1751 i = chooser.choose(range(len(self.spans)))
1753 prev = self.shrink_target
1754 nodes = self.shrink_target.nodes
1755 span = self.spans[i]
1756 prefix = nodes[: span.start]
1757 replacement = tuple(
1758 [
1759 (
1760 node
1761 if node.was_forced
1762 else node.copy(
1763 with_value=choice_from_index(0, node.type, node.constraints)
1764 )
1765 )
1766 for node in nodes[span.start : span.end]
1767 ]
1768 )
1769 suffix = nodes[span.end :]
1770 attempt = self.cached_test_function(prefix + replacement + suffix)[1]
1772 if self.shrink_target is not prev:
1773 return
1775 if isinstance(attempt, ConjectureResult):
1776 new_span = attempt.spans[i]
1777 new_replacement = attempt.nodes[new_span.start : new_span.end]
1778 self.consider_new_nodes(prefix + new_replacement + suffix)
1780 def minimize_individual_choices(self, chooser):
1781 """Attempt to minimize each choice in sequence.
1783 This is the pass that ensures that e.g. each integer we draw is a
1784 minimum value. So it's the part that guarantees that if we e.g. do
1786 x = data.draw(integers())
1787 assert x < 10
1789 then in our shrunk test case, x = 10 rather than say 97.
1791 If we are unsuccessful at minimizing a choice of interest we then
1792 check if that's because it's changing the size of the test case and,
1793 if so, we also make an attempt to delete parts of the test case to
1794 see if that fixes it.
1796 We handle most of the common cases in try_shrinking_nodes which is
1797 pretty good at clearing out large contiguous blocks of dead space,
1798 but it fails when there is data that has to stay in particular places
1799 in the list.
1800 """
1801 node = chooser.choose(self.nodes, lambda node: not node.trivial)
1802 initial_target = self.shrink_target
1804 self.minimize_nodes([node])
1805 if self.shrink_target is not initial_target:
1806 # the shrink target changed, so our shrink worked. Defer doing
1807 # anything more intelligent until this shrink fails.
1808 return
1810 # the shrink failed. One particularly common case where minimizing a
1811 # node can fail is the antipattern of drawing a size and then drawing a
1812 # collection of that size, or more generally when there is a size
1813 # dependency on some single node. We'll explicitly try and fix up this
1814 # common case here: if decreasing an integer node by one would reduce
1815 # the size of the generated input, we'll try deleting things after that
1816 # node and see if the resulting attempt works.
1818 if node.type != "integer":
1819 # Only try this fixup logic on integer draws. Almost all size
1820 # dependencies are on integer draws, and if it's not, it's doing
1821 # something convoluted enough that it is unlikely to shrink well anyway.
1822 # TODO: extent to floats? we probably currently fail on the following,
1823 # albeit convoluted example:
1824 # n = int(data.draw(st.floats()))
1825 # s = data.draw(st.lists(st.integers(), min_size=n, max_size=n))
1826 return
1828 lowered = (
1829 self.nodes[: node.index]
1830 + (node.copy(with_value=node.value - 1),)
1831 + self.nodes[node.index + 1 :]
1832 )
1833 attempt = self.cached_test_function(lowered)[1]
1834 if (
1835 attempt is None
1836 or attempt.status < Status.VALID
1837 or len(attempt.nodes) == len(self.nodes)
1838 or len(attempt.nodes) == node.index + 1
1839 ):
1840 # no point in trying our size-dependency-logic if our attempt at
1841 # lowering the node resulted in:
1842 # * an invalid conjecture data
1843 # * the same number of nodes as before
1844 # * no nodes beyond the lowered node (nothing to try to delete afterwards)
1845 return
1847 # If it were then the original shrink should have worked and we could
1848 # never have got here.
1849 assert attempt is not self.shrink_target
1851 @self.cached(node.index)
1852 def first_span_after_node():
1853 lo = 0
1854 hi = len(self.spans)
1855 while lo + 1 < hi:
1856 mid = (lo + hi) // 2
1857 span = self.spans[mid]
1858 if span.start >= node.index:
1859 hi = mid
1860 else:
1861 lo = mid
1862 return hi
1864 # we try deleting both entire spans, and single nodes.
1865 # If we wanted to get more aggressive, we could try deleting n
1866 # consecutive nodes (that don't cross a span boundary) for say
1867 # n <= 2 or n <= 3.
1868 if chooser.choose([True, False]):
1869 span = self.spans[
1870 chooser.choose(
1871 range(first_span_after_node, len(self.spans)),
1872 lambda i: self.spans[i].choice_count > 0,
1873 )
1874 ]
1875 self.consider_new_nodes(lowered[: span.start] + lowered[span.end :])
1876 else:
1877 node = self.nodes[chooser.choose(range(node.index + 1, len(self.nodes)))]
1878 self.consider_new_nodes(lowered[: node.index] + lowered[node.index + 1 :])
1880 def reorder_spans(self, chooser):
1881 """This pass allows us to reorder the children of each span.
1883 For example, consider the following:
1885 .. code-block:: python
1887 import hypothesis.strategies as st
1888 from hypothesis import given
1891 @given(st.text(), st.text())
1892 def test_not_equal(x, y):
1893 assert x != y
1895 Without the ability to reorder x and y this could fail either with
1896 ``x=""``, ``y="0"``, or the other way around. With reordering it will
1897 reliably fail with ``x=""``, ``y="0"``.
1898 """
1899 span = chooser.choose(self.spans)
1901 label = chooser.choose(span.children).label
1902 spans = [c for c in span.children if c.label == label]
1903 if len(spans) <= 1:
1904 return
1906 endpoints = [(span.start, span.end) for span in spans]
1907 st = self.shrink_target
1909 Ordering.shrink(
1910 range(len(spans)),
1911 lambda indices: self.consider_new_nodes(
1912 replace_all(
1913 st.nodes,
1914 [
1915 (
1916 u,
1917 v,
1918 st.nodes[spans[i].start : spans[i].end],
1919 )
1920 for (u, v), i in zip(endpoints, indices, strict=True)
1921 ],
1922 )
1923 ),
1924 key=lambda i: sort_key(st.nodes[spans[i].start : spans[i].end]),
1925 )
1927 def run_node_program(self, i, program, original, repeats=1):
1928 """Node programs are a mini-DSL for node rewriting, defined as a sequence
1929 of commands that can be run at some index into the nodes
1931 Commands are:
1933 * "X", delete this node
1935 This method runs the node program in ``program`` at node index
1936 ``i`` on the ConjectureData ``original``. If ``repeats > 1`` then it
1937 will attempt to approximate the results of running it that many times.
1939 Returns True if this successfully changes the underlying shrink target,
1940 else False.
1941 """
1942 if i + len(program) > len(original.nodes) or i < 0:
1943 return False
1944 attempt = list(original.nodes)
1945 for _ in range(repeats):
1946 for k, command in reversed(list(enumerate(program))):
1947 j = i + k
1948 if j >= len(attempt):
1949 return False
1951 if command == "X":
1952 del attempt[j]
1953 else:
1954 raise NotImplementedError(f"Unrecognised command {command!r}")
1956 return self.consider_new_nodes(attempt)