Support for new Hypothesis shrinker

docs-src/changelog.md

@@ -2,7 +2,10 @@
 HypoFuzz uses [calendar-based versioning](https://calver.org/), with a
 `YY-MM-patch` format.
 
+## 24.09.1
 Fixed compatibility with Pytest 8.1 ([#35](https://github.com/Zac-HD/hypofuzz/issues/35)).
+Requires [Hypothesis 6.103](https://hypothesis.readthedocs.io/en/latest/changes.html#v6-103-0)
+or newer, for compatibility with the new and improved shrinker.
 
 ## 24.02.3
 Fixed compatibility with Pytest 8.0 ([#32](https://github.com/Zac-HD/hypofuzz/issues/32)).

setup.py

@@ -37,7 +37,7 @@ def local_file(name: str) -> str:
         "black >= 23.3.0",
         "coverage >= 5.2.1",
         "dash >= 2.0.0",
-        "hypothesis[cli] >= 6.93.2",
+        "hypothesis[cli] >= 6.103.0",
         "libcst >= 1.0.0",
         "pandas >= 1.0.0",
         "psutil >= 3.0.0",

src/hypofuzz/__init__.py

@@ -1,4 +1,4 @@
 """Adaptive fuzzing for property-based tests using Hypothesis."""
 
-__version__ = "24.2.3"
+__version__ = "24.9.1"
 __all__: list = []

src/hypofuzz/corpus.py

@@ -7,7 +7,6 @@
     Callable,
     Counter,
     Dict,
-    FrozenSet,
     Iterable,
     List,
     Optional,
@@ -17,7 +16,7 @@
     Union,
 )
 
-from hypothesis import __version__ as hypothesis_version
+from hypothesis import __version__ as hypothesis_version, settings
 from hypothesis.core import encode_failure
 from hypothesis.database import ExampleDatabase
 from hypothesis.internal.conjecture.data import (
@@ -26,6 +25,7 @@
     Overrun,
     Status,
 )
+from hypothesis.internal.conjecture.engine import ConjectureRunner
 from hypothesis.internal.conjecture.shrinker import Shrinker
 from sortedcontainers import SortedDict
 
@@ -55,36 +55,23 @@ def reproduction_decorator(buffer: bytes) -> str:
     return f"@reproduce_failure({hypothesis_version!r}, {encode_failure(buffer)!r})"
 
 
-class EngineStub:
-    """A knock-off ConjectureEngine, just large enough to run a shrinker."""
-
-    def __init__(
-        self,
-        test_fn: Callable[[bytes], ConjectureData],
-        random: Random,
-        passing_buffers: FrozenSet[bytes],
-    ) -> None:
-        self.cached_test_function = test_fn
-        self.random = random
-        self.call_count = 0
-        self.report_debug_info = False
-        self.__passing_buffers = passing_buffers
-
-    def debug(self, msg: str) -> None:
-        """Unimplemented stub."""
-
-    def explain_next_call_as(self, msg: str) -> None:
-        """Unimplemented stub."""
-
-    def clear_call_explanation(self) -> None:
-        """Unimplemented stub."""
-
-    def passing_buffers(self, prefix: bytes = b"") -> FrozenSet[bytes]:
-        """Return a collection of bytestrings which cause the test to pass.
-
-        Optionally restrict this by a certain prefix, which is useful for explain mode.
-        """
-        return frozenset(b for b in self.__passing_buffers if b.startswith(prefix))
+def get_shrinker(
+    pool: "Pool",
+    fn: Callable[[bytes], ConjectureData],
+    *,
+    initial: Union[ConjectureData, ConjectureResult],
+    predicate: Callable[..., bool],
+    random: Random,
+    explain: bool = False,
+) -> Shrinker:
+    s = settings(database=pool._database, deadline=None)
+    return Shrinker(
+        ConjectureRunner(fn, random=random, database_key=pool._key, settings=s),
+        initial=initial,
+        predicate=predicate,
+        allow_transition=None,
+        explain=explain,
+    )
 
 
 class Pool:
@@ -311,12 +298,12 @@ def distill(self, fn: Callable[[bytes], ConjectureData], random: Random) -> None
                 set(self.covering_buffers) - minimal_branches,
                 key=lambda a: sort_key(self.covering_buffers[a]),
             )
-            shrinker = Shrinker(
-                EngineStub(fn, random, passing_buffers=frozenset()),
-                self.results[self.covering_buffers[arc_to_shrink]],
+            shrinker = get_shrinker(
+                self,
+                fn,
+                initial=self.results[self.covering_buffers[arc_to_shrink]],
                 predicate=lambda d, a=arc_to_shrink: a in d.extra_information.branches,
-                allow_transition=None,
-                explain=False,
+                random=random,
             )
             shrinker.shrink()
             self.__shrunk_to_buffers.add(shrinker.shrink_target.buffer)

src/hypofuzz/hy.py

@@ -22,13 +22,12 @@
 from hypothesis.internal.conjecture.data import ConjectureData, Status
 from hypothesis.internal.conjecture.engine import BUFFER_SIZE
 from hypothesis.internal.conjecture.junkdrawer import stack_depth_of_caller
-from hypothesis.internal.conjecture.shrinker import Shrinker
 from hypothesis.internal.reflection import function_digest, get_signature
 from hypothesis.reporting import with_reporter
 from hypothesis.vendor.pretty import RepresentationPrinter
 from sortedcontainers import SortedKeyList
 
-from .corpus import BlackBoxMutator, CrossOverMutator, EngineStub, HowGenerated, Pool
+from .corpus import BlackBoxMutator, CrossOverMutator, HowGenerated, Pool, get_shrinker
 from .cov import CustomCollectionContext
 
 record_pytrace: Optional[Callable[..., Any]]
@@ -200,20 +199,24 @@ def run_one(self) -> None:
         # seen_count = len(self.pool.arc_counts)
 
         # Run the input
-        result = self._run_test_on(self.generate_prefix(), extend=BUFFER_SIZE)
+        result = self._run_test_on(
+            ConjectureData(
+                max_length=BUFFER_SIZE,
+                prefix=self.generate_prefix(),
+                random=self.random,
+            )
+        )
 
         if result.status is Status.INTERESTING:
             # Shrink to our minimal failing example, since we'll stop after this.
             self.shrinking = True
-            passing_buffers = frozenset(
-                b for b, r in self.pool.results.items() if r.status == Status.VALID
-            )
-            shrinker = Shrinker(
-                EngineStub(self._run_test_on, self.random, passing_buffers),
-                result,
+            shrinker = get_shrinker(
+                self.pool,
+                self._run_test_on,
+                initial=result,
                 predicate=lambda d: d.status is Status.INTERESTING,
-                allow_transition=None,
-                explain=False,  # TODO: enable explain mode
+                random=self.random,
+                explain=True,
             )
             self.stop_shrinking_at = self.elapsed_time + 300
             with contextlib.suppress(HitShrinkTimeoutError):
@@ -222,7 +225,7 @@ def run_one(self) -> None:
             if record_pytrace:
                 # Replay minimal example under our time-travelling debug tracer
                 self._run_test_on(
-                    shrinker.shrink_target.buffer,
+                    shrinker.shrink_target,
                     collector=record_pytrace(self.nodeid),
                 )
             UNDELIVERED_REPORTS.append(self._json_description)
@@ -240,10 +243,8 @@ def run_one(self) -> None:
 
     def _run_test_on(
         self,
-        buffer: bytes,
+        data: ConjectureData,
         *,
-        error_on_discard: bool = False,
-        extend: int = 0,
         source: HowGenerated = HowGenerated.shrinking,
         collector: Optional[contextlib.AbstractContextManager] = None,
     ) -> ConjectureData:
@@ -257,11 +258,6 @@ def _run_test_on(
         collector = collector or CustomCollectionContext()  # type: ignore
         assert collector is not None
         reports: List[str] = []
-        data = ConjectureData(
-            max_length=min(BUFFER_SIZE, len(buffer) + extend),
-            prefix=buffer,
-            random=self.random,
-        )
         try:
             with deterministic_PRNG(), BuildContext(
                 data, is_final=True