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
12from collections import defaultdict
13from collections.abc import Callable, Sequence
14from dataclasses import dataclass
15from typing import (
16 TYPE_CHECKING,
17 Any,
18 Literal,
19 TypeAlias,
20 cast,
21)
23from hypothesis.internal.conjecture.choice import (
24 ChoiceNode,
25 ChoiceT,
26 choice_equal,
27 choice_from_index,
28 choice_key,
29 choice_permitted,
30 choice_to_index,
31)
32from hypothesis.internal.conjecture.data import (
33 ConjectureData,
34 ConjectureResult,
35 Spans,
36 Status,
37 _Overrun,
38 draw_choice,
39)
40from hypothesis.internal.conjecture.junkdrawer import (
41 endswith,
42 find_integer,
43 replace_all,
44 startswith,
45)
46from hypothesis.internal.conjecture.shrinking import (
47 Bytes,
48 Float,
49 Integer,
50 Ordering,
51 String,
52)
53from hypothesis.internal.conjecture.shrinking.choicetree import (
54 ChoiceTree,
55 prefix_selection_order,
56 random_selection_order,
57)
58from hypothesis.internal.floats import MAX_PRECISE_INTEGER
60if TYPE_CHECKING:
61 from random import Random
63 from hypothesis.internal.conjecture.engine import ConjectureRunner
65ShrinkPredicateT: TypeAlias = Callable[[ConjectureResult | _Overrun], bool]
68def sort_key(nodes: Sequence[ChoiceNode]) -> tuple[int, tuple[int, ...]]:
69 """Returns a sort key such that "simpler" choice sequences are smaller than
70 "more complicated" ones.
72 We define sort_key so that x is simpler than y if x is shorter than y or if
73 they have the same length and map(choice_to_index, x) < map(choice_to_index, y).
75 The reason for using this ordering is:
77 1. If x is shorter than y then that means we had to make fewer decisions
78 in constructing the test case when we ran x than we did when we ran y.
79 2. If x is the same length as y then replacing a choice with a lower index
80 choice corresponds to replacing it with a simpler/smaller choice.
81 3. Because choices drawn early in generation potentially get used in more
82 places they potentially have a more significant impact on the final
83 result, so it makes sense to prioritise reducing earlier choices over
84 later ones.
85 """
86 return (
87 len(nodes),
88 tuple(choice_to_index(node.value, node.constraints) for node in nodes),
89 )
92@dataclass(slots=True, frozen=False)
93class ShrinkPass:
94 function: Any
95 name: str | None = None
96 last_prefix: Any = ()
98 # some execution statistics
99 calls: int = 0
100 misaligned: int = 0
101 shrinks: int = 0
102 deletions: int = 0
104 def __post_init__(self):
105 if self.name is None:
106 self.name = self.function.__name__
108 def __hash__(self):
109 return hash(self.name)
112class StopShrinking(Exception):
113 pass
116class Shrinker:
117 """A shrinker is a child object of a ConjectureRunner which is designed to
118 manage the associated state of a particular shrink problem. That is, we
119 have some initial ConjectureData object and some property of interest
120 that it satisfies, and we want to find a ConjectureData object with a
121 shortlex (see sort_key above) smaller choice sequence that exhibits the same
122 property.
124 Currently the only property of interest we use is that the status is
125 INTERESTING and the interesting_origin takes on some fixed value, but we
126 may potentially be interested in other use cases later.
127 However we assume that data with a status < VALID never satisfies the predicate.
129 The shrinker keeps track of a value shrink_target which represents the
130 current best known ConjectureData object satisfying the predicate.
131 It refines this value by repeatedly running *shrink passes*, which are
132 methods that perform a series of transformations to the current shrink_target
133 and evaluate the underlying test function to find new ConjectureData
134 objects. If any of these satisfy the predicate, the shrink_target
135 is updated automatically. Shrinking runs until no shrink pass can
136 improve the shrink_target, at which point it stops. It may also be
137 terminated if the underlying engine throws RunIsComplete, but that
138 is handled by the calling code rather than the Shrinker.
140 =======================
141 Designing Shrink Passes
142 =======================
144 Generally a shrink pass is just any function that calls
145 cached_test_function and/or consider_new_nodes a number of times,
146 but there are a couple of useful things to bear in mind.
148 A shrink pass *makes progress* if running it changes self.shrink_target
149 (i.e. it tries a shortlex smaller ConjectureData object satisfying
150 the predicate). The desired end state of shrinking is to find a
151 value such that no shrink pass can make progress, i.e. that we
152 are at a local minimum for each shrink pass.
154 In aid of this goal, the main invariant that a shrink pass much
155 satisfy is that whether it makes progress must be deterministic.
156 It is fine (encouraged even) for the specific progress it makes
157 to be non-deterministic, but if you run a shrink pass, it makes
158 no progress, and then you immediately run it again, it should
159 never succeed on the second time. This allows us to stop as soon
160 as we have run each shrink pass and seen no progress on any of
161 them.
163 This means that e.g. it's fine to try each of N deletions
164 or replacements in a random order, but it's not OK to try N random
165 deletions (unless you have already shrunk at least once, though we
166 don't currently take advantage of this loophole).
168 Shrink passes need to be written so as to be robust against
169 change in the underlying shrink target. It is generally safe
170 to assume that the shrink target does not change prior to the
171 point of first modification - e.g. if you change no bytes at
172 index ``i``, all spans whose start is ``<= i`` still exist,
173 as do all blocks, and the data object is still of length
174 ``>= i + 1``. This can only be violated by bad user code which
175 relies on an external source of non-determinism.
177 When the underlying shrink_target changes, shrink
178 passes should not run substantially more test_function calls
179 on success than they do on failure. Say, no more than a constant
180 factor more. In particular shrink passes should not iterate to a
181 fixed point.
183 This means that shrink passes are often written with loops that
184 are carefully designed to do the right thing in the case that no
185 shrinks occurred and try to adapt to any changes to do a reasonable
186 job. e.g. say we wanted to write a shrink pass that tried deleting
187 each individual choice (this isn't an especially good pass,
188 but it leads to a simple illustrative example), we might do it
189 by iterating over the choice sequence like so:
191 .. code-block:: python
193 i = 0
194 while i < len(self.shrink_target.nodes):
195 if not self.consider_new_nodes(
196 self.shrink_target.nodes[:i] + self.shrink_target.nodes[i + 1 :]
197 ):
198 i += 1
200 The reason for writing the loop this way is that i is always a
201 valid index into the current choice sequence, even if the current sequence
202 changes as a result of our actions. When the choice sequence changes,
203 we leave the index where it is rather than restarting from the
204 beginning, and carry on. This means that the number of steps we
205 run in this case is always bounded above by the number of steps
206 we would run if nothing works.
208 Another thing to bear in mind about shrink pass design is that
209 they should prioritise *progress*. If you have N operations that
210 you need to run, you should try to order them in such a way as
211 to avoid stalling, where you have long periods of test function
212 invocations where no shrinks happen. This is bad because whenever
213 we shrink we reduce the amount of work the shrinker has to do
214 in future, and often speed up the test function, so we ideally
215 wanted those shrinks to happen much earlier in the process.
217 Sometimes stalls are inevitable of course - e.g. if the pass
218 makes no progress, then the entire thing is just one long stall,
219 but it's helpful to design it so that stalls are less likely
220 in typical behaviour.
222 The two easiest ways to do this are:
224 * Just run the N steps in random order. As long as a
225 reasonably large proportion of the operations succeed, this
226 guarantees the expected stall length is quite short. The
227 book keeping for making sure this does the right thing when
228 it succeeds can be quite annoying.
229 * When you have any sort of nested loop, loop in such a way
230 that both loop variables change each time. This prevents
231 stalls which occur when one particular value for the outer
232 loop is impossible to make progress on, rendering the entire
233 inner loop into a stall.
235 However, although progress is good, too much progress can be
236 a bad sign! If you're *only* seeing successful reductions,
237 that's probably a sign that you are making changes that are
238 too timid. Two useful things to offset this:
240 * It's worth writing shrink passes which are *adaptive*, in
241 the sense that when operations seem to be working really
242 well we try to bundle multiple of them together. This can
243 often be used to turn what would be O(m) successful calls
244 into O(log(m)).
245 * It's often worth trying one or two special minimal values
246 before trying anything more fine grained (e.g. replacing
247 the whole thing with zero).
249 """
251 def derived_value(fn):
252 """It's useful during shrinking to have access to derived values of
253 the current shrink target.
255 This decorator allows you to define these as cached properties. They
256 are calculated once, then cached until the shrink target changes, then
257 recalculated the next time they are used."""
259 def accept(self):
260 try:
261 return self.__derived_values[fn.__name__]
262 except KeyError:
263 return self.__derived_values.setdefault(fn.__name__, fn(self))
265 accept.__name__ = fn.__name__
266 return property(accept)
268 def __init__(
269 self,
270 engine: "ConjectureRunner",
271 initial: ConjectureData | ConjectureResult,
272 predicate: ShrinkPredicateT | None,
273 *,
274 allow_transition: (
275 Callable[[ConjectureData | ConjectureResult, ConjectureData], bool] | None
276 ),
277 explain: bool,
278 in_target_phase: bool = False,
279 ):
280 """Create a shrinker for a particular engine, with a given starting
281 point and predicate. When shrink() is called it will attempt to find an
282 example for which predicate is True and which is strictly smaller than
283 initial.
285 Note that initial is a ConjectureData object, and predicate
286 takes ConjectureData objects.
287 """
288 assert predicate is not None or allow_transition is not None
289 self.engine = engine
290 self.__predicate = predicate or (lambda data: True)
291 self.__allow_transition = allow_transition or (lambda source, destination: True)
292 self.__derived_values: dict = {}
294 self.initial_size = len(initial.choices)
295 # We keep track of the current best example on the shrink_target
296 # attribute.
297 self.shrink_target = initial
298 self.clear_change_tracking()
299 self.shrinks = 0
301 # We terminate shrinks that seem to have reached their logical
302 # conclusion: If we've called the underlying test function at
303 # least self.max_stall times since the last time we shrunk,
304 # it's time to stop shrinking.
305 self.max_stall = 200
306 self.initial_calls = self.engine.call_count
307 self.initial_misaligned = self.engine.misaligned_count
308 self.calls_at_last_shrink = self.initial_calls
310 self.shrink_passes: list[ShrinkPass] = [
311 ShrinkPass(self.try_trivial_spans),
312 self.node_program("X" * 5),
313 self.node_program("X" * 4),
314 self.node_program("X" * 3),
315 self.node_program("X" * 2),
316 self.node_program("X" * 1),
317 ShrinkPass(self.pass_to_descendant),
318 ShrinkPass(self.reorder_spans),
319 ShrinkPass(self.minimize_duplicated_choices),
320 ShrinkPass(self.minimize_individual_choices),
321 ShrinkPass(self.redistribute_numeric_pairs),
322 ShrinkPass(self.lower_integers_together),
323 ShrinkPass(self.lower_duplicated_characters),
324 ]
326 # Because the shrinker is also used to `pareto_optimise` in the target phase,
327 # we sometimes want to allow extending buffers instead of aborting at the end.
328 self.__extend: Literal["full"] | int = "full" if in_target_phase else 0
329 self.should_explain = explain
331 @derived_value # type: ignore
332 def cached_calculations(self):
333 return {}
335 def cached(self, *keys):
336 def accept(f):
337 cache_key = (f.__name__, *keys)
338 try:
339 return self.cached_calculations[cache_key]
340 except KeyError:
341 return self.cached_calculations.setdefault(cache_key, f())
343 return accept
345 @property
346 def calls(self) -> int:
347 """Return the number of calls that have been made to the underlying
348 test function."""
349 return self.engine.call_count
351 @property
352 def misaligned(self) -> int:
353 return self.engine.misaligned_count
355 def check_calls(self) -> None:
356 if self.calls - self.calls_at_last_shrink >= self.max_stall:
357 raise StopShrinking
359 def cached_test_function(
360 self, nodes: Sequence[ChoiceNode]
361 ) -> tuple[bool, ConjectureResult | _Overrun | None]:
362 nodes = nodes[: len(self.nodes)]
364 if startswith(nodes, self.nodes):
365 return (True, None)
367 if sort_key(self.nodes) < sort_key(nodes):
368 return (False, None)
370 # sometimes our shrinking passes try obviously invalid things. We handle
371 # discarding them in one place here.
372 if any(not choice_permitted(node.value, node.constraints) for node in nodes):
373 return (False, None)
375 result = self.engine.cached_test_function(
376 [n.value for n in nodes], extend=self.__extend
377 )
378 previous = self.shrink_target
379 self.incorporate_test_data(result)
380 self.check_calls()
381 return (previous is not self.shrink_target, result)
383 def consider_new_nodes(self, nodes: Sequence[ChoiceNode]) -> bool:
384 return self.cached_test_function(nodes)[0]
386 def incorporate_test_data(self, data):
387 """Takes a ConjectureData or Overrun object updates the current
388 shrink_target if this data represents an improvement over it."""
389 if data.status < Status.VALID or data is self.shrink_target:
390 return
391 if (
392 self.__predicate(data)
393 and sort_key(data.nodes) < sort_key(self.shrink_target.nodes)
394 and self.__allow_transition(self.shrink_target, data)
395 ):
396 self.update_shrink_target(data)
398 def debug(self, msg: str) -> None:
399 self.engine.debug(msg)
401 @property
402 def random(self) -> "Random":
403 return self.engine.random
405 def shrink(self) -> None:
406 """Run the full set of shrinks and update shrink_target.
408 This method is "mostly idempotent" - calling it twice is unlikely to
409 have any effect, though it has a non-zero probability of doing so.
410 """
412 try:
413 self.initial_coarse_reduction()
414 self.greedy_shrink()
415 except StopShrinking:
416 # If we stopped shrinking because we're making slow progress (instead of
417 # reaching a local optimum), don't run the explain-phase logic.
418 self.should_explain = False
419 finally:
420 if self.engine.report_debug_info:
422 def s(n):
423 return "s" if n != 1 else ""
425 total_deleted = self.initial_size - len(self.shrink_target.choices)
426 calls = self.engine.call_count - self.initial_calls
427 misaligned = self.engine.misaligned_count - self.initial_misaligned
429 self.debug(
430 "---------------------\n"
431 "Shrink pass profiling\n"
432 "---------------------\n\n"
433 f"Shrinking made a total of {calls} call{s(calls)} of which "
434 f"{self.shrinks} shrank and {misaligned} were misaligned. This "
435 f"deleted {total_deleted} choices out of {self.initial_size}."
436 )
437 for useful in [True, False]:
438 self.debug("")
439 if useful:
440 self.debug("Useful passes:")
441 else:
442 self.debug("Useless passes:")
443 self.debug("")
444 for pass_ in sorted(
445 self.shrink_passes,
446 key=lambda t: (-t.calls, t.deletions, t.shrinks),
447 ):
448 if pass_.calls == 0:
449 continue
450 if (pass_.shrinks != 0) != useful:
451 continue
453 self.debug(
454 f" * {pass_.name} made {pass_.calls} call{s(pass_.calls)} of which "
455 f"{pass_.shrinks} shrank and {pass_.misaligned} were misaligned, "
456 f"deleting {pass_.deletions} choice{s(pass_.deletions)}."
457 )
458 self.debug("")
459 self.explain()
461 def explain(self) -> None:
463 if not self.should_explain or not self.shrink_target.arg_slices:
464 return
466 self.max_stall = 2**100
467 shrink_target = self.shrink_target
468 nodes = self.nodes
469 choices = self.choices
470 chunks: dict[tuple[int, int], list[tuple[ChoiceT, ...]]] = defaultdict(list)
472 # Before we start running experiments, let's check for known inputs which would
473 # make them redundant. The shrinking process means that we've already tried many
474 # variations on the minimal example, so this can save a lot of time.
475 seen_passing_seq = self.engine.passing_choice_sequences(
476 prefix=self.nodes[: min(self.shrink_target.arg_slices)[0]]
477 )
479 # Now that we've shrunk to a minimal failing example, it's time to try
480 # varying each part that we've noted will go in the final report. Consider
481 # slices in largest-first order
482 for start, end in sorted(
483 self.shrink_target.arg_slices, key=lambda x: (-(x[1] - x[0]), x)
484 ):
485 # Check for any previous examples that match the prefix and suffix,
486 # so we can skip if we found a passing example while shrinking.
487 if any(
488 startswith(seen, nodes[:start]) and endswith(seen, nodes[end:])
489 for seen in seen_passing_seq
490 ):
491 continue
493 # Run our experiments
494 n_same_failures = 0
495 note = "or any other generated value"
496 # TODO: is 100 same-failures out of 500 attempts a good heuristic?
497 for n_attempt in range(500): # pragma: no branch
498 # no-branch here because we don't coverage-test the abort-at-500 logic.
500 if n_attempt - 10 > n_same_failures * 5:
501 # stop early if we're seeing mostly invalid examples
502 break # pragma: no cover
504 # replace start:end with random values
505 replacement = []
506 for i in range(start, end):
507 node = nodes[i]
508 if not node.was_forced:
509 value = draw_choice(
510 node.type, node.constraints, random=self.random
511 )
512 node = node.copy(with_value=value)
513 replacement.append(node.value)
515 attempt = choices[:start] + tuple(replacement) + choices[end:]
516 result = self.engine.cached_test_function(attempt, extend="full")
518 if result.status is Status.OVERRUN:
519 continue # pragma: no cover # flakily covered
520 result = cast(ConjectureResult, result)
521 if not (
522 len(attempt) == len(result.choices)
523 and endswith(result.nodes, nodes[end:])
524 ):
525 # Turns out this was a variable-length part, so grab the infix...
526 for span1, span2 in zip(
527 shrink_target.spans, result.spans, strict=False
528 ):
529 assert span1.start == span2.start
530 assert span1.start <= start
531 assert span1.label == span2.label
532 if span1.start == start and span1.end == end:
533 result_end = span2.end
534 break
535 else:
536 raise NotImplementedError("Expected matching prefixes")
538 attempt = (
539 choices[:start]
540 + result.choices[start:result_end]
541 + choices[end:]
542 )
543 chunks[(start, end)].append(result.choices[start:result_end])
544 result = self.engine.cached_test_function(attempt)
546 if result.status is Status.OVERRUN:
547 continue # pragma: no cover # flakily covered
548 result = cast(ConjectureResult, result)
549 else:
550 chunks[(start, end)].append(result.choices[start:end])
552 if shrink_target is not self.shrink_target: # pragma: no cover
553 # If we've shrunk further without meaning to, bail out.
554 self.shrink_target.slice_comments.clear()
555 return
556 if result.status is Status.VALID:
557 # The test passed, indicating that this param can't vary freely.
558 # However, it's really hard to write a simple and reliable covering
559 # test, because of our `seen_passing_buffers` check above.
560 break # pragma: no cover
561 if self.__predicate(result): # pragma: no branch
562 n_same_failures += 1
563 if n_same_failures >= 100:
564 self.shrink_target.slice_comments[(start, end)] = note
565 break
567 # Finally, if we've found multiple independently-variable parts, check whether
568 # they can all be varied together.
569 if len(self.shrink_target.slice_comments) <= 1:
570 return
571 n_same_failures_together = 0
572 chunks_by_start_index = sorted(chunks.items())
573 for _ in range(500): # pragma: no branch
574 # no-branch here because we don't coverage-test the abort-at-500 logic.
575 new_choices: list[ChoiceT] = []
576 prev_end = 0
577 for (start, end), ls in chunks_by_start_index:
578 assert prev_end <= start < end, "these chunks must be nonoverlapping"
579 new_choices.extend(choices[prev_end:start])
580 new_choices.extend(self.random.choice(ls))
581 prev_end = end
583 result = self.engine.cached_test_function(new_choices)
585 # This *can't* be a shrink because none of the components were.
586 assert shrink_target is self.shrink_target
587 if result.status == Status.VALID:
588 self.shrink_target.slice_comments[(0, 0)] = (
589 "The test sometimes passed when commented parts were varied together."
590 )
591 break # Test passed, this param can't vary freely.
592 if self.__predicate(result): # pragma: no branch
593 n_same_failures_together += 1
594 if n_same_failures_together >= 100:
595 self.shrink_target.slice_comments[(0, 0)] = (
596 "The test always failed when commented parts were varied together."
597 )
598 break
600 def greedy_shrink(self) -> None:
601 """Run a full set of greedy shrinks (that is, ones that will only ever
602 move to a better target) and update shrink_target appropriately.
604 This method iterates to a fixed point and so is idempontent - calling
605 it twice will have exactly the same effect as calling it once.
606 """
607 self.fixate_shrink_passes(self.shrink_passes)
609 def initial_coarse_reduction(self):
610 """Performs some preliminary reductions that should not be
611 repeated as part of the main shrink passes.
613 The main reason why these can't be included as part of shrink
614 passes is that they have much more ability to make the test
615 case "worse". e.g. they might rerandomise part of it, significantly
616 increasing the value of individual nodes, which works in direct
617 opposition to the lexical shrinking and will frequently undo
618 its work.
619 """
620 self.reduce_each_alternative()
622 @derived_value # type: ignore
623 def spans_starting_at(self):
624 result = [[] for _ in self.shrink_target.nodes]
625 for i, ex in enumerate(self.spans):
626 # We can have zero-length spans that start at the end
627 if ex.start < len(result):
628 result[ex.start].append(i)
629 return tuple(map(tuple, result))
631 def reduce_each_alternative(self):
632 """This is a pass that is designed to rerandomise use of the
633 one_of strategy or things that look like it, in order to try
634 to move from later strategies to earlier ones in the branch
635 order.
637 It does this by trying to systematically lower each value it
638 finds that looks like it might be the branch decision for
639 one_of, and then attempts to repair any changes in shape that
640 this causes.
641 """
642 i = 0
643 while i < len(self.shrink_target.nodes):
644 nodes = self.shrink_target.nodes
645 node = nodes[i]
646 if (
647 node.type == "integer"
648 and not node.was_forced
649 and node.value <= 10
650 and node.constraints["min_value"] == 0
651 ):
652 assert isinstance(node.value, int)
654 # We've found a plausible candidate for a ``one_of`` choice.
655 # We now want to see if the shape of the test case actually depends
656 # on it. If it doesn't, then we don't need to do this (comparatively
657 # costly) pass, and can let much simpler lexicographic reduction
658 # handle it later.
659 #
660 # We test this by trying to set the value to zero and seeing if the
661 # shape changes, as measured by either changing the number of subsequent
662 # nodes, or changing the nodes in such a way as to cause one of the
663 # previous values to no longer be valid in its position.
664 zero_attempt = self.cached_test_function(
665 nodes[:i] + (nodes[i].copy(with_value=0),) + nodes[i + 1 :]
666 )[1]
667 if (
668 zero_attempt is not self.shrink_target
669 and zero_attempt is not None
670 and zero_attempt.status >= Status.VALID
671 ):
672 changed_shape = len(zero_attempt.nodes) != len(nodes)
674 if not changed_shape:
675 for j in range(i + 1, len(nodes)):
676 zero_node = zero_attempt.nodes[j]
677 orig_node = nodes[j]
678 if (
679 zero_node.type != orig_node.type
680 or not choice_permitted(
681 orig_node.value, zero_node.constraints
682 )
683 ):
684 changed_shape = True
685 break
686 if changed_shape:
687 for v in range(node.value):
688 if self.try_lower_node_as_alternative(i, v):
689 break
690 i += 1
692 def try_lower_node_as_alternative(self, i, v):
693 """Attempt to lower `self.shrink_target.nodes[i]` to `v`,
694 while rerandomising and attempting to repair any subsequent
695 changes to the shape of the test case that this causes."""
696 nodes = self.shrink_target.nodes
697 if self.consider_new_nodes(
698 nodes[:i] + (nodes[i].copy(with_value=v),) + nodes[i + 1 :]
699 ):
700 return True
702 prefix = nodes[:i] + (nodes[i].copy(with_value=v),)
703 initial = self.shrink_target
704 spans = self.spans_starting_at[i]
705 for _ in range(3):
706 random_attempt = self.engine.cached_test_function(
707 [n.value for n in prefix], extend=len(nodes)
708 )
709 if random_attempt.status < Status.VALID:
710 continue
711 self.incorporate_test_data(random_attempt)
712 for j in spans:
713 initial_ex = initial.spans[j]
714 attempt_ex = random_attempt.spans[j]
715 contents = random_attempt.nodes[attempt_ex.start : attempt_ex.end]
716 self.consider_new_nodes(nodes[:i] + contents + nodes[initial_ex.end :])
717 if initial is not self.shrink_target:
718 return True
719 return False
721 @derived_value # type: ignore
722 def shrink_pass_choice_trees(self) -> dict[Any, ChoiceTree]:
723 return defaultdict(ChoiceTree)
725 def step(self, shrink_pass: ShrinkPass, *, random_order: bool = False) -> bool:
726 tree = self.shrink_pass_choice_trees[shrink_pass]
727 if tree.exhausted:
728 return False
730 initial_shrinks = self.shrinks
731 initial_calls = self.calls
732 initial_misaligned = self.misaligned
733 size = len(self.shrink_target.choices)
734 assert shrink_pass.name is not None
735 self.engine.explain_next_call_as(shrink_pass.name)
737 if random_order:
738 selection_order = random_selection_order(self.random)
739 else:
740 selection_order = prefix_selection_order(shrink_pass.last_prefix)
742 try:
743 shrink_pass.last_prefix = tree.step(
744 selection_order,
745 lambda chooser: shrink_pass.function(chooser),
746 )
747 finally:
748 shrink_pass.calls += self.calls - initial_calls
749 shrink_pass.misaligned += self.misaligned - initial_misaligned
750 shrink_pass.shrinks += self.shrinks - initial_shrinks
751 shrink_pass.deletions += size - len(self.shrink_target.choices)
752 self.engine.clear_call_explanation()
753 return True
755 def fixate_shrink_passes(self, passes: list[ShrinkPass]) -> None:
756 """Run steps from each pass in ``passes`` until the current shrink target
757 is a fixed point of all of them."""
758 any_ran = True
759 while any_ran:
760 any_ran = False
762 reordering = {}
764 # We run remove_discarded after every pass to do cleanup
765 # keeping track of whether that actually works. Either there is
766 # no discarded data and it is basically free, or it reliably works
767 # and deletes data, or it doesn't work. In that latter case we turn
768 # it off for the rest of this loop through the passes, but will
769 # try again once all of the passes have been run.
770 can_discard = self.remove_discarded()
772 calls_at_loop_start = self.calls
774 # We keep track of how many calls can be made by a single step
775 # without making progress and use this to test how much to pad
776 # out self.max_stall by as we go along.
777 max_calls_per_failing_step = 1
779 for sp in passes:
780 if can_discard:
781 can_discard = self.remove_discarded()
783 before_sp = self.shrink_target
785 # Run the shrink pass until it fails to make any progress
786 # max_failures times in a row. This implicitly boosts shrink
787 # passes that are more likely to work.
788 failures = 0
789 max_failures = 20
790 while failures < max_failures:
791 # We don't allow more than max_stall consecutive failures
792 # to shrink, but this means that if we're unlucky and the
793 # shrink passes are in a bad order where only the ones at
794 # the end are useful, if we're not careful this heuristic
795 # might stop us before we've tried everything. In order to
796 # avoid that happening, we make sure that there's always
797 # plenty of breathing room to make it through a single
798 # iteration of the fixate_shrink_passes loop.
799 self.max_stall = max(
800 self.max_stall,
801 2 * max_calls_per_failing_step
802 + (self.calls - calls_at_loop_start),
803 )
805 prev = self.shrink_target
806 initial_calls = self.calls
807 # It's better for us to run shrink passes in a deterministic
808 # order, to avoid repeat work, but this can cause us to create
809 # long stalls when there are a lot of steps which fail to do
810 # anything useful. In order to avoid this, once we've noticed
811 # we're in a stall (i.e. half of max_failures calls have failed
812 # to do anything) we switch to randomly jumping around. If we
813 # find a success then we'll resume deterministic order from
814 # there which, with any luck, is in a new good region.
815 if not self.step(sp, random_order=failures >= max_failures // 2):
816 # step returns False when there is nothing to do because
817 # the entire choice tree is exhausted. If this happens
818 # we break because we literally can't run this pass any
819 # more than we already have until something else makes
820 # progress.
821 break
822 any_ran = True
824 # Don't count steps that didn't actually try to do
825 # anything as failures. Otherwise, this call is a failure
826 # if it failed to make any changes to the shrink target.
827 if initial_calls != self.calls:
828 if prev is not self.shrink_target:
829 failures = 0
830 else:
831 max_calls_per_failing_step = max(
832 max_calls_per_failing_step, self.calls - initial_calls
833 )
834 failures += 1
836 # We reorder the shrink passes so that on our next run through
837 # we try good ones first. The rule is that shrink passes that
838 # did nothing useful are the worst, shrink passes that reduced
839 # the length are the best.
840 if self.shrink_target is before_sp:
841 reordering[sp] = 1
842 elif len(self.choices) < len(before_sp.choices):
843 reordering[sp] = -1
844 else:
845 reordering[sp] = 0
847 passes.sort(key=reordering.__getitem__)
849 @property
850 def nodes(self) -> tuple[ChoiceNode, ...]:
851 return self.shrink_target.nodes
853 @property
854 def choices(self) -> tuple[ChoiceT, ...]:
855 return self.shrink_target.choices
857 @property
858 def spans(self) -> Spans:
859 return self.shrink_target.spans
861 @derived_value # type: ignore
862 def spans_by_label(self):
863 """
864 A mapping of labels to a list of spans with that label. Spans in the list
865 are ordered by their normal index order.
866 """
868 spans_by_label = defaultdict(list)
869 for ex in self.spans:
870 spans_by_label[ex.label].append(ex)
871 return dict(spans_by_label)
873 @derived_value # type: ignore
874 def distinct_labels(self):
875 return sorted(self.spans_by_label, key=str)
877 def pass_to_descendant(self, chooser):
878 """Attempt to replace each span with a descendant span.
880 This is designed to deal with strategies that call themselves
881 recursively. For example, suppose we had:
883 binary_tree = st.deferred(
884 lambda: st.one_of(
885 st.integers(), st.tuples(binary_tree, binary_tree)))
887 This pass guarantees that we can replace any binary tree with one of
888 its subtrees - each of those will create an interval that the parent
889 could validly be replaced with, and this pass will try doing that.
891 This is pretty expensive - it takes O(len(intervals)^2) - so we run it
892 late in the process when we've got the number of intervals as far down
893 as possible.
894 """
896 label = chooser.choose(
897 self.distinct_labels, lambda l: len(self.spans_by_label[l]) >= 2
898 )
900 spans = self.spans_by_label[label]
901 i = chooser.choose(range(len(spans) - 1))
902 ancestor = spans[i]
904 if i + 1 == len(spans) or spans[i + 1].start >= ancestor.end:
905 return
907 @self.cached(label, i)
908 def descendants():
909 lo = i + 1
910 hi = len(spans)
911 while lo + 1 < hi:
912 mid = (lo + hi) // 2
913 if spans[mid].start >= ancestor.end:
914 hi = mid
915 else:
916 lo = mid
917 return [
918 span
919 for span in spans[i + 1 : hi]
920 if span.choice_count < ancestor.choice_count
921 ]
923 descendant = chooser.choose(descendants, lambda ex: ex.choice_count > 0)
925 assert ancestor.start <= descendant.start
926 assert ancestor.end >= descendant.end
927 assert descendant.choice_count < ancestor.choice_count
929 self.consider_new_nodes(
930 self.nodes[: ancestor.start]
931 + self.nodes[descendant.start : descendant.end]
932 + self.nodes[ancestor.end :]
933 )
935 def lower_common_node_offset(self):
936 """Sometimes we find ourselves in a situation where changes to one part
937 of the choice sequence unlock changes to other parts. Sometimes this is
938 good, but sometimes this can cause us to exhibit exponential slow
939 downs!
941 e.g. suppose we had the following:
943 m = draw(integers(min_value=0))
944 n = draw(integers(min_value=0))
945 assert abs(m - n) > 1
947 If this fails then we'll end up with a loop where on each iteration we
948 reduce each of m and n by 2 - m can't go lower because of n, then n
949 can't go lower because of m.
951 This will take us O(m) iterations to complete, which is exponential in
952 the data size, as we gradually zig zag our way towards zero.
954 This can only happen if we're failing to reduce the size of the choice
955 sequence: The number of iterations that reduce the length of the choice
956 sequence is bounded by that length.
958 So what we do is this: We keep track of which nodes are changing, and
959 then if there's some non-zero common offset to them we try and minimize
960 them all at once by lowering that offset.
962 This may not work, and it definitely won't get us out of all possible
963 exponential slow downs (an example of where it doesn't is where the
964 shape of the nodes changes as a result of this bouncing behaviour),
965 but it fails fast when it doesn't work and gets us out of a really
966 nastily slow case when it does.
967 """
968 if len(self.__changed_nodes) <= 1:
969 return
971 changed = []
972 for i in sorted(self.__changed_nodes):
973 node = self.nodes[i]
974 if node.trivial or node.type != "integer":
975 continue
976 changed.append(node)
978 if not changed:
979 return
981 ints = [
982 abs(node.value - node.constraints["shrink_towards"]) for node in changed
983 ]
984 offset = min(ints)
985 assert offset > 0
987 for i in range(len(ints)):
988 ints[i] -= offset
990 st = self.shrink_target
992 def offset_node(node, n):
993 return (
994 node.index,
995 node.index + 1,
996 [node.copy(with_value=node.constraints["shrink_towards"] + n)],
997 )
999 def consider(n, sign):
1000 return self.consider_new_nodes(
1001 replace_all(
1002 st.nodes,
1003 [
1004 offset_node(node, sign * (n + v))
1005 for node, v in zip(changed, ints, strict=False)
1006 ],
1007 )
1008 )
1010 # shrink from both sides
1011 Integer.shrink(offset, lambda n: consider(n, 1))
1012 Integer.shrink(offset, lambda n: consider(n, -1))
1013 self.clear_change_tracking()
1015 def clear_change_tracking(self):
1016 self.__last_checked_changed_at = self.shrink_target
1017 self.__all_changed_nodes = set()
1019 def mark_changed(self, i):
1020 self.__changed_nodes.add(i)
1022 @property
1023 def __changed_nodes(self) -> set[int]:
1024 if self.__last_checked_changed_at is self.shrink_target:
1025 return self.__all_changed_nodes
1027 prev_target = self.__last_checked_changed_at
1028 new_target = self.shrink_target
1029 assert prev_target is not new_target
1030 prev_nodes = prev_target.nodes
1031 new_nodes = new_target.nodes
1032 assert sort_key(new_target.nodes) < sort_key(prev_target.nodes)
1034 if len(prev_nodes) != len(new_nodes) or any(
1035 n1.type != n2.type for n1, n2 in zip(prev_nodes, new_nodes, strict=True)
1036 ):
1037 # should we check constraints are equal as well?
1038 self.__all_changed_nodes = set()
1039 else:
1040 assert len(prev_nodes) == len(new_nodes)
1041 for i, (n1, n2) in enumerate(zip(prev_nodes, new_nodes, strict=True)):
1042 assert n1.type == n2.type
1043 if not choice_equal(n1.value, n2.value):
1044 self.__all_changed_nodes.add(i)
1046 return self.__all_changed_nodes
1048 def update_shrink_target(self, new_target):
1049 assert isinstance(new_target, ConjectureResult)
1050 self.shrinks += 1
1051 # If we are just taking a long time to shrink we don't want to
1052 # trigger this heuristic, so whenever we shrink successfully
1053 # we give ourselves a bit of breathing room to make sure we
1054 # would find a shrink that took that long to find the next time.
1055 # The case where we're taking a long time but making steady
1056 # progress is handled by `finish_shrinking_deadline` in engine.py
1057 self.max_stall = max(
1058 self.max_stall, (self.calls - self.calls_at_last_shrink) * 2
1059 )
1060 self.calls_at_last_shrink = self.calls
1061 self.shrink_target = new_target
1062 self.__derived_values = {}
1064 def try_shrinking_nodes(self, nodes, n):
1065 """Attempts to replace each node in the nodes list with n. Returns
1066 True if it succeeded (which may include some additional modifications
1067 to shrink_target).
1069 In current usage it is expected that each of the nodes currently have
1070 the same value and choice_type, although this is not essential. Note that
1071 n must be < the node at min(nodes) or this is not a valid shrink.
1073 This method will attempt to do some small amount of work to delete data
1074 that occurs after the end of the nodes. This is useful for cases where
1075 there is some size dependency on the value of a node.
1076 """
1077 # If the length of the shrink target has changed from under us such that
1078 # the indices are out of bounds, give up on the replacement.
1079 # TODO_BETTER_SHRINK: we probably want to narrow down the root cause here at some point.
1080 if any(node.index >= len(self.nodes) for node in nodes):
1081 return # pragma: no cover
1083 initial_attempt = replace_all(
1084 self.nodes,
1085 [(node.index, node.index + 1, [node.copy(with_value=n)]) for node in nodes],
1086 )
1088 attempt = self.cached_test_function(initial_attempt)[1]
1090 if attempt is None:
1091 return False
1093 if attempt is self.shrink_target:
1094 # if the initial shrink was a success, try lowering offsets.
1095 self.lower_common_node_offset()
1096 return True
1098 # If this produced something completely invalid we ditch it
1099 # here rather than trying to persevere.
1100 if attempt.status is Status.OVERRUN:
1101 return False
1103 if attempt.status is Status.INVALID:
1104 return False
1106 if attempt.misaligned_at is not None:
1107 # we're invalid due to a misalignment in the tree. We'll try to fix
1108 # a very specific type of misalignment here: where we have a node of
1109 # {"size": n} and tried to draw the same node, but with {"size": m < n}.
1110 # This can occur with eg
1111 #
1112 # n = data.draw_integer()
1113 # s = data.draw_string(min_size=n)
1114 #
1115 # where we try lowering n, resulting in the test_function drawing a lower
1116 # min_size than our attempt had for the draw_string node.
1117 #
1118 # We'll now try realigning this tree by:
1119 # * replacing the constraints in our attempt with what test_function tried
1120 # to draw in practice
1121 # * truncating the value of that node to match min_size
1122 #
1123 # This helps in the specific case of drawing a value and then drawing
1124 # a collection of that size...and not much else. In practice this
1125 # helps because this antipattern is fairly common.
1127 # TODO we'll probably want to apply the same trick as in the valid
1128 # case of this function of preserving from the right instead of
1129 # preserving from the left. see test_can_shrink_variable_string_draws.
1131 (index, attempt_choice_type, attempt_constraints, _attempt_forced) = (
1132 attempt.misaligned_at
1133 )
1134 node = self.nodes[index]
1135 if node.type != attempt_choice_type:
1136 return False # pragma: no cover
1137 if node.was_forced:
1138 return False # pragma: no cover
1140 if node.type in {"string", "bytes"}:
1141 # if the size *increased*, we would have to guess what to pad with
1142 # in order to try fixing up this attempt. Just give up.
1143 if node.constraints["min_size"] <= attempt_constraints["min_size"]:
1144 # attempts which increase min_size tend to overrun rather than
1145 # be misaligned, making a covering case difficult.
1146 return False # pragma: no cover
1147 # the size decreased in our attempt. Try again, but truncate the value
1148 # to that size by removing any elements past min_size.
1149 return self.consider_new_nodes(
1150 initial_attempt[: node.index]
1151 + [
1152 initial_attempt[node.index].copy(
1153 with_constraints=attempt_constraints,
1154 with_value=initial_attempt[node.index].value[
1155 : attempt_constraints["min_size"]
1156 ],
1157 )
1158 ]
1159 + initial_attempt[node.index :]
1160 )
1162 lost_nodes = len(self.nodes) - len(attempt.nodes)
1163 if lost_nodes <= 0:
1164 return False
1166 start = nodes[0].index
1167 end = nodes[-1].index + 1
1168 # We now look for contiguous regions to delete that might help fix up
1169 # this failed shrink. We only look for contiguous regions of the right
1170 # lengths because doing anything more than that starts to get very
1171 # expensive. See minimize_individual_choices for where we
1172 # try to be more aggressive.
1173 regions_to_delete = {(end, end + lost_nodes)}
1175 for ex in self.spans:
1176 if ex.start > start:
1177 continue
1178 if ex.end <= end:
1179 continue
1181 if ex.index >= len(attempt.spans):
1182 continue # pragma: no cover
1184 replacement = attempt.spans[ex.index]
1185 in_original = [c for c in ex.children if c.start >= end]
1186 in_replaced = [c for c in replacement.children if c.start >= end]
1188 if len(in_replaced) >= len(in_original) or not in_replaced:
1189 continue
1191 # We've found a span where some of the children went missing
1192 # as a result of this change, and just replacing it with the data
1193 # it would have had and removing the spillover didn't work. This
1194 # means that some of its children towards the right must be
1195 # important, so we try to arrange it so that it retains its
1196 # rightmost children instead of its leftmost.
1197 regions_to_delete.add(
1198 (in_original[0].start, in_original[-len(in_replaced)].start)
1199 )
1201 for u, v in sorted(regions_to_delete, key=lambda x: x[1] - x[0], reverse=True):
1202 try_with_deleted = initial_attempt[:u] + initial_attempt[v:]
1203 if self.consider_new_nodes(try_with_deleted):
1204 return True
1206 return False
1208 def remove_discarded(self):
1209 """Try removing all bytes marked as discarded.
1211 This is primarily to deal with data that has been ignored while
1212 doing rejection sampling - e.g. as a result of an integer range, or a
1213 filtered strategy.
1215 Such data will also be handled by the adaptive_example_deletion pass,
1216 but that pass is necessarily more conservative and will try deleting
1217 each interval individually. The common case is that all data drawn and
1218 rejected can just be thrown away immediately in one block, so this pass
1219 will be much faster than trying each one individually when it works.
1221 returns False if there is discarded data and removing it does not work,
1222 otherwise returns True.
1223 """
1224 while self.shrink_target.has_discards:
1225 discarded = []
1227 for ex in self.shrink_target.spans:
1228 if (
1229 ex.choice_count > 0
1230 and ex.discarded
1231 and (not discarded or ex.start >= discarded[-1][-1])
1232 ):
1233 discarded.append((ex.start, ex.end))
1235 # This can happen if we have discards but they are all of
1236 # zero length. This shouldn't happen very often so it's
1237 # faster to check for it here than at the point of example
1238 # generation.
1239 if not discarded:
1240 break
1242 attempt = list(self.nodes)
1243 for u, v in reversed(discarded):
1244 del attempt[u:v]
1246 if not self.consider_new_nodes(tuple(attempt)):
1247 return False
1248 return True
1250 @derived_value # type: ignore
1251 def duplicated_nodes(self):
1252 """Returns a list of nodes grouped (choice_type, value)."""
1253 duplicates = defaultdict(list)
1254 for node in self.nodes:
1255 duplicates[(node.type, choice_key(node.value))].append(node)
1256 return list(duplicates.values())
1258 def node_program(self, program: str) -> ShrinkPass:
1259 return ShrinkPass(
1260 lambda chooser: self._node_program(chooser, program),
1261 name=f"node_program_{program}",
1262 )
1264 def _node_program(self, chooser, program):
1265 n = len(program)
1266 # Adaptively attempt to run the node program at the current
1267 # index. If this successfully applies the node program ``k`` times
1268 # then this runs in ``O(log(k))`` test function calls.
1269 i = chooser.choose(range(len(self.nodes) - n + 1))
1271 # First, run the node program at the chosen index. If this fails,
1272 # don't do any extra work, so that failure is as cheap as possible.
1273 if not self.run_node_program(i, program, original=self.shrink_target):
1274 return
1276 # Because we run in a random order we will often find ourselves in the middle
1277 # of a region where we could run the node program. We thus start by moving
1278 # left to the beginning of that region if possible in order to to start from
1279 # the beginning of that region.
1280 def offset_left(k):
1281 return i - k * n
1283 i = offset_left(
1284 find_integer(
1285 lambda k: self.run_node_program(
1286 offset_left(k), program, original=self.shrink_target
1287 )
1288 )
1289 )
1291 original = self.shrink_target
1292 # Now try to run the node program multiple times here.
1293 find_integer(
1294 lambda k: self.run_node_program(i, program, original=original, repeats=k)
1295 )
1297 def minimize_duplicated_choices(self, chooser):
1298 """Find choices that have been duplicated in multiple places and attempt
1299 to minimize all of the duplicates simultaneously.
1301 This lets us handle cases where two values can't be shrunk
1302 independently of each other but can easily be shrunk together.
1303 For example if we had something like:
1305 ls = data.draw(lists(integers()))
1306 y = data.draw(integers())
1307 assert y not in ls
1309 Suppose we drew y = 3 and after shrinking we have ls = [3]. If we were
1310 to replace both 3s with 0, this would be a valid shrink, but if we were
1311 to replace either 3 with 0 on its own the test would start passing.
1313 It is also useful for when that duplication is accidental and the value
1314 of the choices don't matter very much because it allows us to replace
1315 more values at once.
1316 """
1317 nodes = chooser.choose(self.duplicated_nodes)
1318 # we can't lower any nodes which are trivial. try proceeding with the
1319 # remaining nodes.
1320 nodes = [node for node in nodes if not node.trivial]
1321 if len(nodes) <= 1:
1322 return
1324 self.minimize_nodes(nodes)
1326 def redistribute_numeric_pairs(self, chooser):
1327 """If there is a sum of generated numbers that we need their sum
1328 to exceed some bound, lowering one of them requires raising the
1329 other. This pass enables that."""
1331 # look for a pair of nodes (node1, node2) which are both numeric
1332 # and aren't separated by too many other nodes. We'll decrease node1 and
1333 # increase node2 (note that the other way around doesn't make sense as
1334 # it's strictly worse in the ordering).
1335 def can_choose_node(node):
1336 # don't choose nan, inf, or floats above the threshold where f + 1 > f
1337 # (which is not necessarily true for floats above MAX_PRECISE_INTEGER).
1338 # The motivation for the last condition is to avoid trying weird
1339 # non-shrinks where we raise one node and think we lowered another
1340 # (but didn't).
1341 return node.type in {"integer", "float"} and not (
1342 node.type == "float"
1343 and (math.isnan(node.value) or abs(node.value) >= MAX_PRECISE_INTEGER)
1344 )
1346 node1 = chooser.choose(
1347 self.nodes,
1348 lambda node: can_choose_node(node) and not node.trivial,
1349 )
1350 node2 = chooser.choose(
1351 self.nodes,
1352 lambda node: can_choose_node(node)
1353 # Note that it's fine for node2 to be trivial, because we're going to
1354 # explicitly make it *not* trivial by adding to its value.
1355 and not node.was_forced
1356 # to avoid quadratic behavior, scan ahead only a small amount for
1357 # the related node.
1358 and node1.index < node.index <= node1.index + 4,
1359 )
1361 m: int | float = node1.value
1362 n: int | float = node2.value
1364 def boost(k: int) -> bool:
1365 # floats always shrink towards 0
1366 shrink_towards = (
1367 node1.constraints["shrink_towards"] if node1.type == "integer" else 0
1368 )
1369 if k > abs(m - shrink_towards):
1370 return False
1372 # We are trying to move node1 (m) closer to shrink_towards, and node2
1373 # (n) farther away from shrink_towards. If m is below shrink_towards,
1374 # we want to add to m and subtract from n, and vice versa if above
1375 # shrink_towards.
1376 if m < shrink_towards:
1377 k = -k
1379 try:
1380 v1 = m - k
1381 v2 = n + k
1382 except OverflowError: # pragma: no cover
1383 # if n or m is a float and k is over sys.float_info.max, coercing
1384 # k to a float will overflow.
1385 return False
1387 # if we've increased node2 to the point that we're past max precision,
1388 # give up - things have become too unstable.
1389 if node1.type == "float" and abs(v2) >= MAX_PRECISE_INTEGER:
1390 return False
1392 return self.consider_new_nodes(
1393 self.nodes[: node1.index]
1394 + (node1.copy(with_value=v1),)
1395 + self.nodes[node1.index + 1 : node2.index]
1396 + (node2.copy(with_value=v2),)
1397 + self.nodes[node2.index + 1 :]
1398 )
1400 find_integer(boost)
1402 def lower_integers_together(self, chooser):
1403 node1 = chooser.choose(
1404 self.nodes, lambda n: n.type == "integer" and not n.trivial
1405 )
1406 # Search up to 3 nodes ahead, to avoid quadratic time.
1407 node2 = self.nodes[
1408 chooser.choose(
1409 range(node1.index + 1, min(len(self.nodes), node1.index + 3 + 1)),
1410 lambda i: self.nodes[i].type == "integer"
1411 and not self.nodes[i].was_forced,
1412 )
1413 ]
1415 # one might expect us to require node2 to be nontrivial, and to minimize
1416 # the node which is closer to its shrink_towards, rather than node1
1417 # unconditionally. In reality, it's acceptable for us to transition node2
1418 # from trivial to nontrivial, because the shrink ordering is dominated by
1419 # the complexity of the earlier node1. What matters is minimizing node1.
1420 shrink_towards = node1.constraints["shrink_towards"]
1422 def consider(n):
1423 return self.consider_new_nodes(
1424 self.nodes[: node1.index]
1425 + (node1.copy(with_value=node1.value - n),)
1426 + self.nodes[node1.index + 1 : node2.index]
1427 + (node2.copy(with_value=node2.value - n),)
1428 + self.nodes[node2.index + 1 :]
1429 )
1431 find_integer(lambda n: consider(shrink_towards - n))
1432 find_integer(lambda n: consider(n - shrink_towards))
1434 def lower_duplicated_characters(self, chooser):
1435 """
1436 Select two string choices no more than 4 choices apart and simultaneously
1437 lower characters which appear in both strings. This helps cases where the
1438 same character must appear in two strings, but the actual value of the
1439 character is not relevant.
1441 This shrinking pass currently only tries lowering *all* instances of the
1442 duplicated character in both strings. So for instance, given two choices:
1444 "bbac"
1445 "abbb"
1447 we would try lowering all five of the b characters simultaneously. This
1448 may fail to shrink some cases where only certain character indices are
1449 correlated, for instance if only the b at index 1 could be lowered
1450 simultaneously and the rest did in fact actually have to be a `b`.
1452 It would be nice to try shrinking that case as well, but we would need good
1453 safeguards because it could get very expensive to try all combinations.
1454 I expect lowering all duplicates to handle most cases in the meantime.
1455 """
1456 node1 = chooser.choose(
1457 self.nodes, lambda n: n.type == "string" and not n.trivial
1458 )
1460 # limit search to up to 4 choices ahead, to avoid quadratic behavior
1461 node2 = self.nodes[
1462 chooser.choose(
1463 range(node1.index + 1, min(len(self.nodes), node1.index + 1 + 4)),
1464 lambda i: self.nodes[i].type == "string" and not self.nodes[i].trivial
1465 # select nodes which have at least one of the same character present
1466 and set(node1.value) & set(self.nodes[i].value),
1467 )
1468 ]
1470 duplicated_characters = set(node1.value) & set(node2.value)
1471 # deterministic ordering
1472 char = chooser.choose(sorted(duplicated_characters))
1473 intervals = node1.constraints["intervals"]
1475 def copy_node(node, n):
1476 # replace all duplicate characters in each string. This might miss
1477 # some shrinks compared to only replacing some, but trying all possible
1478 # combinations of indices could get expensive if done without some
1479 # thought.
1480 return node.copy(
1481 with_value=node.value.replace(char, intervals.char_in_shrink_order(n))
1482 )
1484 Integer.shrink(
1485 intervals.index_from_char_in_shrink_order(char),
1486 lambda n: self.consider_new_nodes(
1487 self.nodes[: node1.index]
1488 + (copy_node(node1, n),)
1489 + self.nodes[node1.index + 1 : node2.index]
1490 + (copy_node(node2, n),)
1491 + self.nodes[node2.index + 1 :]
1492 ),
1493 )
1495 def minimize_nodes(self, nodes):
1496 choice_type = nodes[0].type
1497 value = nodes[0].value
1498 # unlike choice_type and value, constraints are *not* guaranteed to be equal among all
1499 # passed nodes. We arbitrarily use the constraints of the first node. I think
1500 # this is unsound (= leads to us trying shrinks that could not have been
1501 # generated), but those get discarded at test-time, and this enables useful
1502 # slips where constraints are not equal but are close enough that doing the
1503 # same operation on both basically just works.
1504 constraints = nodes[0].constraints
1505 assert all(
1506 node.type == choice_type and choice_equal(node.value, value)
1507 for node in nodes
1508 )
1510 if choice_type == "integer":
1511 shrink_towards = constraints["shrink_towards"]
1512 # try shrinking from both sides towards shrink_towards.
1513 # we're starting from n = abs(shrink_towards - value). Because the
1514 # shrinker will not check its starting value, we need to try
1515 # shrinking to n first.
1516 self.try_shrinking_nodes(nodes, abs(shrink_towards - value))
1517 Integer.shrink(
1518 abs(shrink_towards - value),
1519 lambda n: self.try_shrinking_nodes(nodes, shrink_towards + n),
1520 )
1521 Integer.shrink(
1522 abs(shrink_towards - value),
1523 lambda n: self.try_shrinking_nodes(nodes, shrink_towards - n),
1524 )
1525 elif choice_type == "float":
1526 self.try_shrinking_nodes(nodes, abs(value))
1527 Float.shrink(
1528 abs(value),
1529 lambda val: self.try_shrinking_nodes(nodes, val),
1530 )
1531 Float.shrink(
1532 abs(value),
1533 lambda val: self.try_shrinking_nodes(nodes, -val),
1534 )
1535 elif choice_type == "boolean":
1536 # must be True, otherwise would be trivial and not selected.
1537 assert value is True
1538 # only one thing to try: false!
1539 self.try_shrinking_nodes(nodes, False)
1540 elif choice_type == "bytes":
1541 Bytes.shrink(
1542 value,
1543 lambda val: self.try_shrinking_nodes(nodes, val),
1544 min_size=constraints["min_size"],
1545 )
1546 elif choice_type == "string":
1547 String.shrink(
1548 value,
1549 lambda val: self.try_shrinking_nodes(nodes, val),
1550 intervals=constraints["intervals"],
1551 min_size=constraints["min_size"],
1552 )
1553 else:
1554 raise NotImplementedError
1556 def try_trivial_spans(self, chooser):
1557 i = chooser.choose(range(len(self.spans)))
1559 prev = self.shrink_target
1560 nodes = self.shrink_target.nodes
1561 ex = self.spans[i]
1562 prefix = nodes[: ex.start]
1563 replacement = tuple(
1564 [
1565 (
1566 node
1567 if node.was_forced
1568 else node.copy(
1569 with_value=choice_from_index(0, node.type, node.constraints)
1570 )
1571 )
1572 for node in nodes[ex.start : ex.end]
1573 ]
1574 )
1575 suffix = nodes[ex.end :]
1576 attempt = self.cached_test_function(prefix + replacement + suffix)[1]
1578 if self.shrink_target is not prev:
1579 return
1581 if isinstance(attempt, ConjectureResult):
1582 new_ex = attempt.spans[i]
1583 new_replacement = attempt.nodes[new_ex.start : new_ex.end]
1584 self.consider_new_nodes(prefix + new_replacement + suffix)
1586 def minimize_individual_choices(self, chooser):
1587 """Attempt to minimize each choice in sequence.
1589 This is the pass that ensures that e.g. each integer we draw is a
1590 minimum value. So it's the part that guarantees that if we e.g. do
1592 x = data.draw(integers())
1593 assert x < 10
1595 then in our shrunk example, x = 10 rather than say 97.
1597 If we are unsuccessful at minimizing a choice of interest we then
1598 check if that's because it's changing the size of the test case and,
1599 if so, we also make an attempt to delete parts of the test case to
1600 see if that fixes it.
1602 We handle most of the common cases in try_shrinking_nodes which is
1603 pretty good at clearing out large contiguous blocks of dead space,
1604 but it fails when there is data that has to stay in particular places
1605 in the list.
1606 """
1607 node = chooser.choose(self.nodes, lambda node: not node.trivial)
1608 initial_target = self.shrink_target
1610 self.minimize_nodes([node])
1611 if self.shrink_target is not initial_target:
1612 # the shrink target changed, so our shrink worked. Defer doing
1613 # anything more intelligent until this shrink fails.
1614 return
1616 # the shrink failed. One particularly common case where minimizing a
1617 # node can fail is the antipattern of drawing a size and then drawing a
1618 # collection of that size, or more generally when there is a size
1619 # dependency on some single node. We'll explicitly try and fix up this
1620 # common case here: if decreasing an integer node by one would reduce
1621 # the size of the generated input, we'll try deleting things after that
1622 # node and see if the resulting attempt works.
1624 if node.type != "integer":
1625 # Only try this fixup logic on integer draws. Almost all size
1626 # dependencies are on integer draws, and if it's not, it's doing
1627 # something convoluted enough that it is unlikely to shrink well anyway.
1628 # TODO: extent to floats? we probably currently fail on the following,
1629 # albeit convoluted example:
1630 # n = int(data.draw(st.floats()))
1631 # s = data.draw(st.lists(st.integers(), min_size=n, max_size=n))
1632 return
1634 lowered = (
1635 self.nodes[: node.index]
1636 + (node.copy(with_value=node.value - 1),)
1637 + self.nodes[node.index + 1 :]
1638 )
1639 attempt = self.cached_test_function(lowered)[1]
1640 if (
1641 attempt is None
1642 or attempt.status < Status.VALID
1643 or len(attempt.nodes) == len(self.nodes)
1644 or len(attempt.nodes) == node.index + 1
1645 ):
1646 # no point in trying our size-dependency-logic if our attempt at
1647 # lowering the node resulted in:
1648 # * an invalid conjecture data
1649 # * the same number of nodes as before
1650 # * no nodes beyond the lowered node (nothing to try to delete afterwards)
1651 return
1653 # If it were then the original shrink should have worked and we could
1654 # never have got here.
1655 assert attempt is not self.shrink_target
1657 @self.cached(node.index)
1658 def first_span_after_node():
1659 lo = 0
1660 hi = len(self.spans)
1661 while lo + 1 < hi:
1662 mid = (lo + hi) // 2
1663 ex = self.spans[mid]
1664 if ex.start >= node.index:
1665 hi = mid
1666 else:
1667 lo = mid
1668 return hi
1670 # we try deleting both entire spans, and single nodes.
1671 # If we wanted to get more aggressive, we could try deleting n
1672 # consecutive nodes (that don't cross a span boundary) for say
1673 # n <= 2 or n <= 3.
1674 if chooser.choose([True, False]):
1675 ex = self.spans[
1676 chooser.choose(
1677 range(first_span_after_node, len(self.spans)),
1678 lambda i: self.spans[i].choice_count > 0,
1679 )
1680 ]
1681 self.consider_new_nodes(lowered[: ex.start] + lowered[ex.end :])
1682 else:
1683 node = self.nodes[chooser.choose(range(node.index + 1, len(self.nodes)))]
1684 self.consider_new_nodes(lowered[: node.index] + lowered[node.index + 1 :])
1686 def reorder_spans(self, chooser):
1687 """This pass allows us to reorder the children of each span.
1689 For example, consider the following:
1691 .. code-block:: python
1693 import hypothesis.strategies as st
1694 from hypothesis import given
1697 @given(st.text(), st.text())
1698 def test_not_equal(x, y):
1699 assert x != y
1701 Without the ability to reorder x and y this could fail either with
1702 ``x=""``, ``y="0"``, or the other way around. With reordering it will
1703 reliably fail with ``x=""``, ``y="0"``.
1704 """
1705 ex = chooser.choose(self.spans)
1706 label = chooser.choose(ex.children).label
1708 spans = [c for c in ex.children if c.label == label]
1709 if len(spans) <= 1:
1710 return
1711 st = self.shrink_target
1712 endpoints = [(ex.start, ex.end) for ex in spans]
1714 Ordering.shrink(
1715 range(len(spans)),
1716 lambda indices: self.consider_new_nodes(
1717 replace_all(
1718 st.nodes,
1719 [
1720 (
1721 u,
1722 v,
1723 st.nodes[spans[i].start : spans[i].end],
1724 )
1725 for (u, v), i in zip(endpoints, indices, strict=True)
1726 ],
1727 )
1728 ),
1729 key=lambda i: sort_key(st.nodes[spans[i].start : spans[i].end]),
1730 )
1732 def run_node_program(self, i, program, original, repeats=1):
1733 """Node programs are a mini-DSL for node rewriting, defined as a sequence
1734 of commands that can be run at some index into the nodes
1736 Commands are:
1738 * "X", delete this node
1740 This method runs the node program in ``program`` at node index
1741 ``i`` on the ConjectureData ``original``. If ``repeats > 1`` then it
1742 will attempt to approximate the results of running it that many times.
1744 Returns True if this successfully changes the underlying shrink target,
1745 else False.
1746 """
1747 if i + len(program) > len(original.nodes) or i < 0:
1748 return False
1749 attempt = list(original.nodes)
1750 for _ in range(repeats):
1751 for k, command in reversed(list(enumerate(program))):
1752 j = i + k
1753 if j >= len(attempt):
1754 return False
1756 if command == "X":
1757 del attempt[j]
1758 else:
1759 raise NotImplementedError(f"Unrecognised command {command!r}")
1761 return self.consider_new_nodes(attempt)