From 662beae6dec0b5b750a3c2730c934a5211e972bf Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Thu, 22 Oct 2026 17:41:54 -0400
Subject: [PATCH 01/12] gh-107862: Add example.xfail() and example.via() to the
 Hypothesis stubs

Hypothesis lets a test pin a known-failing input with
`@example(...).xfail(reason=..., raises=...)`, which is the natural way
to record a bug found by a property test without taking the input out of
the strategy or writing a separate expectedFailure test. The stub
`example` is a plain decorator function, so any test using `.xfail()` or
`.via()` failed to import when Hypothesis is not installed.

This turns the stub into a class with those two methods. An expected
failure is strict, as in Hypothesis: when the examples are replayed the
test must raise one of the `raises` exceptions for that input, and an
unexpected pass fails the test. `.via()` is a no-op, as it is upstream.
---
 .../support/_hypothesis_stubs/__init__.py     | 68 +++++++++++++++----
 ...-10-22-10-00-00.gh-issue-107862.vX7pLm.rst |  3 +
 2 files changed, 59 insertions(+), 12 deletions(-)
 create mode 100644 Misc/NEWS.d/next/Tests/2026-10-22-10-00-00.gh-issue-107862.vX7pLm.rst

diff --git a/Lib/test/support/_hypothesis_stubs/__init__.py b/Lib/test/support/_hypothesis_stubs/__init__.py
index e3cb54aad736d2..94b6f7eaf06a44 100644
--- a/Lib/test/support/_hypothesis_stubs/__init__.py
+++ b/Lib/test/support/_hypothesis_stubs/__init__.py
@@ -1,3 +1,4 @@
+from dataclasses import dataclass, replace
 from enum import Enum
 import functools
 import unittest
@@ -18,21 +19,52 @@
 from . import stateful, strategies
 
 
+@dataclass(frozen=True)
+class _Example:
+    args: tuple
+    kwargs: dict
+    # Set by example.xfail(): the exception types the example must raise.
+    raises: type[BaseException] | tuple[type[BaseException], ...] | None = None
+    reason: str = ""
+
+    def __repr__(self):
+        bits = [repr(a) for a in self.args]
+        bits += [f"{k}={v!r}" for k, v in self.kwargs.items()]
+        return f"@example({', '.join(bits)})"
+
+
+def _run_example(f, self, ex):
+    if ex.raises is None:
+        f(self, *ex.args, **ex.kwargs)
+        return
+    try:
+        f(self, *ex.args, **ex.kwargs)
+    except Exception as exc:
+        if not isinstance(exc, ex.raises):
+            raise
+        return
+    reason = f" because {ex.reason}" if ex.reason else ""
+    raise AssertionError(
+        f"Expected an exception from {ex!r}{reason}, "
+        "but no exception was raised"
+    )
+
+
 def given(*_args, **_kwargs):
     def decorator(f):
         if examples := getattr(f, "_examples", []):
 
             @functools.wraps(f)
             def test_function(self):
-                for example_args, example_kwargs in examples:
-                    if len(example_args) < 2:
-                        subtest_args = example_args
+                for ex in examples:
+                    if len(ex.args) < 2:
+                        subtest_args = ex.args
                     else:
                         # subTest takes up to one positional argument.
                         # When there are more, display them as a tuple
-                        subtest_args = [example_args]
-                    with self.subTest(*subtest_args, **example_kwargs):
-                        f(self, *example_args, **example_kwargs)
+                        subtest_args = [ex.args]
+                    with self.subTest(*subtest_args, **ex.kwargs):
+                        _run_example(f, self, ex)
 
         else:
             # If we have found no examples, we must skip the test. If @example
@@ -49,16 +81,18 @@ def test_function(self):
     return decorator
 
 
-def example(*args, **kwargs):
-    if bool(args) == bool(kwargs):
-        raise ValueError("Must specify exactly one of *args or **kwargs")
+class example:
+    def __init__(self, *args, **kwargs):
+        if bool(args) == bool(kwargs):
+            raise ValueError("Must specify exactly one of *args or **kwargs")
+        self._example = _Example(args, kwargs)
 
-    def decorator(f):
+    def __call__(self, f):
         base_func = getattr(f, "__wrapped__", f)
         if not hasattr(base_func, "_examples"):
             base_func._examples = []
 
-        base_func._examples.append((args, kwargs))
+        base_func._examples.append(self._example)
 
         if getattr(f, "_given", False):
             # If the given decorator is below all the example decorators,
@@ -68,7 +102,17 @@ def decorator(f):
 
         return f
 
-    return decorator
+    def xfail(self, condition=True, *, reason="", raises=BaseException):
+        # Like real Hypothesis, an expected failure is strict: the example
+        # must raise one of the given exceptions, or the test fails.
+        if condition:
+            self._example = replace(self._example, raises=raises,
+                                    reason=reason)
+        return self
+
+    def via(self, whence, /):
+        # A label for tooling; it does not change what runs.
+        return self
 
 
 def assume(condition):
diff --git a/Misc/NEWS.d/next/Tests/2026-10-22-10-00-00.gh-issue-107862.vX7pLm.rst b/Misc/NEWS.d/next/Tests/2026-10-22-10-00-00.gh-issue-107862.vX7pLm.rst
new file mode 100644
index 00000000000000..6c583e75c39c56
--- /dev/null
+++ b/Misc/NEWS.d/next/Tests/2026-10-22-10-00-00.gh-issue-107862.vX7pLm.rst
@@ -0,0 +1,3 @@
+Add ``example.xfail()`` and ``example.via()`` to the Hypothesis stubs used
+when Hypothesis is not installed, so that property tests can mark an explicit
+example as an expected failure in both modes.

From b2b2000552e9fabb2ea831eaaef9d8ce8960f015 Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Tue, 20 Oct 2026 15:57:26 -0400
Subject: [PATCH 02/12] gh-107862: Add Hypothesis tests for the compression
 modules

The incremental decompressors in bz2, lzma, zlib and compression.zstd,
the older zlib.decompressobj protocol, and the DecompressReader file
layer are CPython code sharing one contract, but the existing tests only
exercise them with fixed chunk sizes. This adds property tests that feed
random data through them in random pieces with random max_length values,
with the module-level one-shot functions as the oracle, plus a stateful
test per backend for the decompressor protocol and for the read-mode file
objects. The strategies avoid filters so the tests can also be used as
fuzz targets.

Writing these turned up three bugs, each pinned by an expectedFailure
test with the minimal reproducer as an example:

- bz2.BZ2Decompressor and zlib._ZlibDecompressor report needs_input=True
  when max_length cut the output and more is still buffered.
- lzma.LZMADecompressor raises LZMAError("Insufficient buffer space") on
  the second of two consecutive max_length=0 calls with no new input;
  the other backends return b''. The broad lzma tests leave out
  max_length=0 so that they keep checking everything else.
- zlib._ZlibDecompressor(zdict=...) raises SystemError on a zlib-wrapped
  stream, because the Z_NEED_DICT path returns NULL without setting an
  exception.
---
 Lib/test/test_compression_property.py | 994 ++++++++++++++++++++++++++
 1 file changed, 994 insertions(+)
 create mode 100644 Lib/test/test_compression_property.py

diff --git a/Lib/test/test_compression_property.py b/Lib/test/test_compression_property.py
new file mode 100644
index 00000000000000..1ece7329e3e2e7
--- /dev/null
+++ b/Lib/test/test_compression_property.py
@@ -0,0 +1,994 @@
+"""Property-based tests for the incremental compression APIs.
+
+The compression modules wrap external libraries, but the incremental
+``decompress(data, max_length)`` protocol (``eof``, ``needs_input``,
+``unused_data``), the ``zlib.decompressobj`` variant, and the file layer built
+on ``compression._common._streams.DecompressReader`` are implemented in
+CPython and share one contract across bz2, gzip, lzma, zlib and zstd.  The
+tests here exercise that contract with random data, random chunking and
+random ``max_length`` values, using the module-level one-shot functions as
+oracles.
+
+The strategies only use ``.map()``, never ``.filter()`` or ``assume()``, so
+every generated input is a valid test case.  That keeps the tests usable as
+fuzz targets through each test's ``hypothesis.fuzz_one_input`` attribute.
+"""
+import dataclasses
+import io
+import unittest
+from collections.abc import Callable
+
+from test import support
+from test.support.hypothesis_helper import hypothesis
+
+try:
+    import bz2
+except ImportError:
+    bz2 = None
+try:
+    import gzip
+except ImportError:
+    gzip = None
+try:
+    import lzma
+except ImportError:
+    lzma = None
+try:
+    import zlib
+except ImportError:
+    zlib = None
+try:
+    from compression import zstd
+except ImportError:
+    zstd = None
+
+st = hypothesis.strategies
+stateful = hypothesis.stateful
+
+
+def _repeat(pair):
+    chunk, count = pair
+    return chunk * count
+
+
+# Small inputs keep each example cheap; the repeat count makes some of the
+# data compressible so that not every stream is a stored block.
+data_strategy = st.tuples(st.binary(max_size=512),
+                          st.integers(1, 4)).map(_repeat)
+trailer_strategy = st.binary(max_size=16)
+chunk_sizes_strategy = st.lists(st.integers(0, 64), max_size=16)
+max_length_strategy = st.integers(-1, 64)
+max_lengths_strategy = st.lists(max_length_strategy, max_size=16)
+
+
+def _chunks(data, sizes):
+    """Split *data* at the given sizes; whatever is left is the last chunk."""
+    pos = 0
+    for size in sizes:
+        yield data[pos:pos + size]
+        pos += size
+    yield data[pos:]
+
+
+def _cycle(values, index, default):
+    return values[index % len(values)] if values else default
+
+
+@dataclasses.dataclass(frozen=True)
+class _Backend:
+    name: str
+    new_decompressor: Callable[[], object]
+    new_compressor: Callable[[], object]
+    compress: Callable[[bytes], bytes]
+    decompress: Callable[[bytes], bytes]
+    # Raised by the incremental decompressor on invalid input.
+    error: type[Exception]
+    # Everything the module-level decompress may raise on invalid input.
+    decompress_errors: tuple[type[Exception], ...]
+    # Opens a binary file object for reading, if the module has one.
+    open_file: Callable[[io.BytesIO], io.BufferedIOBase] | None = None
+
+
+def _bz2_backend():
+    return _Backend(
+        name="bz2",
+        new_decompressor=bz2.BZ2Decompressor,
+        new_compressor=lambda: bz2.BZ2Compressor(1),
+        compress=lambda data: bz2.compress(data, 1),
+        decompress=bz2.decompress,
+        error=OSError,
+        decompress_errors=(OSError, ValueError),
+        open_file=bz2.BZ2File,
+    )
+
+
+def _gzip_backend():
+    return _Backend(
+        name="gzip",
+        new_decompressor=lambda: zlib._ZlibDecompressor(31),
+        new_compressor=lambda: zlib.compressobj(1, wbits=31),
+        compress=lambda data: gzip.compress(data, 1),
+        decompress=gzip.decompress,
+        error=zlib.error,
+        decompress_errors=(gzip.BadGzipFile, EOFError, zlib.error),
+        open_file=lambda fp: gzip.GzipFile(fileobj=fp),
+    )
+
+
+def _lzma_backend():
+    return _Backend(
+        name="lzma",
+        new_decompressor=lzma.LZMADecompressor,
+        new_compressor=lambda: lzma.LZMACompressor(preset=0),
+        compress=lambda data: lzma.compress(data, preset=0),
+        decompress=lzma.decompress,
+        error=lzma.LZMAError,
+        decompress_errors=(lzma.LZMAError,),
+        open_file=lzma.LZMAFile,
+    )
+
+
+def _zlib_backend(wbits):
+    return _Backend(
+        name=f"zlib(wbits={wbits})",
+        new_decompressor=lambda: zlib._ZlibDecompressor(wbits),
+        new_compressor=lambda: zlib.compressobj(1, wbits=wbits),
+        compress=lambda data: zlib.compress(data, 1, wbits=wbits),
+        decompress=lambda data: zlib.decompress(data, wbits=wbits),
+        error=zlib.error,
+        decompress_errors=(zlib.error,),
+    )
+
+
+def _zstd_backend():
+    return _Backend(
+        name="zstd",
+        new_decompressor=zstd.ZstdDecompressor,
+        new_compressor=lambda: zstd.ZstdCompressor(1),
+        compress=lambda data: zstd.compress(data, 1),
+        decompress=zstd.decompress,
+        error=zstd.ZstdError,
+        decompress_errors=(zstd.ZstdError,),
+        open_file=zstd.ZstdFile,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Incremental decompressor protocol
+# ---------------------------------------------------------------------------
+
+class _DecompressorMachine(stateful.RuleBasedStateMachine):
+    """Feed one compressed stream through a decompressor in random pieces.
+
+    After every call the output must be a prefix of the original data and no
+    longer than ``max_length``; ``eof`` is set exactly when all of the data
+    is out, at which point ``unused_data`` holds whatever followed the
+    stream; and any call after ``eof`` raises ``EOFError``.
+
+    Subclasses set ``backend``.
+    """
+
+    backend: _Backend
+
+    @stateful.initialize(data=data_strategy, trailer=trailer_strategy)
+    def start(self, data, trailer):
+        self.data = data
+        self.trailer = trailer
+        self.stream = self.backend.compress(data) + trailer
+        self.pos = 0
+        self.out = b""
+        self.decomp = self.backend.new_decompressor()
+
+    def _feed(self, chunk, max_length):
+        if self.decomp.eof:
+            try:
+                self.decomp.decompress(chunk, max_length)
+            except EOFError:
+                return
+            raise AssertionError("decompress() after eof did not raise")
+        ret = self.decomp.decompress(chunk, max_length)
+        assert isinstance(ret, bytes)
+        if max_length >= 0:
+            assert len(ret) <= max_length, (len(ret), max_length)
+        self.out += ret
+        assert self.data.startswith(self.out)
+        if self.decomp.eof:
+            assert self.out == self.data
+            rest = self.stream[self.pos:]
+            assert self.decomp.unused_data + rest == self.trailer
+        else:
+            assert self.decomp.unused_data == b""
+
+    def _feed_next(self, n, max_length):
+        chunk = self.stream[self.pos:self.pos + n]
+        self.pos += len(chunk)
+        self._feed(chunk, max_length)
+
+    @stateful.rule(n=st.integers(0, 64), max_length=max_length_strategy)
+    def feed(self, n, max_length):
+        self._feed_next(n, max_length)
+
+    @stateful.rule()
+    def finish(self):
+        if self.decomp.eof:
+            self._feed(b"", -1)
+            return
+        self._feed_next(len(self.stream), -1)
+        # The input is exhausted; unlimited calls must drain any output that
+        # is still buffered and reach the end of the stream.
+        for _ in range(3):
+            if self.decomp.eof:
+                break
+            self._feed(b"", -1)
+        assert self.decomp.eof
+        assert self.out == self.data
+
+    @stateful.invariant()
+    def output_is_a_prefix(self):
+        assert self.data.startswith(self.out)
+        assert isinstance(self.decomp.needs_input, bool)
+        if self.decomp.eof:
+            assert self.out == self.data
+
+
+class _DecompressorTests:
+    """Deterministic companions to the state machine, replayable by the stub.
+
+    Subclasses set ``backend``.
+    """
+
+    backend: _Backend
+
+    def _walk(self, data, sizes, max_lengths, trailer=b""):
+        """Feed the stream in pieces, yielding the decompressor after each.
+
+        ``max_lengths`` is cycled over the chunks; an empty list means every
+        call is unlimited.  Once the pieces run out the stream is drained
+        and the final state checked.
+        """
+        stream = self.backend.compress(data) + trailer
+        decomp = self.backend.new_decompressor()
+        out = b""
+        fed = 0
+        for i, chunk in enumerate(_chunks(stream, sizes)):
+            max_length = _cycle(max_lengths, i, -1)
+            if decomp.eof:
+                with self.assertRaises(EOFError):
+                    decomp.decompress(chunk, max_length)
+                break
+            ret = decomp.decompress(chunk, max_length)
+            fed += len(chunk)
+            if max_length >= 0:
+                self.assertLessEqual(len(ret), max_length)
+            out += ret
+            self.assertTrue(data.startswith(out))
+            yield decomp
+        while not decomp.eof:
+            out += decomp.decompress(b"", -1)
+        self.assertEqual(out, data)
+        self.assertEqual(decomp.unused_data + stream[fed:], trailer)
+
+    def _check_chunked_decompress(self, data, trailer, sizes, max_lengths):
+        for _ in self._walk(data, sizes, max_lengths, trailer):
+            pass
+
+    @hypothesis.given(
+        data=data_strategy,
+        trailer=trailer_strategy,
+        sizes=chunk_sizes_strategy,
+        max_lengths=max_lengths_strategy,
+    )
+    @hypothesis.example(data=b"", trailer=b"", sizes=[], max_lengths=[])
+    @hypothesis.example(data=b"", trailer=b"tail", sizes=[1, 1, 1],
+                        max_lengths=[0])
+    @hypothesis.example(data=b"ab" * 40, trailer=b"", sizes=[5] * 12,
+                        max_lengths=[1, 0, 3])
+    @hypothesis.example(data=bytes(range(256)), trailer=b"x",
+                        sizes=[0, 7, 0, 300], max_lengths=[2, -1])
+    def test_chunked_decompress(self, data, trailer, sizes, max_lengths):
+        self._check_chunked_decompress(data, trailer, sizes, max_lengths)
+
+    def _check_needs_input_means_drained(self, data, sizes, max_lengths):
+        for decomp in self._walk(data, sizes, max_lengths):
+            if decomp.needs_input and not decomp.eof:
+                # A decompressor that asks for input has no output to give.
+                self.assertEqual(decomp.decompress(b"", -1), b"")
+
+    @hypothesis.given(
+        data=data_strategy,
+        sizes=chunk_sizes_strategy,
+        max_lengths=max_lengths_strategy,
+    )
+    @hypothesis.example(data=b"", sizes=[], max_lengths=[])
+    @hypothesis.example(data=b"abc" * 50, sizes=[10] * 10,
+                        max_lengths=[1, 5, 0])
+    def test_needs_input_means_drained(self, data, sizes, max_lengths):
+        self._check_needs_input_means_drained(data, sizes, max_lengths)
+
+    def _check_repeated_max_length_zero(self, data, count):
+        # Asking for no output, repeatedly, must not break the decompressor.
+        decomp = self.backend.new_decompressor()
+        for _ in range(count):
+            self.assertEqual(decomp.decompress(b"", 0), b"")
+            self.assertFalse(decomp.eof)
+        stream = self.backend.compress(data)
+        self.assertEqual(decomp.decompress(stream, -1), data)
+        self.assertTrue(decomp.eof)
+
+    @hypothesis.given(data=data_strategy, count=st.integers(1, 4))
+    @hypothesis.example(data=b"", count=1)
+    @hypothesis.example(data=b"", count=2)
+    @hypothesis.example(data=b"spam" * 10, count=3)
+    def test_repeated_max_length_zero(self, data, count):
+        self._check_repeated_max_length_zero(data, count)
+
+    @hypothesis.given(data=data_strategy, cut=st.integers(0, 10**6))
+    @hypothesis.example(data=b"", cut=0)
+    @hypothesis.example(data=b"", cut=10**6)
+    @hypothesis.example(data=b"hello" * 20, cut=3)
+    def test_truncated_stream_does_not_raise(self, data, cut):
+        stream = self.backend.compress(data)
+        cut %= len(stream) + 1
+        decomp = self.backend.new_decompressor()
+        out = decomp.decompress(stream[:cut], -1)
+        self.assertTrue(data.startswith(out))
+        self.assertEqual(decomp.eof, cut == len(stream))
+        if not decomp.eof:
+            self.assertTrue(decomp.needs_input)
+            self.assertEqual(decomp.unused_data, b"")
+
+    @hypothesis.given(data=data_strategy, sizes=chunk_sizes_strategy)
+    @hypothesis.example(data=b"", sizes=[])
+    @hypothesis.example(data=b"", sizes=[0, 0])
+    @hypothesis.example(data=b"xyzzy" * 30, sizes=[1, 0, 2, 100])
+    def test_chunked_compress_round_trip(self, data, sizes):
+        comp = self.backend.new_compressor()
+        stream = b"".join(comp.compress(chunk)
+                          for chunk in _chunks(data, sizes))
+        stream += comp.flush()
+        self.assertEqual(self.backend.decompress(stream), data)
+        decomp = self.backend.new_decompressor()
+        self.assertEqual(decomp.decompress(stream), data)
+        self.assertTrue(decomp.eof)
+        self.assertEqual(decomp.unused_data, b"")
+
+    # Invalid input: random bytes, or a valid stream with one byte replaced
+    # or with its end cut off.  The only acceptable outcomes are success or
+    # the backend's own error type.
+    @hypothesis.given(
+        garbage=st.binary(max_size=256),
+        data=data_strategy,
+        index=st.integers(0, 10**6),
+        byte=st.integers(0, 255),
+        sizes=chunk_sizes_strategy,
+        max_lengths=max_lengths_strategy,
+    )
+    @hypothesis.example(garbage=b"", data=b"", index=0, byte=0, sizes=[],
+                        max_lengths=[])
+    @hypothesis.example(garbage=b"\xff" * 20, data=b"", index=0, byte=0,
+                        sizes=[1], max_lengths=[0])
+    @hypothesis.example(garbage=b"", data=b"abc" * 30, index=10**6, byte=0,
+                        sizes=[3], max_lengths=[1, -1])
+    @hypothesis.example(garbage=b"", data=b"abc" * 30, index=5, byte=255,
+                        sizes=[], max_lengths=[])
+    def test_invalid_input(self, garbage, data, index, byte, sizes,
+                           max_lengths):
+        stream = self.backend.compress(data)
+        index %= len(stream) + 1
+        if index == len(stream):
+            corrupted = stream[:byte % (len(stream) + 1)]
+        else:
+            corrupted = stream[:index] + bytes([byte]) + stream[index + 1:]
+        for blob in (garbage, corrupted):
+            with self.subTest(blob=blob):
+                decomp = self.backend.new_decompressor()
+                try:
+                    for i, chunk in enumerate(_chunks(blob, sizes)):
+                        decomp.decompress(chunk, _cycle(max_lengths, i, -1))
+                except (self.backend.error, EOFError):
+                    pass
+                try:
+                    self.backend.decompress(blob)
+                except self.backend.decompress_errors:
+                    pass
+
+
+@support.requires_bz2()
+class BZ2DecompressorTests(_DecompressorTests, unittest.TestCase):
+    backend = _bz2_backend() if bz2 is not None else None
+
+    # bz2 reports needs_input=True as soon as its input buffer is empty, even
+    # when max_length cut the output short and more of it is pending.
+    @unittest.expectedFailure
+    @hypothesis.given(
+        data=data_strategy,
+        sizes=chunk_sizes_strategy,
+        max_lengths=max_lengths_strategy,
+    )
+    @hypothesis.example(data=b"\x00\x10 0@" * 2, sizes=[40], max_lengths=[0])
+    def test_needs_input_means_drained(self, data, sizes, max_lengths):
+        self._check_needs_input_means_drained(data, sizes, max_lengths)
+
+
+# liblzma reports an error when two consecutive calls can make no progress,
+# which lzma.LZMADecompressor lets through as LZMAError("Insufficient buffer
+# space") for two max_length=0 calls in a row with nothing buffered.  The
+# broad lzma tests avoid max_length=0 so that they keep checking everything
+# else; test_repeated_max_length_zero below pins the bug.
+lzma_max_length_strategy = st.integers(1, 64) | st.just(-1)
+lzma_max_lengths_strategy = st.lists(lzma_max_length_strategy, max_size=16)
+
+
+@support.requires_lzma()
+class LZMADecompressorTests(_DecompressorTests, unittest.TestCase):
+    backend = _lzma_backend() if lzma is not None else None
+
+    @hypothesis.given(
+        data=data_strategy,
+        trailer=trailer_strategy,
+        sizes=chunk_sizes_strategy,
+        max_lengths=lzma_max_lengths_strategy,
+    )
+    @hypothesis.example(data=b"", trailer=b"", sizes=[], max_lengths=[])
+    @hypothesis.example(data=b"", trailer=b"tail", sizes=[1, 1, 1],
+                        max_lengths=[1])
+    @hypothesis.example(data=b"ab" * 40, trailer=b"", sizes=[5] * 12,
+                        max_lengths=[1, 2, 3])
+    @hypothesis.example(data=bytes(range(256)), trailer=b"x",
+                        sizes=[0, 7, 0, 300], max_lengths=[2, -1])
+    def test_chunked_decompress(self, data, trailer, sizes, max_lengths):
+        self._check_chunked_decompress(data, trailer, sizes, max_lengths)
+
+    @hypothesis.given(
+        data=data_strategy,
+        sizes=chunk_sizes_strategy,
+        max_lengths=lzma_max_lengths_strategy,
+    )
+    @hypothesis.example(data=b"", sizes=[], max_lengths=[])
+    @hypothesis.example(data=b"abc" * 50, sizes=[10] * 10,
+                        max_lengths=[1, 5, 2])
+    def test_needs_input_means_drained(self, data, sizes, max_lengths):
+        self._check_needs_input_means_drained(data, sizes, max_lengths)
+
+    @unittest.expectedFailure
+    @hypothesis.given(data=data_strategy, count=st.integers(1, 4))
+    @hypothesis.example(data=b"", count=2)
+    def test_repeated_max_length_zero(self, data, count):
+        self._check_repeated_max_length_zero(data, count)
+
+
+class _ZlibDecompressorTests(_DecompressorTests):
+    # zlib._ZlibDecompressor has the same needs_input problem as bz2.  One
+    # example per wbits variant: zlib wrapper, raw deflate, gzip wrapper.
+    @unittest.expectedFailure
+    @hypothesis.given(
+        data=data_strategy,
+        sizes=chunk_sizes_strategy,
+        max_lengths=max_lengths_strategy,
+    )
+    @hypothesis.example(data=b"\x00", sizes=[4], max_lengths=[0])
+    @hypothesis.example(data=b"\x00" * 21, sizes=[], max_lengths=[2])
+    @hypothesis.example(
+        data=b"\x00\x00\x00\x01\x00\x00\x02\x00\x00\x03\x00\x00"
+             b"\x90\x00\x90\x90\x90\x91\x90" * 2,
+        sizes=[34],
+        max_lengths=[20],
+    )
+    def test_needs_input_means_drained(self, data, sizes, max_lengths):
+        self._check_needs_input_means_drained(data, sizes, max_lengths)
+
+
+@support.requires_zlib()
+class ZlibDecompressorTests(_ZlibDecompressorTests, unittest.TestCase):
+    backend = _zlib_backend(zlib.MAX_WBITS) if zlib is not None else None
+
+
+@support.requires_zlib()
+class RawDeflateDecompressorTests(_ZlibDecompressorTests, unittest.TestCase):
+    backend = _zlib_backend(-15) if zlib is not None else None
+
+
+@support.requires_gzip()
+class GzipDecompressorTests(_ZlibDecompressorTests, unittest.TestCase):
+    backend = _gzip_backend() if gzip is not None else None
+
+
+@support.requires_zstd()
+class ZstdDecompressorTests(_DecompressorTests, unittest.TestCase):
+    backend = _zstd_backend() if zstd is not None else None
+
+
+class BZ2DecompressorMachine(_DecompressorMachine):
+    backend = _bz2_backend() if bz2 is not None else None
+
+
+class LZMADecompressorMachine(_DecompressorMachine):
+    backend = _lzma_backend() if lzma is not None else None
+
+    # See lzma_max_length_strategy above.
+    @stateful.rule(n=st.integers(0, 64), max_length=lzma_max_length_strategy)
+    def feed(self, n, max_length):
+        self._feed_next(n, max_length)
+
+
+class ZlibDecompressorMachine(_DecompressorMachine):
+    backend = _zlib_backend(zlib.MAX_WBITS) if zlib is not None else None
+
+
+class RawDeflateDecompressorMachine(_DecompressorMachine):
+    backend = _zlib_backend(-15) if zlib is not None else None
+
+
+class GzipDecompressorMachine(_DecompressorMachine):
+    backend = _gzip_backend() if gzip is not None else None
+
+
+class ZstdDecompressorMachine(_DecompressorMachine):
+    backend = _zstd_backend() if zstd is not None else None
+
+
+BZ2DecompressorMachineTest = support.requires_bz2()(
+    BZ2DecompressorMachine.TestCase)
+LZMADecompressorMachineTest = support.requires_lzma()(
+    LZMADecompressorMachine.TestCase)
+ZlibDecompressorMachineTest = support.requires_zlib()(
+    ZlibDecompressorMachine.TestCase)
+RawDeflateDecompressorMachineTest = support.requires_zlib()(
+    RawDeflateDecompressorMachine.TestCase)
+GzipDecompressorMachineTest = support.requires_gzip()(
+    GzipDecompressorMachine.TestCase)
+ZstdDecompressorMachineTest = support.requires_zstd()(
+    ZstdDecompressorMachine.TestCase)
+
+
+# ---------------------------------------------------------------------------
+# Compressor flush modes
+# ---------------------------------------------------------------------------
+
+@support.requires_zlib()
+class ZlibFlushModeTests(unittest.TestCase):
+    @hypothesis.given(
+        data=data_strategy,
+        sizes=chunk_sizes_strategy,
+        modes=st.lists(st.sampled_from(["Z_NO_FLUSH", "Z_PARTIAL_FLUSH",
+                                        "Z_SYNC_FLUSH", "Z_FULL_FLUSH",
+                                        "Z_BLOCK"]), max_size=16),
+        level=st.integers(0, 9),
+        wbits=st.sampled_from([15, -15, 31]),
+    )
+    @hypothesis.example(data=b"", sizes=[], modes=[], level=0, wbits=15)
+    @hypothesis.example(data=b"", sizes=[0, 0], modes=["Z_SYNC_FLUSH"],
+                        level=9, wbits=-15)
+    @hypothesis.example(data=b"abc" * 100, sizes=[1, 2, 3],
+                        modes=["Z_FULL_FLUSH", "Z_BLOCK", "Z_NO_FLUSH",
+                               "Z_PARTIAL_FLUSH"],
+                        level=6, wbits=31)
+    def test_flush_modes(self, data, sizes, modes, level, wbits):
+        # Any sequence of intermediate flushes still produces one valid
+        # stream that decompresses to the input.
+        comp = zlib.compressobj(level, wbits=wbits)
+        stream = b""
+        for i, chunk in enumerate(_chunks(data, sizes)):
+            stream += comp.compress(chunk)
+            if modes:
+                stream += comp.flush(getattr(zlib, _cycle(modes, i, None)))
+        stream += comp.flush()
+        self.assertEqual(zlib.decompress(stream, wbits=wbits), data)
+        decomp = zlib._ZlibDecompressor(wbits)
+        self.assertEqual(decomp.decompress(stream), data)
+        self.assertTrue(decomp.eof)
+        self.assertEqual(decomp.unused_data, b"")
+
+
+@support.requires_zstd()
+class ZstdCompressModeTests(unittest.TestCase):
+    @hypothesis.given(
+        data=data_strategy,
+        sizes=chunk_sizes_strategy,
+        modes=st.lists(st.sampled_from(["CONTINUE", "FLUSH_BLOCK",
+                                        "FLUSH_FRAME"]), max_size=16),
+    )
+    @hypothesis.example(data=b"", sizes=[], modes=[])
+    @hypothesis.example(data=b"", sizes=[0, 0], modes=["FLUSH_FRAME"])
+    @hypothesis.example(data=b"abc" * 100, sizes=[1, 2, 3, 50],
+                        modes=["FLUSH_BLOCK", "CONTINUE", "FLUSH_FRAME",
+                               "FLUSH_BLOCK"])
+    def test_compress_modes(self, data, sizes, modes):
+        # FLUSH_FRAME ends a frame and the next call starts a new one, so the
+        # output is a sequence of frames.  The module-level decompress() must
+        # handle all of them and the incremental decompressor must stop at
+        # each frame boundary, leaving the rest in unused_data.
+        comp = zstd.ZstdCompressor(1)
+        self.assertEqual(comp.last_mode, zstd.ZstdCompressor.FLUSH_FRAME)
+        stream = b""
+        for i, chunk in enumerate(_chunks(data, sizes)):
+            mode = getattr(zstd.ZstdCompressor,
+                           _cycle(modes, i, "CONTINUE"))
+            stream += comp.compress(chunk, mode)
+            self.assertEqual(comp.last_mode, mode)
+        stream += comp.flush()
+        self.assertEqual(comp.last_mode, zstd.ZstdCompressor.FLUSH_FRAME)
+        self.assertEqual(zstd.decompress(stream), data)
+
+        out = b""
+        rest = stream
+        while rest:
+            decomp = zstd.ZstdDecompressor()
+            out += decomp.decompress(rest)
+            self.assertTrue(decomp.eof)
+            self.assertEqual(zstd.get_frame_size(rest),
+                             len(rest) - len(decomp.unused_data))
+            rest = decomp.unused_data
+        self.assertEqual(out, data)
+
+
+# ---------------------------------------------------------------------------
+# zlib.decompressobj: the older protocol with unconsumed_tail
+# ---------------------------------------------------------------------------
+
+@support.requires_zlib()
+class DecompressObjTests(unittest.TestCase):
+    @hypothesis.given(
+        data=data_strategy,
+        trailer=trailer_strategy,
+        sizes=chunk_sizes_strategy,
+        max_lengths=st.lists(st.integers(0, 64), max_size=16),
+        wbits=st.sampled_from([15, -15, 31]),
+    )
+    @hypothesis.example(data=b"", trailer=b"", sizes=[], max_lengths=[],
+                        wbits=15)
+    @hypothesis.example(data=b"", trailer=b"tail", sizes=[1, 1],
+                        max_lengths=[0], wbits=-15)
+    @hypothesis.example(data=b"ab" * 40, trailer=b"x", sizes=[5] * 12,
+                        max_lengths=[1, 0, 3], wbits=31)
+    def test_max_length_and_unconsumed_tail(self, data, trailer, sizes,
+                                            max_lengths, wbits):
+        # Unlike the newer decompressors, decompressobj() keeps no input
+        # buffer: whatever max_length prevented it from consuming comes back
+        # in unconsumed_tail and must be passed in again.  Data after the
+        # end of the stream accumulates in unused_data instead.
+        stream = zlib.compress(data, 1, wbits=wbits) + trailer
+        decomp = zlib.decompressobj(wbits)
+        out = b""
+        fed = 0
+        for i, chunk in enumerate(_chunks(stream, sizes)):
+            max_length = _cycle(max_lengths, i, 0)
+            ret = decomp.decompress(decomp.unconsumed_tail + chunk,
+                                    max_length)
+            fed += len(chunk)
+            if max_length:
+                self.assertLessEqual(len(ret), max_length)
+            out += ret
+            self.assertTrue(data.startswith(out))
+            if decomp.eof:
+                self.assertEqual(out, data)
+                self.assertEqual(decomp.unconsumed_tail, b"")
+                self.assertEqual(decomp.unused_data + stream[fed:], trailer)
+            else:
+                self.assertEqual(decomp.unused_data, b"")
+        while not decomp.eof:
+            out += decomp.decompress(decomp.unconsumed_tail)
+        self.assertEqual(out, data)
+        self.assertEqual(decomp.unused_data + stream[fed:], trailer)
+
+    @hypothesis.given(data=data_strategy, cut=st.integers(0, 10**6))
+    @hypothesis.example(data=b"", cut=0)
+    @hypothesis.example(data=b"", cut=10**6)
+    @hypothesis.example(data=b"hello" * 20, cut=7)
+    def test_flush(self, data, cut):
+        stream = zlib.compress(data, 1)
+        cut %= len(stream) + 1
+        decomp = zlib.decompressobj()
+        out = decomp.decompress(stream[:cut])
+        self.assertTrue(data.startswith(out))
+        self.assertEqual(decomp.eof, cut == len(stream))
+        # flush() on a truncated stream returns what it can without raising;
+        # on a complete stream it is a no-op and may be repeated.
+        out += decomp.flush()
+        self.assertTrue(data.startswith(out))
+        if cut == len(stream):
+            self.assertEqual(out, data)
+            self.assertTrue(decomp.eof)
+            self.assertEqual(decomp.flush(), b"")
+
+    @hypothesis.given(
+        data=data_strategy,
+        zdict=st.binary(min_size=1, max_size=256),
+        level=st.integers(0, 9),
+    )
+    @hypothesis.example(data=b"", zdict=b"\x00", level=0)
+    @hypothesis.example(data=b"hello world" * 10, zdict=b"hello world",
+                        level=9)
+    def test_zdict_round_trip(self, data, zdict, level):
+        for wbits in (15, -15):
+            with self.subTest(wbits=wbits):
+                comp = zlib.compressobj(level, wbits=wbits, zdict=zdict)
+                stream = comp.compress(data) + comp.flush()
+                decomp = zlib.decompressobj(wbits, zdict=zdict)
+                self.assertEqual(decomp.decompress(stream), data)
+                self.assertTrue(decomp.eof)
+
+
+@support.requires_zlib()
+class ZlibDecompressorZdictTests(unittest.TestCase):
+    def _check_round_trip(self, data, zdict, level, wbits):
+        comp = zlib.compressobj(level, wbits=wbits, zdict=zdict)
+        stream = comp.compress(data) + comp.flush()
+        decomp = zlib._ZlibDecompressor(wbits, zdict=zdict)
+        self.assertEqual(decomp.decompress(stream), data)
+        self.assertTrue(decomp.eof)
+
+    @hypothesis.given(
+        data=data_strategy,
+        zdict=st.binary(min_size=1, max_size=256),
+        level=st.integers(0, 9),
+    )
+    @hypothesis.example(data=b"", zdict=b"\x00", level=0)
+    @hypothesis.example(data=b"hello world" * 10, zdict=b"hello world",
+                        level=9)
+    def test_raw_deflate(self, data, zdict, level):
+        self._check_round_trip(data, zdict, level, -15)
+
+    # _ZlibDecompressor only applies zdict up front for raw deflate; for a
+    # zlib-wrapped stream it hits Z_NEED_DICT and returns NULL without
+    # setting an exception, which surfaces as SystemError.
+    @unittest.expectedFailure
+    @hypothesis.given(
+        data=data_strategy,
+        zdict=st.binary(min_size=1, max_size=256),
+        level=st.integers(0, 9),
+    )
+    @hypothesis.example(data=b"", zdict=b"\x00", level=0)
+    @hypothesis.example(data=b"hello world" * 10, zdict=b"hello world",
+                        level=9)
+    def test_zlib_wrapper(self, data, zdict, level):
+        self._check_round_trip(data, zdict, level, 15)
+
+
+# ---------------------------------------------------------------------------
+# File layer: BZ2File, GzipFile, LZMAFile and ZstdFile in read mode
+# ---------------------------------------------------------------------------
+
+members_strategy = st.lists(data_strategy, min_size=1, max_size=3)
+read_size_strategy = st.integers(0, 300) | st.just(-1)
+
+
+class _StreamReaderMachine(stateful.RuleBasedStateMachine):
+    """Compare a compressed file object against io.BytesIO on the same data.
+
+    The compressed input is several concatenated streams (members, frames),
+    which the file layer must present as one contiguous file.
+
+    Subclasses set ``backend``.
+    """
+
+    backend: _Backend
+
+    @stateful.initialize(members=members_strategy)
+    def start(self, members):
+        self.plain = b"".join(members)
+        stream = b"".join(self.backend.compress(member) for member in members)
+        self.file = self.backend.open_file(io.BytesIO(stream))
+        self.oracle = io.BytesIO(self.plain)
+
+    def teardown(self):
+        self.file.close()
+
+    def _remaining(self):
+        return self.plain[self.oracle.tell():]
+
+    @stateful.rule(size=read_size_strategy)
+    def read(self, size):
+        assert self.file.read(size) == self.oracle.read(size)
+
+    @stateful.rule(size=read_size_strategy)
+    def read1(self, size):
+        ret = self.file.read1(size)
+        remaining = self._remaining()
+        assert remaining.startswith(ret)
+        if size != 0 and remaining:
+            assert ret, "read1() returned nothing before EOF"
+        if size >= 0:
+            assert len(ret) <= size
+        self.oracle.seek(len(ret), io.SEEK_CUR)
+
+    @stateful.rule(size=st.integers(0, 300))
+    def peek(self, size):
+        pos = self.file.tell()
+        ret = self.file.peek(size)
+        remaining = self._remaining()
+        assert remaining.startswith(ret)
+        if remaining:
+            assert ret, "peek() returned nothing before EOF"
+        assert self.file.tell() == pos
+
+    @stateful.rule(size=st.integers(0, 300))
+    def readinto(self, size):
+        buf = bytearray(size)
+        n = self.file.readinto(buf)
+        expected = self.oracle.read(size)
+        assert n == len(expected)
+        assert buf[:n] == expected
+
+    @stateful.rule()
+    def readline(self):
+        assert self.file.readline() == self.oracle.readline()
+
+    @stateful.rule(offset=st.integers(-10**6, 10**6),
+                   whence=st.sampled_from([io.SEEK_SET, io.SEEK_CUR,
+                                           io.SEEK_END]))
+    def seek(self, offset, whence):
+        # DecompressReader clamps to the file, unlike BytesIO.
+        if whence == io.SEEK_SET:
+            target = offset
+        elif whence == io.SEEK_CUR:
+            target = self.oracle.tell() + offset
+        else:
+            target = len(self.plain) + offset
+        target = min(max(target, 0), len(self.plain))
+        assert self.file.seek(offset, whence) == target
+        self.oracle.seek(target)
+
+    @stateful.invariant()
+    def positions_agree(self):
+        assert self.file.tell() == self.oracle.tell()
+
+
+class _StreamReaderTests:
+    """Deterministic companion to the state machine.
+
+    Subclasses set ``backend``.
+    """
+
+    backend: _Backend
+
+    @hypothesis.given(members=members_strategy,
+                      sizes=st.lists(read_size_strategy, max_size=16))
+    @hypothesis.example(members=[b""], sizes=[])
+    @hypothesis.example(members=[b"", b"a", b""], sizes=[0, 1, -1, 1])
+    @hypothesis.example(members=[b"abc\n" * 30, b"def" * 20],
+                        sizes=[5, 0, 300, 7])
+    def test_read_sizes(self, members, sizes):
+        plain = b"".join(members)
+        stream = b"".join(self.backend.compress(member) for member in members)
+        with self.backend.open_file(io.BytesIO(stream)) as f:
+            out = b""
+            for size in sizes:
+                ret = f.read(size)
+                expected = plain[len(out):]
+                if size >= 0:
+                    expected = expected[:size]
+                self.assertEqual(ret, expected)
+                out += ret
+                self.assertEqual(f.tell(), len(out))
+            self.assertEqual(out + f.read(), plain)
+            self.assertEqual(f.seek(0), 0)
+            self.assertEqual(f.read(), plain)
+
+
+@support.requires_bz2()
+class BZ2StreamReaderTests(_StreamReaderTests, unittest.TestCase):
+    backend = _bz2_backend() if bz2 is not None else None
+
+
+@support.requires_gzip()
+class GzipStreamReaderTests(_StreamReaderTests, unittest.TestCase):
+    backend = _gzip_backend() if gzip is not None else None
+
+
+@support.requires_lzma()
+class LZMAStreamReaderTests(_StreamReaderTests, unittest.TestCase):
+    backend = _lzma_backend() if lzma is not None else None
+
+
+@support.requires_zstd()
+class ZstdStreamReaderTests(_StreamReaderTests, unittest.TestCase):
+    backend = _zstd_backend() if zstd is not None else None
+
+
+class BZ2StreamReaderMachine(_StreamReaderMachine):
+    backend = _bz2_backend() if bz2 is not None else None
+
+
+class GzipStreamReaderMachine(_StreamReaderMachine):
+    backend = _gzip_backend() if gzip is not None else None
+
+
+class LZMAStreamReaderMachine(_StreamReaderMachine):
+    backend = _lzma_backend() if lzma is not None else None
+
+
+class ZstdStreamReaderMachine(_StreamReaderMachine):
+    backend = _zstd_backend() if zstd is not None else None
+
+
+BZ2StreamReaderMachineTest = support.requires_bz2()(
+    BZ2StreamReaderMachine.TestCase)
+GzipStreamReaderMachineTest = support.requires_gzip()(
+    GzipStreamReaderMachine.TestCase)
+LZMAStreamReaderMachineTest = support.requires_lzma()(
+    LZMAStreamReaderMachine.TestCase)
+ZstdStreamReaderMachineTest = support.requires_zstd()(
+    ZstdStreamReaderMachine.TestCase)
+
+
+# ---------------------------------------------------------------------------
+# zstd: frames, dictionaries and pledged sizes
+# ---------------------------------------------------------------------------
+
+@support.requires_zstd()
+class ZstdFrameTests(unittest.TestCase):
+    @hypothesis.given(data=data_strategy, level=st.integers(1, 5),
+                      trailer=trailer_strategy)
+    @hypothesis.example(data=b"", level=1, trailer=b"")
+    @hypothesis.example(data=b"x" * 100, level=3, trailer=b"junk")
+    def test_frame_size_and_info(self, data, level, trailer):
+        frame = zstd.compress(data, level)
+        self.assertEqual(zstd.get_frame_size(frame + trailer), len(frame))
+        info = zstd.get_frame_info(frame)
+        self.assertEqual(info.decompressed_size, len(data))
+        self.assertEqual(info.dictionary_id, 0)
+
+    @hypothesis.given(members=members_strategy)
+    @hypothesis.example(members=[b""])
+    @hypothesis.example(members=[b"", b"abc" * 20, b""])
+    def test_multiple_frames(self, members):
+        stream = b"".join(zstd.compress(member, 1) for member in members)
+        self.assertEqual(zstd.decompress(stream), b"".join(members))
+        rest = stream
+        for member in members:
+            decomp = zstd.ZstdDecompressor()
+            self.assertEqual(decomp.decompress(rest), member)
+            self.assertTrue(decomp.eof)
+            rest = decomp.unused_data
+        self.assertEqual(rest, b"")
+
+    @hypothesis.given(
+        data=data_strategy,
+        dict_content=st.binary(min_size=8, max_size=256),
+        level=st.integers(1, 5),
+    )
+    @hypothesis.example(data=b"", dict_content=b"\x00" * 8, level=1)
+    @hypothesis.example(data=b"hello world" * 10, dict_content=b"hello world!",
+                        level=3)
+    def test_raw_dict_round_trip(self, data, dict_content, level):
+        # A raw content dictionary has no ID and can be used either as a
+        # dictionary or as a prefix; the frame is the same either way and
+        # decompresses with any of the loading modes.
+        zd = zstd.ZstdDict(dict_content, is_raw=True)
+        self.assertEqual(zd.dict_id, 0)
+        self.assertEqual(zd.dict_content, dict_content)
+        modes = [zd, zd.as_digested_dict, zd.as_undigested_dict, zd.as_prefix]
+        for comp_mode in modes:
+            frame = zstd.compress(data, level, zstd_dict=comp_mode)
+            self.assertEqual(zstd.get_frame_info(frame).dictionary_id, 0)
+            for decomp_mode in modes:
+                self.assertEqual(
+                    zstd.decompress(frame, zstd_dict=decomp_mode), data)
+
+    @hypothesis.given(data=data_strategy, extra=st.integers(1, 100))
+    @hypothesis.example(data=b"", extra=1)
+    @hypothesis.example(data=b"abc" * 30, extra=7)
+    def test_pledged_input_size(self, data, extra):
+        comp = zstd.ZstdCompressor(1)
+        comp.set_pledged_input_size(len(data))
+        frame = comp.compress(data) + comp.flush()
+        self.assertEqual(zstd.get_frame_info(frame).decompressed_size,
+                         len(data))
+        self.assertEqual(zstd.decompress(frame), data)
+
+        # A wrong pledge is detected when the frame is finished, and the
+        # compressor is reset so it can be reused.  (A frame compressed in a
+        # single FLUSH_FRAME call is exempt: libzstd then knows the real
+        # size and records that instead.)
+        comp.set_pledged_input_size(len(data) + extra)
+        with self.assertRaises(zstd.ZstdError):
+            comp.compress(data)
+            comp.flush()
+        self.assertEqual(comp.last_mode, zstd.ZstdCompressor.FLUSH_FRAME)
+        frame = comp.compress(data, zstd.ZstdCompressor.FLUSH_FRAME)
+        self.assertEqual(zstd.decompress(frame), data)
+
+
+if __name__ == "__main__":
+    unittest.main()

From a05b584513739e9c1c8c29e3f915345df23e6a38 Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Tue, 20 Oct 2026 15:57:26 -0400
Subject: [PATCH 03/12] gh-107862: Use Hypothesis for the zlib checksum
 combination tests

These ran 1000 iterations over random 32 and 64 byte inputs with
unseeded random initial values, so a failure could not be reproduced.
Hypothesis gives the same coverage with shrinking and a persistent
example database.

The Adler-32 initial values are now restricted to valid checksums (both
16-bit sums below 65521). adler32_combine() is only defined on those,
and the old test only avoided the invalid region by chance, since
random.randint(1, 0x80000000) rarely produced a low half of 65521 or
more.
---
 Lib/test/test_zlib.py | 100 +++++++++++++++++++++++++++---------------
 1 file changed, 64 insertions(+), 36 deletions(-)

diff --git a/Lib/test/test_zlib.py b/Lib/test/test_zlib.py
index dcb4eac1b77952..a38df17e2961aa 100644
--- a/Lib/test/test_zlib.py
+++ b/Lib/test/test_zlib.py
@@ -1,6 +1,7 @@
 import unittest
 from test import support
 from test.support import import_helper
+from test.support.hypothesis_helper import hypothesis
 import binascii
 import copy
 import pickle
@@ -182,20 +183,18 @@ def test_same_as_binascii_crc32(self):
 class ChecksumCombineMixin:
     """Mixin class for testing checksum combination."""
 
-    N = 1000
     default_iv: int
+    # Initial values that are valid checksums for the algorithm.
+    iv_strategy: object
 
     def parse_iv(self, iv):
         """Parse an IV value.
 
         - The default IV is returned if *iv* is None.
-        - A random IV is returned if *iv* is -1.
         - Otherwise, *iv* is returned as is.
         """
         if iv is None:
             return self.default_iv
-        if iv == -1:
-            return random.randint(1, 0x80000000)
         return iv
 
     def checksum(self, data, init=None):
@@ -220,18 +219,24 @@ def get_random_data(self, data_len, *, iv=None):
         checksum = self.checksum(data, init)
         return data, init, checksum
 
-    def test_combine_empty(self):
-        for _ in range(self.N):
-            a, iv, checksum = self.get_random_data(32, iv=-1)
-            res = self.combine(iv, self.checksum(a), len(a))
-            self.assertEqual(res, checksum)
-
-    def test_combine_no_iv(self):
-        for _ in range(self.N):
-            a, _, chk_a = self.get_random_data(32)
-            b, _, chk_b = self.get_random_data(64)
-            res = self.combine(chk_a, chk_b, len(b))
-            self.assertEqual(res, self.checksum(a + b))
+    @hypothesis.given(a=hypothesis.strategies.binary(max_size=256),
+                      iv=hypothesis.strategies.runner().flatmap(
+                          lambda self: self.iv_strategy))
+    @hypothesis.example(a=b'', iv=0)
+    @hypothesis.example(a=b'\x00' * 32, iv=1)
+    @hypothesis.example(a=bytes(range(256)), iv=0x7fffffff)
+    def test_combine_empty(self, a, iv):
+        res = self.combine(iv, self.checksum(a), len(a))
+        self.assertEqual(res, self.checksum(a, iv))
+
+    @hypothesis.given(a=hypothesis.strategies.binary(max_size=256),
+                      b=hypothesis.strategies.binary(max_size=256))
+    @hypothesis.example(a=b'', b=b'')
+    @hypothesis.example(a=b'\x00' * 32, b=b'\xff' * 64)
+    @hypothesis.example(a=bytes(range(256)), b=bytes(range(255, -1, -1)))
+    def test_combine_no_iv(self, a, b):
+        res = self.combine(self.checksum(a), self.checksum(b), len(b))
+        self.assertEqual(res, self.checksum(a + b))
 
     def test_combine_no_iv_invalid_length(self):
         a, _, chk_a = self.get_random_data(32)
@@ -246,31 +251,48 @@ def test_combine_no_iv_invalid_length(self):
         self.assertRaises(OverflowError, self.combine, 0, 0, 2**1000)
         self.assertRaises(OverflowError, self.combine, 0, 0, -2**1000)
 
-    def test_combine_with_iv(self):
-        for _ in range(self.N):
-            a, iv_a, chk_a_with_iv = self.get_random_data(32, iv=-1)
-            chk_a_no_iv = self.checksum(a)
-            b, iv_b, chk_b_with_iv = self.get_random_data(64, iv=-1)
-            chk_b_no_iv = self.checksum(b)
-
-            # We can represent c = COMBINE(CHK(a, iv_a), CHK(b, iv_b)) as:
-            #
-            #   c = CHK(CHK(b'', iv_a) + CHK(a) + CHK(b'', iv_b) + CHK(b))
-            #     = COMBINE(
-            #           COMBINE(CHK(b'', iv_a), CHK(a)),
-            #           COMBINE(CHK(b'', iv_b), CHK(b)),
-            #       )
-            #     = COMBINE(COMBINE(iv_a, CHK(a)), COMBINE(iv_b, CHK(b)))
-            tmp0 = self.combine(iv_a, chk_a_no_iv, len(a))
-            tmp1 = self.combine(iv_b, chk_b_no_iv, len(b))
-            expected = self.combine(tmp0, tmp1, len(b))
-            checksum = self.combine(chk_a_with_iv, chk_b_with_iv, len(b))
-            self.assertEqual(checksum, expected)
+    @hypothesis.given(a=hypothesis.strategies.binary(max_size=256),
+                      b=hypothesis.strategies.binary(max_size=256),
+                      iv_a=hypothesis.strategies.runner().flatmap(
+                          lambda self: self.iv_strategy),
+                      iv_b=hypothesis.strategies.runner().flatmap(
+                          lambda self: self.iv_strategy))
+    @hypothesis.example(a=b'', b=b'', iv_a=0, iv_b=0)
+    @hypothesis.example(a=b'\x00' * 32, b=b'\xff' * 64, iv_a=1,
+                        iv_b=0x7fffffff)
+    @hypothesis.example(a=bytes(range(256)), b=bytes(range(255, -1, -1)),
+                        iv_a=0x12345678, iv_b=0x0badf00d)
+    def test_combine_with_iv(self, a, b, iv_a, iv_b):
+        chk_a_with_iv = self.checksum(a, iv_a)
+        chk_a_no_iv = self.checksum(a)
+        chk_b_with_iv = self.checksum(b, iv_b)
+        chk_b_no_iv = self.checksum(b)
+
+        # We can represent c = COMBINE(CHK(a, iv_a), CHK(b, iv_b)) as:
+        #
+        #   c = CHK(CHK(b'', iv_a) + CHK(a) + CHK(b'', iv_b) + CHK(b))
+        #     = COMBINE(
+        #           COMBINE(CHK(b'', iv_a), CHK(a)),
+        #           COMBINE(CHK(b'', iv_b), CHK(b)),
+        #       )
+        #     = COMBINE(COMBINE(iv_a, CHK(a)), COMBINE(iv_b, CHK(b)))
+        tmp0 = self.combine(iv_a, chk_a_no_iv, len(a))
+        tmp1 = self.combine(iv_b, chk_b_no_iv, len(b))
+        expected = self.combine(tmp0, tmp1, len(b))
+        checksum = self.combine(chk_a_with_iv, chk_b_with_iv, len(b))
+        self.assertEqual(checksum, expected)
+
+
+def _adler32_value(halves):
+    """Build a valid Adler-32 value from its two 16-bit sums."""
+    s1, s2 = halves
+    return (s2 << 16) | s1
 
 
 class CRC32CombineTestCase(ChecksumCombineMixin, unittest.TestCase):
 
     default_iv = 0
+    iv_strategy = hypothesis.strategies.integers(0, 0xffffffff)
 
     def _checksum(self, data, init):
         return zlib.crc32(data, init)
@@ -282,6 +304,12 @@ def combine(self, a, b, blen):
 class Adler32CombineTestCase(ChecksumCombineMixin, unittest.TestCase):
 
     default_iv = 1
+    # Both 16-bit sums of an Adler-32 value are reduced modulo 65521, and
+    # adler32_combine() is only defined on such values.
+    iv_strategy = hypothesis.strategies.tuples(
+        hypothesis.strategies.integers(0, 65520),
+        hypothesis.strategies.integers(0, 65520),
+    ).map(_adler32_value)
 
     def _checksum(self, data, init):
         return zlib.adler32(data, init)

From 6239ff33c20658049b586853fe9a50e3109bcf6b Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Tue, 20 Oct 2026 15:57:26 -0400
Subject: [PATCH 04/12] gh-107862: Add NEWS entry for the compression
 Hypothesis tests

---
 .../next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst  | 3 +++
 1 file changed, 3 insertions(+)
 create mode 100644 Misc/NEWS.d/next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst

diff --git a/Misc/NEWS.d/next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst b/Misc/NEWS.d/next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst
new file mode 100644
index 00000000000000..84d94a9f8b02b5
--- /dev/null
+++ b/Misc/NEWS.d/next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst
@@ -0,0 +1,3 @@
+Add Hypothesis property tests for the incremental compression APIs of
+:mod:`bz2`, :mod:`lzma`, :mod:`zlib`, :mod:`gzip` and :mod:`compression.zstd`,
+and convert the ``zlib`` checksum combination tests to Hypothesis.

From 61d818984b4ecea9f58fef1641dcb0ff9a1f1311 Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Tue, 20 Oct 2026 17:01:02 -0400
Subject: [PATCH 05/12] Drop the NEWS entry for the compression Hypothesis
 tests

---
 .../next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst  | 3 ---
 1 file changed, 3 deletions(-)
 delete mode 100644 Misc/NEWS.d/next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst

diff --git a/Misc/NEWS.d/next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst b/Misc/NEWS.d/next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst
deleted file mode 100644
index 84d94a9f8b02b5..00000000000000
--- a/Misc/NEWS.d/next/Tests/2026-10-20-11-30-00.gh-issue-107862.zK4pLm.rst
+++ /dev/null
@@ -1,3 +0,0 @@
-Add Hypothesis property tests for the incremental compression APIs of
-:mod:`bz2`, :mod:`lzma`, :mod:`zlib`, :mod:`gzip` and :mod:`compression.zstd`,
-and convert the ``zlib`` checksum combination tests to Hypothesis.

From adb21ce36da426e168936ce415dc65777d27b4ec Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Tue, 20 Oct 2026 17:01:02 -0400
Subject: [PATCH 06/12] Use itertools, functools.partial and unittest
 assertions in the compression tests

Review feedback on the first version:

- Cycling through the per-call values (max_length, flush mode) is now
  itertools.cycle zipped with the chunks instead of a hand-rolled index
  helper.
- The backend tables use functools.partial rather than lambdas wherever
  the call can be expressed that way.
- The state machines checked with plain assert statements. They now go
  through a small base class that borrows the unittest assertion methods
  from a throwaway TestCase, so a failing rule reports the same diffs as
  everything else in the suite.

The chunk splitter stays a custom generator: itertools.batched only makes
equal-size pieces, and the point of the sizes list is to put the call
boundaries at arbitrary positions in the stream (the pinned needs_input
reproducers depend on that).
---
 Lib/test/test_compression_property.py | 150 +++++++++++++++-----------
 1 file changed, 87 insertions(+), 63 deletions(-)

diff --git a/Lib/test/test_compression_property.py b/Lib/test/test_compression_property.py
index 1ece7329e3e2e7..12008d09b91f32 100644
--- a/Lib/test/test_compression_property.py
+++ b/Lib/test/test_compression_property.py
@@ -14,7 +14,9 @@
 fuzz targets through each test's ``hypothesis.fuzz_one_input`` attribute.
 """
 import dataclasses
+import functools
 import io
+import itertools
 import unittest
 from collections.abc import Callable
 
@@ -62,7 +64,11 @@ def _repeat(pair):
 
 
 def _chunks(data, sizes):
-    """Split *data* at the given sizes; whatever is left is the last chunk."""
+    """Split *data* at the given sizes; whatever is left is the last chunk.
+
+    Unlike ``itertools.batched`` the pieces have different sizes, so the
+    boundaries between calls land at arbitrary positions in the stream.
+    """
     pos = 0
     for size in sizes:
         yield data[pos:pos + size]
@@ -70,8 +76,9 @@ def _chunks(data, sizes):
     yield data[pos:]
 
 
-def _cycle(values, index, default):
-    return values[index % len(values)] if values else default
+def _chunks_with(data, sizes, values, default):
+    """Pair each chunk with the next of *values*, cycling through them."""
+    return zip(_chunks(data, sizes), itertools.cycle(values or [default]))
 
 
 @dataclasses.dataclass(frozen=True)
@@ -93,8 +100,8 @@ def _bz2_backend():
     return _Backend(
         name="bz2",
         new_decompressor=bz2.BZ2Decompressor,
-        new_compressor=lambda: bz2.BZ2Compressor(1),
-        compress=lambda data: bz2.compress(data, 1),
+        new_compressor=functools.partial(bz2.BZ2Compressor, 1),
+        compress=functools.partial(bz2.compress, compresslevel=1),
         decompress=bz2.decompress,
         error=OSError,
         decompress_errors=(OSError, ValueError),
@@ -102,16 +109,20 @@ def _bz2_backend():
     )
 
 
+def _open_gzip(fp):
+    return gzip.GzipFile(fileobj=fp)
+
+
 def _gzip_backend():
     return _Backend(
         name="gzip",
-        new_decompressor=lambda: zlib._ZlibDecompressor(31),
-        new_compressor=lambda: zlib.compressobj(1, wbits=31),
-        compress=lambda data: gzip.compress(data, 1),
+        new_decompressor=functools.partial(zlib._ZlibDecompressor, 31),
+        new_compressor=functools.partial(zlib.compressobj, 1, wbits=31),
+        compress=functools.partial(gzip.compress, compresslevel=1),
         decompress=gzip.decompress,
         error=zlib.error,
         decompress_errors=(gzip.BadGzipFile, EOFError, zlib.error),
-        open_file=lambda fp: gzip.GzipFile(fileobj=fp),
+        open_file=_open_gzip,
     )
 
 
@@ -119,8 +130,8 @@ def _lzma_backend():
     return _Backend(
         name="lzma",
         new_decompressor=lzma.LZMADecompressor,
-        new_compressor=lambda: lzma.LZMACompressor(preset=0),
-        compress=lambda data: lzma.compress(data, preset=0),
+        new_compressor=functools.partial(lzma.LZMACompressor, preset=0),
+        compress=functools.partial(lzma.compress, preset=0),
         decompress=lzma.decompress,
         error=lzma.LZMAError,
         decompress_errors=(lzma.LZMAError,),
@@ -131,10 +142,10 @@ def _lzma_backend():
 def _zlib_backend(wbits):
     return _Backend(
         name=f"zlib(wbits={wbits})",
-        new_decompressor=lambda: zlib._ZlibDecompressor(wbits),
-        new_compressor=lambda: zlib.compressobj(1, wbits=wbits),
-        compress=lambda data: zlib.compress(data, 1, wbits=wbits),
-        decompress=lambda data: zlib.decompress(data, wbits=wbits),
+        new_decompressor=functools.partial(zlib._ZlibDecompressor, wbits),
+        new_compressor=functools.partial(zlib.compressobj, 1, wbits=wbits),
+        compress=functools.partial(zlib.compress, level=1, wbits=wbits),
+        decompress=functools.partial(zlib.decompress, wbits=wbits),
         error=zlib.error,
         decompress_errors=(zlib.error,),
     )
@@ -144,8 +155,8 @@ def _zstd_backend():
     return _Backend(
         name="zstd",
         new_decompressor=zstd.ZstdDecompressor,
-        new_compressor=lambda: zstd.ZstdCompressor(1),
-        compress=lambda data: zstd.compress(data, 1),
+        new_compressor=functools.partial(zstd.ZstdCompressor, 1),
+        compress=functools.partial(zstd.compress, level=1),
         decompress=zstd.decompress,
         error=zstd.ZstdError,
         decompress_errors=(zstd.ZstdError,),
@@ -153,11 +164,28 @@ def _zstd_backend():
     )
 
 
+class _StateMachine(stateful.RuleBasedStateMachine):
+    """A state machine that checks with the unittest assertion methods.
+
+    The methods are borrowed from a throwaway ``TestCase`` so that failures
+    inside rules get the same diagnostics as the rest of the test suite.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self._case = unittest.TestCase()
+
+    def __getattr__(self, name):
+        if name.startswith("assert"):
+            return getattr(self._case, name)
+        raise AttributeError(name)
+
+
 # ---------------------------------------------------------------------------
 # Incremental decompressor protocol
 # ---------------------------------------------------------------------------
 
-class _DecompressorMachine(stateful.RuleBasedStateMachine):
+class _DecompressorMachine(_StateMachine):
     """Feed one compressed stream through a decompressor in random pieces.
 
     After every call the output must be a prefix of the original data and no
@@ -181,23 +209,21 @@ def start(self, data, trailer):
 
     def _feed(self, chunk, max_length):
         if self.decomp.eof:
-            try:
+            with self.assertRaises(EOFError):
                 self.decomp.decompress(chunk, max_length)
-            except EOFError:
-                return
-            raise AssertionError("decompress() after eof did not raise")
+            return
         ret = self.decomp.decompress(chunk, max_length)
-        assert isinstance(ret, bytes)
+        self.assertIsInstance(ret, bytes)
         if max_length >= 0:
-            assert len(ret) <= max_length, (len(ret), max_length)
+            self.assertLessEqual(len(ret), max_length)
         self.out += ret
-        assert self.data.startswith(self.out)
+        self.assertStartsWith(self.data, self.out)
         if self.decomp.eof:
-            assert self.out == self.data
+            self.assertEqual(self.out, self.data)
             rest = self.stream[self.pos:]
-            assert self.decomp.unused_data + rest == self.trailer
+            self.assertEqual(self.decomp.unused_data + rest, self.trailer)
         else:
-            assert self.decomp.unused_data == b""
+            self.assertEqual(self.decomp.unused_data, b"")
 
     def _feed_next(self, n, max_length):
         chunk = self.stream[self.pos:self.pos + n]
@@ -220,15 +246,15 @@ def finish(self):
             if self.decomp.eof:
                 break
             self._feed(b"", -1)
-        assert self.decomp.eof
-        assert self.out == self.data
+        self.assertTrue(self.decomp.eof)
+        self.assertEqual(self.out, self.data)
 
     @stateful.invariant()
     def output_is_a_prefix(self):
-        assert self.data.startswith(self.out)
-        assert isinstance(self.decomp.needs_input, bool)
+        self.assertStartsWith(self.data, self.out)
+        self.assertIsInstance(self.decomp.needs_input, bool)
         if self.decomp.eof:
-            assert self.out == self.data
+            self.assertEqual(self.out, self.data)
 
 
 class _DecompressorTests:
@@ -250,8 +276,7 @@ def _walk(self, data, sizes, max_lengths, trailer=b""):
         decomp = self.backend.new_decompressor()
         out = b""
         fed = 0
-        for i, chunk in enumerate(_chunks(stream, sizes)):
-            max_length = _cycle(max_lengths, i, -1)
+        for chunk, max_length in _chunks_with(stream, sizes, max_lengths, -1):
             if decomp.eof:
                 with self.assertRaises(EOFError):
                     decomp.decompress(chunk, max_length)
@@ -261,7 +286,7 @@ def _walk(self, data, sizes, max_lengths, trailer=b""):
             if max_length >= 0:
                 self.assertLessEqual(len(ret), max_length)
             out += ret
-            self.assertTrue(data.startswith(out))
+            self.assertStartsWith(data, out)
             yield decomp
         while not decomp.eof:
             out += decomp.decompress(b"", -1)
@@ -331,7 +356,7 @@ def test_truncated_stream_does_not_raise(self, data, cut):
         cut %= len(stream) + 1
         decomp = self.backend.new_decompressor()
         out = decomp.decompress(stream[:cut], -1)
-        self.assertTrue(data.startswith(out))
+        self.assertStartsWith(data, out)
         self.assertEqual(decomp.eof, cut == len(stream))
         if not decomp.eof:
             self.assertTrue(decomp.needs_input)
@@ -383,8 +408,9 @@ def test_invalid_input(self, garbage, data, index, byte, sizes,
             with self.subTest(blob=blob):
                 decomp = self.backend.new_decompressor()
                 try:
-                    for i, chunk in enumerate(_chunks(blob, sizes)):
-                        decomp.decompress(chunk, _cycle(max_lengths, i, -1))
+                    for chunk, max_length in _chunks_with(blob, sizes,
+                                                          max_lengths, -1):
+                        decomp.decompress(chunk, max_length)
                 except (self.backend.error, EOFError):
                     pass
                 try:
@@ -568,10 +594,10 @@ def test_flush_modes(self, data, sizes, modes, level, wbits):
         # stream that decompresses to the input.
         comp = zlib.compressobj(level, wbits=wbits)
         stream = b""
-        for i, chunk in enumerate(_chunks(data, sizes)):
+        for chunk, mode in _chunks_with(data, sizes, modes, None):
             stream += comp.compress(chunk)
-            if modes:
-                stream += comp.flush(getattr(zlib, _cycle(modes, i, None)))
+            if mode is not None:
+                stream += comp.flush(getattr(zlib, mode))
         stream += comp.flush()
         self.assertEqual(zlib.decompress(stream, wbits=wbits), data)
         decomp = zlib._ZlibDecompressor(wbits)
@@ -601,9 +627,8 @@ def test_compress_modes(self, data, sizes, modes):
         comp = zstd.ZstdCompressor(1)
         self.assertEqual(comp.last_mode, zstd.ZstdCompressor.FLUSH_FRAME)
         stream = b""
-        for i, chunk in enumerate(_chunks(data, sizes)):
-            mode = getattr(zstd.ZstdCompressor,
-                           _cycle(modes, i, "CONTINUE"))
+        for chunk, mode in _chunks_with(data, sizes, modes, "CONTINUE"):
+            mode = getattr(zstd.ZstdCompressor, mode)
             stream += comp.compress(chunk, mode)
             self.assertEqual(comp.last_mode, mode)
         stream += comp.flush()
@@ -651,15 +676,14 @@ def test_max_length_and_unconsumed_tail(self, data, trailer, sizes,
         decomp = zlib.decompressobj(wbits)
         out = b""
         fed = 0
-        for i, chunk in enumerate(_chunks(stream, sizes)):
-            max_length = _cycle(max_lengths, i, 0)
+        for chunk, max_length in _chunks_with(stream, sizes, max_lengths, 0):
             ret = decomp.decompress(decomp.unconsumed_tail + chunk,
                                     max_length)
             fed += len(chunk)
             if max_length:
                 self.assertLessEqual(len(ret), max_length)
             out += ret
-            self.assertTrue(data.startswith(out))
+            self.assertStartsWith(data, out)
             if decomp.eof:
                 self.assertEqual(out, data)
                 self.assertEqual(decomp.unconsumed_tail, b"")
@@ -680,12 +704,12 @@ def test_flush(self, data, cut):
         cut %= len(stream) + 1
         decomp = zlib.decompressobj()
         out = decomp.decompress(stream[:cut])
-        self.assertTrue(data.startswith(out))
+        self.assertStartsWith(data, out)
         self.assertEqual(decomp.eof, cut == len(stream))
         # flush() on a truncated stream returns what it can without raising;
         # on a complete stream it is a no-op and may be repeated.
         out += decomp.flush()
-        self.assertTrue(data.startswith(out))
+        self.assertStartsWith(data, out)
         if cut == len(stream):
             self.assertEqual(out, data)
             self.assertTrue(decomp.eof)
@@ -753,7 +777,7 @@ def test_zlib_wrapper(self, data, zdict, level):
 read_size_strategy = st.integers(0, 300) | st.just(-1)
 
 
-class _StreamReaderMachine(stateful.RuleBasedStateMachine):
+class _StreamReaderMachine(_StateMachine):
     """Compare a compressed file object against io.BytesIO on the same data.
 
     The compressed input is several concatenated streams (members, frames),
@@ -779,17 +803,17 @@ def _remaining(self):
 
     @stateful.rule(size=read_size_strategy)
     def read(self, size):
-        assert self.file.read(size) == self.oracle.read(size)
+        self.assertEqual(self.file.read(size), self.oracle.read(size))
 
     @stateful.rule(size=read_size_strategy)
     def read1(self, size):
         ret = self.file.read1(size)
         remaining = self._remaining()
-        assert remaining.startswith(ret)
+        self.assertStartsWith(remaining, ret)
         if size != 0 and remaining:
-            assert ret, "read1() returned nothing before EOF"
+            self.assertTrue(ret, "read1() returned nothing before EOF")
         if size >= 0:
-            assert len(ret) <= size
+            self.assertLessEqual(len(ret), size)
         self.oracle.seek(len(ret), io.SEEK_CUR)
 
     @stateful.rule(size=st.integers(0, 300))
@@ -797,22 +821,22 @@ def peek(self, size):
         pos = self.file.tell()
         ret = self.file.peek(size)
         remaining = self._remaining()
-        assert remaining.startswith(ret)
+        self.assertStartsWith(remaining, ret)
         if remaining:
-            assert ret, "peek() returned nothing before EOF"
-        assert self.file.tell() == pos
+            self.assertTrue(ret, "peek() returned nothing before EOF")
+        self.assertEqual(self.file.tell(), pos)
 
     @stateful.rule(size=st.integers(0, 300))
     def readinto(self, size):
         buf = bytearray(size)
         n = self.file.readinto(buf)
         expected = self.oracle.read(size)
-        assert n == len(expected)
-        assert buf[:n] == expected
+        self.assertEqual(n, len(expected))
+        self.assertEqual(buf[:n], expected)
 
     @stateful.rule()
     def readline(self):
-        assert self.file.readline() == self.oracle.readline()
+        self.assertEqual(self.file.readline(), self.oracle.readline())
 
     @stateful.rule(offset=st.integers(-10**6, 10**6),
                    whence=st.sampled_from([io.SEEK_SET, io.SEEK_CUR,
@@ -826,12 +850,12 @@ def seek(self, offset, whence):
         else:
             target = len(self.plain) + offset
         target = min(max(target, 0), len(self.plain))
-        assert self.file.seek(offset, whence) == target
+        self.assertEqual(self.file.seek(offset, whence), target)
         self.oracle.seek(target)
 
     @stateful.invariant()
     def positions_agree(self):
-        assert self.file.tell() == self.oracle.tell()
+        self.assertEqual(self.file.tell(), self.oracle.tell())
 
 
 class _StreamReaderTests:

From a1335bc07833fadf7718cb91c143050ab5655d71 Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Tue, 20 Oct 2026 18:04:09 -0400
Subject: [PATCH 07/12] Explain the machine/companion split and the assertion
 forwarding

---
 Lib/test/test_compression_property.py | 16 +++++++++++++---
 1 file changed, 13 insertions(+), 3 deletions(-)

diff --git a/Lib/test/test_compression_property.py b/Lib/test/test_compression_property.py
index 12008d09b91f32..f321db7b1e00e0 100644
--- a/Lib/test/test_compression_property.py
+++ b/Lib/test/test_compression_property.py
@@ -9,9 +9,15 @@
 random ``max_length`` values, using the module-level one-shot functions as
 oracles.
 
-The strategies only use ``.map()``, never ``.filter()`` or ``assume()``, so
-every generated input is a valid test case.  That keeps the tests usable as
-fuzz targets through each test's ``hypothesis.fuzz_one_input`` attribute.
+Each API is covered twice.  A ``RuleBasedStateMachine`` declares the calls
+a user may make (feed a chunk, seek, read a line, ...) with their argument
+ranges and what must hold after each one; Hypothesis then generates and
+shrinks whole call sequences against it.  State machines have no
+``@example`` equivalent, so each one is paired with a ``@given`` companion
+test that expresses one call sequence as data (chunk sizes, ``max_length``
+values, read sizes).  The companions carry the pinned examples, including
+the shrunk reproducers for the known bugs, and are what the stub replays
+when Hypothesis is not installed; the machines are skipped then.
 """
 import dataclasses
 import functools
@@ -175,6 +181,10 @@ def __init__(self):
         super().__init__()
         self._case = unittest.TestCase()
 
+    # Hypothesis runs a machine from the ``runTest`` of the ``TestCase`` it
+    # generates, creating a fresh instance with ``Machine()`` for every
+    # example.  The machine is not that test case and has none of its
+    # assertion state, so the methods are forwarded to a throwaway one.
     def __getattr__(self, name):
         if name.startswith("assert"):
             return getattr(self._case, name)

From 2172e72176eb663d968aa62f769936147a20c19b Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Tue, 20 Oct 2026 18:25:00 -0400
Subject: [PATCH 08/12] Build the machine test cases with a factory and keep
 the companions explicit

The companion test classes stay ordinary class statements, since they are
the ones with content worth reading and grepping for. The per-backend
machine subclass only adds a backend attribute and Machine.TestCase is
generated by Hypothesis anyway, so _machine_test() now builds both from
a name, a machine base, a backend factory and a requires_* decorator.

The backend factories return None when their module is missing, which
removes the conditional backend assignments from every class. The
per-backend overrides for the known bugs (and the lzma max_length=0
carve-out) are mixins listed first in the bases, so the MRO shows which
backends carry one.
---
 Lib/test/test_compression_property.py | 217 +++++++++++++-------------
 1 file changed, 112 insertions(+), 105 deletions(-)

diff --git a/Lib/test/test_compression_property.py b/Lib/test/test_compression_property.py
index f321db7b1e00e0..0538f771dd43d7 100644
--- a/Lib/test/test_compression_property.py
+++ b/Lib/test/test_compression_property.py
@@ -102,7 +102,12 @@ class _Backend:
     open_file: Callable[[io.BytesIO], io.BufferedIOBase] | None = None
 
 
+# Each factory returns None when its module is missing, so that the test
+# classes below can be defined unconditionally and skipped by requires_*().
+
 def _bz2_backend():
+    if bz2 is None:
+        return None
     return _Backend(
         name="bz2",
         new_decompressor=bz2.BZ2Decompressor,
@@ -120,6 +125,8 @@ def _open_gzip(fp):
 
 
 def _gzip_backend():
+    if gzip is None or zlib is None:
+        return None
     return _Backend(
         name="gzip",
         new_decompressor=functools.partial(zlib._ZlibDecompressor, 31),
@@ -133,6 +140,8 @@ def _gzip_backend():
 
 
 def _lzma_backend():
+    if lzma is None:
+        return None
     return _Backend(
         name="lzma",
         new_decompressor=lzma.LZMADecompressor,
@@ -146,6 +155,8 @@ def _lzma_backend():
 
 
 def _zlib_backend(wbits):
+    if zlib is None:
+        return None
     return _Backend(
         name=f"zlib(wbits={wbits})",
         new_decompressor=functools.partial(zlib._ZlibDecompressor, wbits),
@@ -158,6 +169,8 @@ def _zlib_backend(wbits):
 
 
 def _zstd_backend():
+    if zstd is None:
+        return None
     return _Backend(
         name="zstd",
         new_decompressor=zstd.ZstdDecompressor,
@@ -191,6 +204,16 @@ def __getattr__(self, name):
         raise AttributeError(name)
 
 
+def _machine_test(name, machine_base, backend_factory, requires):
+    """Build the unittest TestCase that runs *machine_base* on one backend.
+
+    The machine subclass only adds the backend, and ``Machine.TestCase`` is
+    generated by Hypothesis anyway, so neither needs a class statement.
+    """
+    machine = type(name, (machine_base,), {"backend": backend_factory()})
+    return requires(machine.TestCase)
+
+
 # ---------------------------------------------------------------------------
 # Incremental decompressor protocol
 # ---------------------------------------------------------------------------
@@ -429,10 +452,10 @@ def test_invalid_input(self, garbage, data, index, byte, sizes,
                     pass
 
 
-@support.requires_bz2()
-class BZ2DecompressorTests(_DecompressorTests, unittest.TestCase):
-    backend = _bz2_backend() if bz2 is not None else None
+# Known bugs, as mixins that override the affected test for one backend.
+# They go first in the bases so that they win over _DecompressorTests.
 
+class _BZ2NeedsInputBug:
     # bz2 reports needs_input=True as soon as its input buffer is empty, even
     # when max_length cut the output short and more of it is pending.
     @unittest.expectedFailure
@@ -446,19 +469,37 @@ def test_needs_input_means_drained(self, data, sizes, max_lengths):
         self._check_needs_input_means_drained(data, sizes, max_lengths)
 
 
+class _ZlibNeedsInputBug:
+    # zlib._ZlibDecompressor has the same needs_input problem as bz2.  One
+    # example per wbits variant: zlib wrapper, raw deflate, gzip wrapper.
+    @unittest.expectedFailure
+    @hypothesis.given(
+        data=data_strategy,
+        sizes=chunk_sizes_strategy,
+        max_lengths=max_lengths_strategy,
+    )
+    @hypothesis.example(data=b"\x00", sizes=[4], max_lengths=[0])
+    @hypothesis.example(data=b"\x00" * 21, sizes=[], max_lengths=[2])
+    @hypothesis.example(
+        data=b"\x00\x00\x00\x01\x00\x00\x02\x00\x00\x03\x00\x00"
+             b"\x90\x00\x90\x90\x90\x91\x90" * 2,
+        sizes=[34],
+        max_lengths=[20],
+    )
+    def test_needs_input_means_drained(self, data, sizes, max_lengths):
+        self._check_needs_input_means_drained(data, sizes, max_lengths)
+
+
 # liblzma reports an error when two consecutive calls can make no progress,
 # which lzma.LZMADecompressor lets through as LZMAError("Insufficient buffer
 # space") for two max_length=0 calls in a row with nothing buffered.  The
 # broad lzma tests avoid max_length=0 so that they keep checking everything
-# else; test_repeated_max_length_zero below pins the bug.
+# else; test_repeated_max_length_zero pins the bug.
 lzma_max_length_strategy = st.integers(1, 64) | st.just(-1)
 lzma_max_lengths_strategy = st.lists(lzma_max_length_strategy, max_size=16)
 
 
-@support.requires_lzma()
-class LZMADecompressorTests(_DecompressorTests, unittest.TestCase):
-    backend = _lzma_backend() if lzma is not None else None
-
+class _LZMAMaxLengthZeroBug:
     @hypothesis.given(
         data=data_strategy,
         trailer=trailer_strategy,
@@ -493,88 +534,66 @@ def test_repeated_max_length_zero(self, data, count):
         self._check_repeated_max_length_zero(data, count)
 
 
-class _ZlibDecompressorTests(_DecompressorTests):
-    # zlib._ZlibDecompressor has the same needs_input problem as bz2.  One
-    # example per wbits variant: zlib wrapper, raw deflate, gzip wrapper.
-    @unittest.expectedFailure
-    @hypothesis.given(
-        data=data_strategy,
-        sizes=chunk_sizes_strategy,
-        max_lengths=max_lengths_strategy,
-    )
-    @hypothesis.example(data=b"\x00", sizes=[4], max_lengths=[0])
-    @hypothesis.example(data=b"\x00" * 21, sizes=[], max_lengths=[2])
-    @hypothesis.example(
-        data=b"\x00\x00\x00\x01\x00\x00\x02\x00\x00\x03\x00\x00"
-             b"\x90\x00\x90\x90\x90\x91\x90" * 2,
-        sizes=[34],
-        max_lengths=[20],
-    )
-    def test_needs_input_means_drained(self, data, sizes, max_lengths):
-        self._check_needs_input_means_drained(data, sizes, max_lengths)
-
-
-@support.requires_zlib()
-class ZlibDecompressorTests(_ZlibDecompressorTests, unittest.TestCase):
-    backend = _zlib_backend(zlib.MAX_WBITS) if zlib is not None else None
-
-
-@support.requires_zlib()
-class RawDeflateDecompressorTests(_ZlibDecompressorTests, unittest.TestCase):
-    backend = _zlib_backend(-15) if zlib is not None else None
-
-
-@support.requires_gzip()
-class GzipDecompressorTests(_ZlibDecompressorTests, unittest.TestCase):
-    backend = _gzip_backend() if gzip is not None else None
-
-
-@support.requires_zstd()
-class ZstdDecompressorTests(_DecompressorTests, unittest.TestCase):
-    backend = _zstd_backend() if zstd is not None else None
-
-
-class BZ2DecompressorMachine(_DecompressorMachine):
-    backend = _bz2_backend() if bz2 is not None else None
-
-
 class LZMADecompressorMachine(_DecompressorMachine):
-    backend = _lzma_backend() if lzma is not None else None
-
     # See lzma_max_length_strategy above.
     @stateful.rule(n=st.integers(0, 64), max_length=lzma_max_length_strategy)
     def feed(self, n, max_length):
         self._feed_next(n, max_length)
 
 
-class ZlibDecompressorMachine(_DecompressorMachine):
-    backend = _zlib_backend(zlib.MAX_WBITS) if zlib is not None else None
+@support.requires_bz2()
+class BZ2DecompressorTests(_BZ2NeedsInputBug, _DecompressorTests,
+                           unittest.TestCase):
+    backend = _bz2_backend()
+
+
+@support.requires_gzip()
+class GzipDecompressorTests(_ZlibNeedsInputBug, _DecompressorTests,
+                            unittest.TestCase):
+    backend = _gzip_backend()
 
 
-class RawDeflateDecompressorMachine(_DecompressorMachine):
-    backend = _zlib_backend(-15) if zlib is not None else None
+@support.requires_lzma()
+class LZMADecompressorTests(_LZMAMaxLengthZeroBug, _DecompressorTests,
+                            unittest.TestCase):
+    backend = _lzma_backend()
 
 
-class GzipDecompressorMachine(_DecompressorMachine):
-    backend = _gzip_backend() if gzip is not None else None
+@support.requires_zlib()
+class ZlibDecompressorTests(_ZlibNeedsInputBug, _DecompressorTests,
+                            unittest.TestCase):
+    backend = _zlib_backend(15)
 
 
-class ZstdDecompressorMachine(_DecompressorMachine):
-    backend = _zstd_backend() if zstd is not None else None
+@support.requires_zlib()
+class RawDeflateDecompressorTests(_ZlibNeedsInputBug, _DecompressorTests,
+                                  unittest.TestCase):
+    backend = _zlib_backend(-15)
 
 
-BZ2DecompressorMachineTest = support.requires_bz2()(
-    BZ2DecompressorMachine.TestCase)
-LZMADecompressorMachineTest = support.requires_lzma()(
-    LZMADecompressorMachine.TestCase)
-ZlibDecompressorMachineTest = support.requires_zlib()(
-    ZlibDecompressorMachine.TestCase)
-RawDeflateDecompressorMachineTest = support.requires_zlib()(
-    RawDeflateDecompressorMachine.TestCase)
-GzipDecompressorMachineTest = support.requires_gzip()(
-    GzipDecompressorMachine.TestCase)
-ZstdDecompressorMachineTest = support.requires_zstd()(
-    ZstdDecompressorMachine.TestCase)
+@support.requires_zstd()
+class ZstdDecompressorTests(_DecompressorTests, unittest.TestCase):
+    backend = _zstd_backend()
+
+
+BZ2DecompressorMachineTest = _machine_test(
+    "BZ2DecompressorMachine", _DecompressorMachine, _bz2_backend,
+    support.requires_bz2())
+GzipDecompressorMachineTest = _machine_test(
+    "GzipDecompressorMachine", _DecompressorMachine, _gzip_backend,
+    support.requires_gzip())
+LZMADecompressorMachineTest = _machine_test(
+    "LZMADecompressorMachine", LZMADecompressorMachine, _lzma_backend,
+    support.requires_lzma())
+ZlibDecompressorMachineTest = _machine_test(
+    "ZlibDecompressorMachine", _DecompressorMachine,
+    functools.partial(_zlib_backend, 15), support.requires_zlib())
+RawDeflateDecompressorMachineTest = _machine_test(
+    "RawDeflateDecompressorMachine", _DecompressorMachine,
+    functools.partial(_zlib_backend, -15), support.requires_zlib())
+ZstdDecompressorMachineTest = _machine_test(
+    "ZstdDecompressorMachine", _DecompressorMachine, _zstd_backend,
+    support.requires_zstd())
 
 
 # ---------------------------------------------------------------------------
@@ -902,48 +921,36 @@ def test_read_sizes(self, members, sizes):
 
 @support.requires_bz2()
 class BZ2StreamReaderTests(_StreamReaderTests, unittest.TestCase):
-    backend = _bz2_backend() if bz2 is not None else None
+    backend = _bz2_backend()
 
 
 @support.requires_gzip()
 class GzipStreamReaderTests(_StreamReaderTests, unittest.TestCase):
-    backend = _gzip_backend() if gzip is not None else None
+    backend = _gzip_backend()
 
 
 @support.requires_lzma()
 class LZMAStreamReaderTests(_StreamReaderTests, unittest.TestCase):
-    backend = _lzma_backend() if lzma is not None else None
+    backend = _lzma_backend()
 
 
 @support.requires_zstd()
 class ZstdStreamReaderTests(_StreamReaderTests, unittest.TestCase):
-    backend = _zstd_backend() if zstd is not None else None
-
-
-class BZ2StreamReaderMachine(_StreamReaderMachine):
-    backend = _bz2_backend() if bz2 is not None else None
-
-
-class GzipStreamReaderMachine(_StreamReaderMachine):
-    backend = _gzip_backend() if gzip is not None else None
-
-
-class LZMAStreamReaderMachine(_StreamReaderMachine):
-    backend = _lzma_backend() if lzma is not None else None
-
-
-class ZstdStreamReaderMachine(_StreamReaderMachine):
-    backend = _zstd_backend() if zstd is not None else None
-
-
-BZ2StreamReaderMachineTest = support.requires_bz2()(
-    BZ2StreamReaderMachine.TestCase)
-GzipStreamReaderMachineTest = support.requires_gzip()(
-    GzipStreamReaderMachine.TestCase)
-LZMAStreamReaderMachineTest = support.requires_lzma()(
-    LZMAStreamReaderMachine.TestCase)
-ZstdStreamReaderMachineTest = support.requires_zstd()(
-    ZstdStreamReaderMachine.TestCase)
+    backend = _zstd_backend()
+
+
+BZ2StreamReaderMachineTest = _machine_test(
+    "BZ2StreamReaderMachine", _StreamReaderMachine, _bz2_backend,
+    support.requires_bz2())
+GzipStreamReaderMachineTest = _machine_test(
+    "GzipStreamReaderMachine", _StreamReaderMachine, _gzip_backend,
+    support.requires_gzip())
+LZMAStreamReaderMachineTest = _machine_test(
+    "LZMAStreamReaderMachine", _StreamReaderMachine, _lzma_backend,
+    support.requires_lzma())
+ZstdStreamReaderMachineTest = _machine_test(
+    "ZstdStreamReaderMachine", _StreamReaderMachine, _zstd_backend,
+    support.requires_zstd())
 
 
 # ---------------------------------------------------------------------------

From 0442665cdebd05845730387cf19b47671e86fcd3 Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Tue, 20 Oct 2026 20:22:56 -0400
Subject: [PATCH 09/12] Fix the known-bug tests on debug builds and zlib-ng

The test pinning gh-160001 expects a SystemError, but in a debug build
the missing exception trips a C assertion in the call machinery before
the SystemError is raised, which aborts the test worker.  All CI builds
are debug builds, so the test is skipped there until the bug is fixed.

The gzip examples for gh-159999 also depend on where inflate stops
consuming input, and none of them triggered the bug with zlib-ng, which
the Windows builds use.  This adds an example found against zlib-ng
2.2.4 in zlib-compat mode; the zlib wrapper and raw deflate examples
already fail with both libraries.

The comments now also reference the filed issues.
---
 Lib/test/test_compression_property.py | 26 +++++++++++++++++++-------
 1 file changed, 19 insertions(+), 7 deletions(-)

diff --git a/Lib/test/test_compression_property.py b/Lib/test/test_compression_property.py
index 0538f771dd43d7..dc476b076e4c8a 100644
--- a/Lib/test/test_compression_property.py
+++ b/Lib/test/test_compression_property.py
@@ -457,7 +457,8 @@ def test_invalid_input(self, garbage, data, index, byte, sizes,
 
 class _BZ2NeedsInputBug:
     # bz2 reports needs_input=True as soon as its input buffer is empty, even
-    # when max_length cut the output short and more of it is pending.
+    # when max_length cut the output short and more of it is pending
+    # (gh-159999).
     @unittest.expectedFailure
     @hypothesis.given(
         data=data_strategy,
@@ -470,8 +471,11 @@ def test_needs_input_means_drained(self, data, sizes, max_lengths):
 
 
 class _ZlibNeedsInputBug:
-    # zlib._ZlibDecompressor has the same needs_input problem as bz2.  One
-    # example per wbits variant: zlib wrapper, raw deflate, gzip wrapper.
+    # zlib._ZlibDecompressor has the same needs_input problem as bz2
+    # (gh-159999).  The examples depend on where the compressor puts block
+    # boundaries, so there are several: at least one must fail for each
+    # wbits variant (zlib wrapper, raw deflate, gzip wrapper) with both zlib
+    # and zlib-ng.
     @unittest.expectedFailure
     @hypothesis.given(
         data=data_strategy,
@@ -486,15 +490,20 @@ class _ZlibNeedsInputBug:
         sizes=[34],
         max_lengths=[20],
     )
+    @hypothesis.example(
+        data=b"\x00\x00\x00\x00\x01\x00\x00\x00\x02\x00\x00\x03" * 2,
+        sizes=[25],
+        max_lengths=[12],
+    )
     def test_needs_input_means_drained(self, data, sizes, max_lengths):
         self._check_needs_input_means_drained(data, sizes, max_lengths)
 
 
 # liblzma reports an error when two consecutive calls can make no progress,
 # which lzma.LZMADecompressor lets through as LZMAError("Insufficient buffer
-# space") for two max_length=0 calls in a row with nothing buffered.  The
-# broad lzma tests avoid max_length=0 so that they keep checking everything
-# else; test_repeated_max_length_zero pins the bug.
+# space") for two max_length=0 calls in a row with nothing buffered
+# (gh-160000).  The broad lzma tests avoid max_length=0 so that they keep
+# checking everything else; test_repeated_max_length_zero pins the bug.
 lzma_max_length_strategy = st.integers(1, 64) | st.just(-1)
 lzma_max_lengths_strategy = st.lists(lzma_max_length_strategy, max_size=16)
 
@@ -784,7 +793,10 @@ def test_raw_deflate(self, data, zdict, level):
 
     # _ZlibDecompressor only applies zdict up front for raw deflate; for a
     # zlib-wrapped stream it hits Z_NEED_DICT and returns NULL without
-    # setting an exception, which surfaces as SystemError.
+    # setting an exception (gh-160001).  That surfaces as SystemError in a
+    # release build and as a fatal C assertion in a debug build, so the test
+    # can only run in the former.
+    @unittest.skipIf(support.Py_DEBUG, "gh-160001 aborts debug builds")
     @unittest.expectedFailure
     @hypothesis.given(
         data=data_strategy,

From 432ac6f11c60ea63544234c3b99bb1f5761f1416 Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Wed, 21 Oct 2026 17:31:18 -0400
Subject: [PATCH 10/12] Add a pinned test for max_length=0 calls with output
 pending

Fuzzing the zstd tests with Atheris found that ZstdDecompressor raises
ZstdError on the sixteenth consecutive decompress(b'', 0) call when a
stream is buffered and its output is pending: the C loop calls
ZSTD_decompressStream with an empty output buffer, and libzstd gives up
after sixteen calls that make no progress.  LZMADecompressor fails the
same shape on the fourth call, a variant of gh-160000.

test_repeated_max_length_zero only exercises a fresh decompressor, so
this adds a companion with a whole stream fed first.  It is an
expectedFailure for lzma and zstd; bz2 and zlib pass.
---
 Lib/test/test_compression_property.py | 46 +++++++++++++++++++++++++--
 1 file changed, 43 insertions(+), 3 deletions(-)

diff --git a/Lib/test/test_compression_property.py b/Lib/test/test_compression_property.py
index dc476b076e4c8a..744d5124838075 100644
--- a/Lib/test/test_compression_property.py
+++ b/Lib/test/test_compression_property.py
@@ -380,6 +380,26 @@ def _check_repeated_max_length_zero(self, data, count):
     def test_repeated_max_length_zero(self, data, count):
         self._check_repeated_max_length_zero(data, count)
 
+    def _check_max_length_zero_with_pending_output(self, data, count):
+        # The same, but with a whole stream buffered and its output pending.
+        decomp = self.backend.new_decompressor()
+        self.assertEqual(decomp.decompress(self.backend.compress(data), 0),
+                         b"")
+        for _ in range(count):
+            if decomp.eof:
+                break
+            self.assertEqual(decomp.decompress(b"", 0), b"")
+        if not decomp.eof:
+            self.assertEqual(decomp.decompress(b"", -1), data)
+        self.assertTrue(decomp.eof)
+
+    @hypothesis.given(data=data_strategy, count=st.integers(1, 20))
+    @hypothesis.example(data=b"", count=1)
+    @hypothesis.example(data=b"\x00", count=16)
+    @hypothesis.example(data=b"spam" * 10, count=20)
+    def test_max_length_zero_with_pending_output(self, data, count):
+        self._check_max_length_zero_with_pending_output(data, count)
+
     @hypothesis.given(data=data_strategy, cut=st.integers(0, 10**6))
     @hypothesis.example(data=b"", cut=0)
     @hypothesis.example(data=b"", cut=10**6)
@@ -502,8 +522,10 @@ def test_needs_input_means_drained(self, data, sizes, max_lengths):
 # liblzma reports an error when two consecutive calls can make no progress,
 # which lzma.LZMADecompressor lets through as LZMAError("Insufficient buffer
 # space") for two max_length=0 calls in a row with nothing buffered
-# (gh-160000).  The broad lzma tests avoid max_length=0 so that they keep
-# checking everything else; test_repeated_max_length_zero pins the bug.
+# (gh-160000), and on the fourth such call when output is pending.  The
+# broad lzma tests avoid max_length=0 so that they keep checking everything
+# else; test_repeated_max_length_zero and
+# test_max_length_zero_with_pending_output pin the bug.
 lzma_max_length_strategy = st.integers(1, 64) | st.just(-1)
 lzma_max_lengths_strategy = st.lists(lzma_max_length_strategy, max_size=16)
 
@@ -542,6 +564,23 @@ def test_needs_input_means_drained(self, data, sizes, max_lengths):
     def test_repeated_max_length_zero(self, data, count):
         self._check_repeated_max_length_zero(data, count)
 
+    @unittest.expectedFailure
+    @hypothesis.given(data=data_strategy, count=st.integers(1, 20))
+    @hypothesis.example(data=b"\x00", count=16)
+    def test_max_length_zero_with_pending_output(self, data, count):
+        self._check_max_length_zero_with_pending_output(data, count)
+
+
+class _ZstdMaxLengthZeroBug:
+    # libzstd refuses the 16th consecutive call that makes no progress, and
+    # ZstdDecompressor calls it even for max_length=0, so sixteen such calls
+    # with output pending raise ZstdError.
+    @unittest.expectedFailure
+    @hypothesis.given(data=data_strategy, count=st.integers(1, 20))
+    @hypothesis.example(data=b"\x00", count=16)
+    def test_max_length_zero_with_pending_output(self, data, count):
+        self._check_max_length_zero_with_pending_output(data, count)
+
 
 class LZMADecompressorMachine(_DecompressorMachine):
     # See lzma_max_length_strategy above.
@@ -581,7 +620,8 @@ class RawDeflateDecompressorTests(_ZlibNeedsInputBug, _DecompressorTests,
 
 
 @support.requires_zstd()
-class ZstdDecompressorTests(_DecompressorTests, unittest.TestCase):
+class ZstdDecompressorTests(_ZstdMaxLengthZeroBug, _DecompressorTests,
+                            unittest.TestCase):
     backend = _zstd_backend()
 
 

From 1c86a3e990374e7f931cc4f23206395c70940ea8 Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Wed, 21 Oct 2026 17:44:58 -0400
Subject: [PATCH 11/12] Keep FDICT zlib headers out of the invalid-input test
 and pin them

Fuzzing the zlib tests with Atheris produced a zlib header with the
FDICT flag set within seconds, and zlib._ZlibDecompressor turns that
into SystemError ("error return without exception set") because the
Z_NEED_DICT branch returns NULL without an exception.  That is the same
defect as gh-160001, reached without any dictionary at all.  Random
invalid input hits it often enough to make test_invalid_input flaky for
the zlib wrapper backend, and in a debug build the missing exception is
a fatal C assertion rather than a failure.

The invalid-input check now asks the class whether to skip a blob, and
the zlib wrapper companion skips valid headers with FDICT set.  The
carve-out is pinned by test_fdict_header, which expects zlib.error as
zlib.decompressobj raises for the same input, and which like
test_zlib_wrapper can only run in a release build.
---
 Lib/test/test_compression_property.py | 45 +++++++++++++++++++++++++--
 1 file changed, 43 insertions(+), 2 deletions(-)

diff --git a/Lib/test/test_compression_property.py b/Lib/test/test_compression_property.py
index 744d5124838075..a71de7c4624909 100644
--- a/Lib/test/test_compression_property.py
+++ b/Lib/test/test_compression_property.py
@@ -451,6 +451,15 @@ def test_chunked_compress_round_trip(self, data, sizes):
                         sizes=[], max_lengths=[])
     def test_invalid_input(self, garbage, data, index, byte, sizes,
                            max_lengths):
+        self._check_invalid_input(garbage, data, index, byte, sizes,
+                                  max_lengths)
+
+    def _skip_invalid_blob(self, blob):
+        # Hook for a known bug: a mixin may rule out inputs that crash.
+        return False
+
+    def _check_invalid_input(self, garbage, data, index, byte, sizes,
+                             max_lengths):
         stream = self.backend.compress(data)
         index %= len(stream) + 1
         if index == len(stream):
@@ -458,6 +467,8 @@ def test_invalid_input(self, garbage, data, index, byte, sizes,
         else:
             corrupted = stream[:index] + bytes([byte]) + stream[index + 1:]
         for blob in (garbage, corrupted):
+            if self._skip_invalid_blob(blob):
+                continue
             with self.subTest(blob=blob):
                 decomp = self.backend.new_decompressor()
                 try:
@@ -519,6 +530,36 @@ def test_needs_input_means_drained(self, data, sizes, max_lengths):
         self._check_needs_input_means_drained(data, sizes, max_lengths)
 
 
+def _zlib_fdict_header(blob):
+    # A valid zlib header whose FDICT flag is set.
+    return (len(blob) >= 2 and blob[0] & 0x0f == 8 and blob[0] >> 4 <= 7
+            and int.from_bytes(blob[:2]) % 31 == 0 and bool(blob[1] & 0x20))
+
+
+class _ZlibNeedsDictBug:
+    # zlib._ZlibDecompressor returns NULL without an exception when inflate
+    # asks for a preset dictionary (gh-160001).  Any zlib-wrapped input whose
+    # header has FDICT set gets there, so random invalid input would hit it,
+    # and in a debug build the missing exception is a fatal C assertion.
+    # test_invalid_input leaves those inputs out; test_fdict_header pins the
+    # bug and can only run in a release build.
+    def _skip_invalid_blob(self, blob):
+        return _zlib_fdict_header(blob)
+
+    @unittest.skipIf(support.Py_DEBUG, "gh-160001 aborts debug builds")
+    @unittest.expectedFailure
+    @hypothesis.given(data=st.binary(max_size=64))
+    @hypothesis.example(data=b"")
+    @hypothesis.example(data=b"\x00\x00\x00\x01\x03\x00")
+    def test_fdict_header(self, data):
+        # zlib.decompressobj raises zlib.error for the same input.
+        stream = b"\x78\xbb" + data
+        with self.assertRaises(zlib.error):
+            zlib.decompressobj().decompress(stream)
+        with self.assertRaises(zlib.error):
+            zlib._ZlibDecompressor().decompress(stream)
+
+
 # liblzma reports an error when two consecutive calls can make no progress,
 # which lzma.LZMADecompressor lets through as LZMAError("Insufficient buffer
 # space") for two max_length=0 calls in a row with nothing buffered
@@ -608,8 +649,8 @@ class LZMADecompressorTests(_LZMAMaxLengthZeroBug, _DecompressorTests,
 
 
 @support.requires_zlib()
-class ZlibDecompressorTests(_ZlibNeedsInputBug, _DecompressorTests,
-                            unittest.TestCase):
+class ZlibDecompressorTests(_ZlibNeedsInputBug, _ZlibNeedsDictBug,
+                            _DecompressorTests, unittest.TestCase):
     backend = _zlib_backend(15)
 
 

From c197707d4523a1eb19550b621878aef016b1b5a1 Mon Sep 17 00:00:00 2001
From: Paul Ganssle <git@m.ganssle.io>
Date: Thu, 22 Oct 2026 20:16:53 -0400
Subject: [PATCH 12/12] Pin the known compression bugs as expected-failure
 examples

Previously each known bug replaced the affected test for its backend
with an expectedFailure version, so the test as a whole was expected to
fail and nothing checked that the failure was the known one. With
example.xfail() in the stubs (gh-160356), the known inputs are now
expected-failure examples on the affected test: the generated input
leaves the failing region out, as the comments say, and each example
must fail, so the test fails again once the bug is fixed.

For the needs_input bug (gh-159999) the generated calls for bz2 and
zlib are unlimited, which is the case they get right; the examples
depend on where the compressor puts block boundaries, so each has been
checked to fail with both zlib 1.3.2 and zlib-ng 1.3.1, since the
expected failures are strict. The FDICT examples (gh-160001) go on the
zlib wrapper's test_invalid_input, which now clears the flag in generated
garbage rather than skipping the input, and they are only registered in
release builds, where the missing exception is a SystemError rather
than a fatal assertion. The max_length=0 tests for lzma and zstd stop
short of the failing call (gh-160000, gh-160121).

Hypothesis also found that the broad zstd tests could reach gh-160121
with sixteen zero-length calls in a row, so zstd now shares the lzma
carve-out of max_length=0, in the companions and the state machine.

The subTest calls inside the invalid-input and zdict checks are gone:
Hypothesis disables subTest under @given anyway, and inside the stub
replay they swallowed the exception an expected-failure example has to
raise.
---
 Lib/test/test_compression_property.py | 398 ++++++++++++++------------
 1 file changed, 214 insertions(+), 184 deletions(-)

diff --git a/Lib/test/test_compression_property.py b/Lib/test/test_compression_property.py
index a71de7c4624909..78bae6d1661bd8 100644
--- a/Lib/test/test_compression_property.py
+++ b/Lib/test/test_compression_property.py
@@ -53,6 +53,11 @@
 st = hypothesis.strategies
 stateful = hypothesis.stateful
 
+# The expected-failure examples below name these at class definition
+# time, whether or not the module is available.
+LZMAError = lzma.LZMAError if lzma is not None else Exception
+ZstdError = zstd.ZstdError if zstd is not None else Exception
+
 
 def _repeat(pair):
     chunk, count = pair
@@ -290,6 +295,37 @@ def output_is_a_prefix(self):
             self.assertEqual(self.out, self.data)
 
 
+def _invalid_input_test(*examples, garbage=st.binary(max_size=256)):
+    # Invalid input: random bytes, or a valid stream with one byte replaced
+    # or with its end cut off.  The only acceptable outcomes are success or
+    # the backend's own error type.  Built by a function so that a backend
+    # with a known bug can narrow the garbage and add its own examples.
+    @hypothesis.given(
+        garbage=garbage,
+        data=data_strategy,
+        index=st.integers(0, 10**6),
+        byte=st.integers(0, 255),
+        sizes=chunk_sizes_strategy,
+        max_lengths=max_lengths_strategy,
+    )
+    @hypothesis.example(garbage=b"", data=b"", index=0, byte=0, sizes=[],
+                        max_lengths=[])
+    @hypothesis.example(garbage=b"\xff" * 20, data=b"", index=0, byte=0,
+                        sizes=[1], max_lengths=[0])
+    @hypothesis.example(garbage=b"", data=b"abc" * 30, index=10**6, byte=0,
+                        sizes=[3], max_lengths=[1, -1])
+    @hypothesis.example(garbage=b"", data=b"abc" * 30, index=5, byte=255,
+                        sizes=[], max_lengths=[])
+    def test_invalid_input(self, garbage, data, index, byte, sizes,
+                           max_lengths):
+        self._check_invalid_input(garbage, data, index, byte, sizes,
+                                  max_lengths)
+
+    for example in examples:
+        test_invalid_input = example(test_invalid_input)
+    return test_invalid_input
+
+
 class _DecompressorTests:
     """Deterministic companions to the state machine, replayable by the stub.
 
@@ -430,32 +466,11 @@ def test_chunked_compress_round_trip(self, data, sizes):
         self.assertTrue(decomp.eof)
         self.assertEqual(decomp.unused_data, b"")
 
-    # Invalid input: random bytes, or a valid stream with one byte replaced
-    # or with its end cut off.  The only acceptable outcomes are success or
-    # the backend's own error type.
-    @hypothesis.given(
-        garbage=st.binary(max_size=256),
-        data=data_strategy,
-        index=st.integers(0, 10**6),
-        byte=st.integers(0, 255),
-        sizes=chunk_sizes_strategy,
-        max_lengths=max_lengths_strategy,
-    )
-    @hypothesis.example(garbage=b"", data=b"", index=0, byte=0, sizes=[],
-                        max_lengths=[])
-    @hypothesis.example(garbage=b"\xff" * 20, data=b"", index=0, byte=0,
-                        sizes=[1], max_lengths=[0])
-    @hypothesis.example(garbage=b"", data=b"abc" * 30, index=10**6, byte=0,
-                        sizes=[3], max_lengths=[1, -1])
-    @hypothesis.example(garbage=b"", data=b"abc" * 30, index=5, byte=255,
-                        sizes=[], max_lengths=[])
-    def test_invalid_input(self, garbage, data, index, byte, sizes,
-                           max_lengths):
-        self._check_invalid_input(garbage, data, index, byte, sizes,
-                                  max_lengths)
+    test_invalid_input = _invalid_input_test()
 
-    def _skip_invalid_blob(self, blob):
-        # Hook for a known bug: a mixin may rule out inputs that crash.
+    def _skip_corrupted_blob(self, blob):
+        # Hook for a known bug: a backend may rule out a corruption that
+        # crashes.
         return False
 
     def _check_invalid_input(self, garbage, data, index, byte, sizes,
@@ -466,68 +481,53 @@ def _check_invalid_input(self, garbage, data, index, byte, sizes,
             corrupted = stream[:byte % (len(stream) + 1)]
         else:
             corrupted = stream[:index] + bytes([byte]) + stream[index + 1:]
-        for blob in (garbage, corrupted):
-            if self._skip_invalid_blob(blob):
-                continue
-            with self.subTest(blob=blob):
-                decomp = self.backend.new_decompressor()
-                try:
-                    for chunk, max_length in _chunks_with(blob, sizes,
-                                                          max_lengths, -1):
-                        decomp.decompress(chunk, max_length)
-                except (self.backend.error, EOFError):
-                    pass
-                try:
-                    self.backend.decompress(blob)
-                except self.backend.decompress_errors:
-                    pass
-
-
-# Known bugs, as mixins that override the affected test for one backend.
-# They go first in the bases so that they win over _DecompressorTests.
-
-class _BZ2NeedsInputBug:
-    # bz2 reports needs_input=True as soon as its input buffer is empty, even
-    # when max_length cut the output short and more of it is pending
-    # (gh-159999).
-    @unittest.expectedFailure
-    @hypothesis.given(
-        data=data_strategy,
-        sizes=chunk_sizes_strategy,
-        max_lengths=max_lengths_strategy,
-    )
-    @hypothesis.example(data=b"\x00\x10 0@" * 2, sizes=[40], max_lengths=[0])
+        blobs = [garbage]
+        if not self._skip_corrupted_blob(corrupted):
+            blobs.append(corrupted)
+        for blob in blobs:
+            decomp = self.backend.new_decompressor()
+            try:
+                for chunk, max_length in _chunks_with(blob, sizes,
+                                                      max_lengths, -1):
+                    decomp.decompress(chunk, max_length)
+            except (self.backend.error, EOFError):
+                pass
+            try:
+                self.backend.decompress(blob)
+            except self.backend.decompress_errors:
+                pass
+
+
+# Known bugs.  A backend with one gets its own version of the affected
+# test: the generated input leaves out what is known to fail, and the
+# expected-failure examples pin it, so that the test fails again once the
+# bug is fixed and the carve-out can go.
+
+def _needs_input_test(*examples):
+    # test_needs_input_means_drained for bz2 and zlib._ZlibDecompressor,
+    # which report needs_input=True as soon as their input buffer is empty,
+    # even when max_length cut the output short and more of it is pending
+    # (gh-159999).  Until that is fixed every generated call is unlimited,
+    # the case they get right, and the examples pin the rest.  The zlib
+    # examples depend on where the compressor puts block boundaries, so
+    # each has been checked to fail with both zlib and zlib-ng.
+    @hypothesis.given(data=data_strategy, sizes=chunk_sizes_strategy,
+                      max_lengths=st.just([]))
+    @hypothesis.example(data=b"", sizes=[], max_lengths=[])
+    @hypothesis.example(data=b"abc" * 50, sizes=[10] * 10, max_lengths=[])
     def test_needs_input_means_drained(self, data, sizes, max_lengths):
         self._check_needs_input_means_drained(data, sizes, max_lengths)
 
+    for example in examples:
+        test_needs_input_means_drained = example(
+            test_needs_input_means_drained)
+    return test_needs_input_means_drained
 
-class _ZlibNeedsInputBug:
-    # zlib._ZlibDecompressor has the same needs_input problem as bz2
-    # (gh-159999).  The examples depend on where the compressor puts block
-    # boundaries, so there are several: at least one must fail for each
-    # wbits variant (zlib wrapper, raw deflate, gzip wrapper) with both zlib
-    # and zlib-ng.
-    @unittest.expectedFailure
-    @hypothesis.given(
-        data=data_strategy,
-        sizes=chunk_sizes_strategy,
-        max_lengths=max_lengths_strategy,
-    )
-    @hypothesis.example(data=b"\x00", sizes=[4], max_lengths=[0])
-    @hypothesis.example(data=b"\x00" * 21, sizes=[], max_lengths=[2])
-    @hypothesis.example(
-        data=b"\x00\x00\x00\x01\x00\x00\x02\x00\x00\x03\x00\x00"
-             b"\x90\x00\x90\x90\x90\x91\x90" * 2,
-        sizes=[34],
-        max_lengths=[20],
-    )
-    @hypothesis.example(
-        data=b"\x00\x00\x00\x00\x01\x00\x00\x00\x02\x00\x00\x03" * 2,
-        sizes=[25],
-        max_lengths=[12],
-    )
-    def test_needs_input_means_drained(self, data, sizes, max_lengths):
-        self._check_needs_input_means_drained(data, sizes, max_lengths)
+
+def _needs_input_example(data, sizes, max_lengths):
+    return hypothesis.example(
+        data=data, sizes=sizes, max_lengths=max_lengths,
+    ).xfail(reason="gh-159999", raises=AssertionError)
 
 
 def _zlib_fdict_header(blob):
@@ -536,47 +536,52 @@ def _zlib_fdict_header(blob):
             and int.from_bytes(blob[:2]) % 31 == 0 and bool(blob[1] & 0x20))
 
 
-class _ZlibNeedsDictBug:
-    # zlib._ZlibDecompressor returns NULL without an exception when inflate
-    # asks for a preset dictionary (gh-160001).  Any zlib-wrapped input whose
-    # header has FDICT set gets there, so random invalid input would hit it,
-    # and in a debug build the missing exception is a fatal C assertion.
-    # test_invalid_input leaves those inputs out; test_fdict_header pins the
-    # bug and can only run in a release build.
-    def _skip_invalid_blob(self, blob):
-        return _zlib_fdict_header(blob)
+def _without_fdict(blob):
+    # Clear the FDICT flag of a zlib header, keeping its check bits valid.
+    if not _zlib_fdict_header(blob):
+        return blob
+    flg = blob[1] & 0xc0
+    flg |= (31 - (blob[0] * 256 + flg) % 31) % 31
+    return bytes([blob[0], flg]) + blob[2:]
+
 
-    @unittest.skipIf(support.Py_DEBUG, "gh-160001 aborts debug builds")
-    @unittest.expectedFailure
-    @hypothesis.given(data=st.binary(max_size=64))
-    @hypothesis.example(data=b"")
-    @hypothesis.example(data=b"\x00\x00\x00\x01\x03\x00")
-    def test_fdict_header(self, data):
-        # zlib.decompressobj raises zlib.error for the same input.
-        stream = b"\x78\xbb" + data
-        with self.assertRaises(zlib.error):
-            zlib.decompressobj().decompress(stream)
-        with self.assertRaises(zlib.error):
-            zlib._ZlibDecompressor().decompress(stream)
-
-
-# liblzma reports an error when two consecutive calls can make no progress,
-# which lzma.LZMADecompressor lets through as LZMAError("Insufficient buffer
-# space") for two max_length=0 calls in a row with nothing buffered
-# (gh-160000), and on the fourth such call when output is pending.  The
-# broad lzma tests avoid max_length=0 so that they keep checking everything
-# else; test_repeated_max_length_zero and
-# test_max_length_zero_with_pending_output pin the bug.
-lzma_max_length_strategy = st.integers(1, 64) | st.just(-1)
-lzma_max_lengths_strategy = st.lists(lzma_max_length_strategy, max_size=16)
-
-
-class _LZMAMaxLengthZeroBug:
+def _unless_debug_build(example):
+    # gh-160001 is a fatal C assertion in a debug build, so its examples
+    # can only be registered in a release build.
+    if support.Py_DEBUG:
+        return lambda test: test
+    return example
+
+
+def _fdict_example(garbage):
+    # zlib._ZlibDecompressor returns NULL without an exception when inflate
+    # asks for a preset dictionary (gh-160001), which any zlib header with
+    # FDICT set followed by a dictionary id does; SystemError in a release
+    # build.  zlib.decompressobj raises zlib.error for the same input.
+    return _unless_debug_build(hypothesis.example(
+        garbage=garbage, data=b"", index=0, byte=0, sizes=[], max_lengths=[],
+    ).xfail(reason="gh-160001", raises=SystemError))
+
+
+# liblzma and libzstd report an error when consecutive calls make no
+# progress, which the decompressors let through for max_length=0 calls in
+# a row: lzma.LZMADecompressor on the second with nothing buffered
+# (gh-160000) and on the fourth with output pending, ZstdDecompressor on
+# the sixteenth with output pending (gh-160121).  The broad tests for
+# those two backends avoid max_length=0 so that they keep checking
+# everything else, the tests about it stop short of the failing call, and
+# the expected-failure examples pin it.
+nonzero_max_length_strategy = st.integers(1, 64) | st.just(-1)
+nonzero_max_lengths_strategy = st.lists(nonzero_max_length_strategy,
+                                        max_size=16)
+
+
+class _NoMaxLengthZeroTests:
     @hypothesis.given(
         data=data_strategy,
         trailer=trailer_strategy,
         sizes=chunk_sizes_strategy,
-        max_lengths=lzma_max_lengths_strategy,
+        max_lengths=nonzero_max_lengths_strategy,
     )
     @hypothesis.example(data=b"", trailer=b"", sizes=[], max_lengths=[])
     @hypothesis.example(data=b"", trailer=b"tail", sizes=[1, 1, 1],
@@ -591,7 +596,7 @@ def test_chunked_decompress(self, data, trailer, sizes, max_lengths):
     @hypothesis.given(
         data=data_strategy,
         sizes=chunk_sizes_strategy,
-        max_lengths=lzma_max_lengths_strategy,
+        max_lengths=nonzero_max_lengths_strategy,
     )
     @hypothesis.example(data=b"", sizes=[], max_lengths=[])
     @hypothesis.example(data=b"abc" * 50, sizes=[10] * 10,
@@ -599,69 +604,101 @@ def test_chunked_decompress(self, data, trailer, sizes, max_lengths):
     def test_needs_input_means_drained(self, data, sizes, max_lengths):
         self._check_needs_input_means_drained(data, sizes, max_lengths)
 
-    @unittest.expectedFailure
-    @hypothesis.given(data=data_strategy, count=st.integers(1, 4))
-    @hypothesis.example(data=b"", count=2)
+
+class _LZMAMaxLengthZeroTests(_NoMaxLengthZeroTests):
+    @hypothesis.given(data=data_strategy, count=st.just(1))
+    @hypothesis.example(data=b"", count=1)
+    @hypothesis.example(data=b"spam" * 10, count=1)
+    @hypothesis.example(data=b"", count=2).xfail(
+        reason="gh-160000", raises=LZMAError)
+    @hypothesis.example(data=b"spam" * 10, count=3).xfail(
+        reason="gh-160000", raises=LZMAError)
     def test_repeated_max_length_zero(self, data, count):
         self._check_repeated_max_length_zero(data, count)
 
-    @unittest.expectedFailure
-    @hypothesis.given(data=data_strategy, count=st.integers(1, 20))
-    @hypothesis.example(data=b"\x00", count=16)
+    @hypothesis.given(data=data_strategy, count=st.integers(1, 3))
+    @hypothesis.example(data=b"", count=1)
+    @hypothesis.example(data=b"\x00", count=3)
+    @hypothesis.example(data=b"\x00", count=4).xfail(
+        reason="gh-160121", raises=LZMAError)
+    @hypothesis.example(data=b"spam" * 10, count=20).xfail(
+        reason="gh-160121", raises=LZMAError)
     def test_max_length_zero_with_pending_output(self, data, count):
         self._check_max_length_zero_with_pending_output(data, count)
 
 
-class _ZstdMaxLengthZeroBug:
-    # libzstd refuses the 16th consecutive call that makes no progress, and
-    # ZstdDecompressor calls it even for max_length=0, so sixteen such calls
-    # with output pending raise ZstdError.
-    @unittest.expectedFailure
-    @hypothesis.given(data=data_strategy, count=st.integers(1, 20))
-    @hypothesis.example(data=b"\x00", count=16)
+class _ZstdMaxLengthZeroTests(_NoMaxLengthZeroTests):
+    @hypothesis.given(data=data_strategy, count=st.integers(1, 15))
+    @hypothesis.example(data=b"", count=1)
+    @hypothesis.example(data=b"\x00", count=15)
+    @hypothesis.example(data=b"\x00", count=16).xfail(
+        reason="gh-160121", raises=ZstdError)
+    @hypothesis.example(data=b"spam" * 10, count=20).xfail(
+        reason="gh-160121", raises=ZstdError)
     def test_max_length_zero_with_pending_output(self, data, count):
         self._check_max_length_zero_with_pending_output(data, count)
 
 
-class LZMADecompressorMachine(_DecompressorMachine):
-    # See lzma_max_length_strategy above.
-    @stateful.rule(n=st.integers(0, 64), max_length=lzma_max_length_strategy)
+class NoMaxLengthZeroDecompressorMachine(_DecompressorMachine):
+    # See nonzero_max_length_strategy above.
+    @stateful.rule(n=st.integers(0, 64),
+                   max_length=nonzero_max_length_strategy)
     def feed(self, n, max_length):
         self._feed_next(n, max_length)
 
 
 @support.requires_bz2()
-class BZ2DecompressorTests(_BZ2NeedsInputBug, _DecompressorTests,
-                           unittest.TestCase):
+class BZ2DecompressorTests(_DecompressorTests, unittest.TestCase):
     backend = _bz2_backend()
+    test_needs_input_means_drained = _needs_input_test(
+        _needs_input_example(b"\x00\x10 0@" * 2, [40], [0]))
 
 
 @support.requires_gzip()
-class GzipDecompressorTests(_ZlibNeedsInputBug, _DecompressorTests,
-                            unittest.TestCase):
+class GzipDecompressorTests(_DecompressorTests, unittest.TestCase):
     backend = _gzip_backend()
+    # The gzip header takes ten bytes, so a chunk of twelve leaves two
+    # bytes of deflate data.
+    test_needs_input_means_drained = _needs_input_test(
+        _needs_input_example(b"\x00", [12], [0]),
+        _needs_input_example(b"ab" * 40, [13], [1]))
 
 
 @support.requires_lzma()
-class LZMADecompressorTests(_LZMAMaxLengthZeroBug, _DecompressorTests,
+class LZMADecompressorTests(_LZMAMaxLengthZeroTests, _DecompressorTests,
                             unittest.TestCase):
     backend = _lzma_backend()
 
 
 @support.requires_zlib()
-class ZlibDecompressorTests(_ZlibNeedsInputBug, _ZlibNeedsDictBug,
-                            _DecompressorTests, unittest.TestCase):
+class ZlibDecompressorTests(_DecompressorTests, unittest.TestCase):
     backend = _zlib_backend(15)
+    test_needs_input_means_drained = _needs_input_test(
+        _needs_input_example(b"\x00", [4], [0]),
+        _needs_input_example(b"ab" * 40, [5], [1]))
+
+    # Random invalid input would hit gh-160001 (see _fdict_example) with
+    # any FDICT header, so generated garbage has the flag cleared, a
+    # flipped byte that sets it is left out, and the examples pin the bug.
+    def _skip_corrupted_blob(self, blob):
+        return _zlib_fdict_header(blob)
+
+    test_invalid_input = _invalid_input_test(
+        _fdict_example(b"\x78\xbb\x00\x00\x00\x01\x03\x00"),
+        _fdict_example(b"\x78\xbb\xff\xff\xff\xff"),
+        garbage=st.binary(max_size=256).map(_without_fdict))
 
 
 @support.requires_zlib()
-class RawDeflateDecompressorTests(_ZlibNeedsInputBug, _DecompressorTests,
-                                  unittest.TestCase):
+class RawDeflateDecompressorTests(_DecompressorTests, unittest.TestCase):
     backend = _zlib_backend(-15)
+    test_needs_input_means_drained = _needs_input_test(
+        _needs_input_example(b"\x00", [2], [0]),
+        _needs_input_example(b"\x00" * 21, [1], [2]))
 
 
 @support.requires_zstd()
-class ZstdDecompressorTests(_ZstdMaxLengthZeroBug, _DecompressorTests,
+class ZstdDecompressorTests(_ZstdMaxLengthZeroTests, _DecompressorTests,
                             unittest.TestCase):
     backend = _zstd_backend()
 
@@ -673,8 +710,8 @@ class ZstdDecompressorTests(_ZstdMaxLengthZeroBug, _DecompressorTests,
     "GzipDecompressorMachine", _DecompressorMachine, _gzip_backend,
     support.requires_gzip())
 LZMADecompressorMachineTest = _machine_test(
-    "LZMADecompressorMachine", LZMADecompressorMachine, _lzma_backend,
-    support.requires_lzma())
+    "LZMADecompressorMachine", NoMaxLengthZeroDecompressorMachine,
+    _lzma_backend, support.requires_lzma())
 ZlibDecompressorMachineTest = _machine_test(
     "ZlibDecompressorMachine", _DecompressorMachine,
     functools.partial(_zlib_backend, 15), support.requires_zlib())
@@ -682,8 +719,8 @@ class ZstdDecompressorTests(_ZstdMaxLengthZeroBug, _DecompressorTests,
     "RawDeflateDecompressorMachine", _DecompressorMachine,
     functools.partial(_zlib_backend, -15), support.requires_zlib())
 ZstdDecompressorMachineTest = _machine_test(
-    "ZstdDecompressorMachine", _DecompressorMachine, _zstd_backend,
-    support.requires_zstd())
+    "ZstdDecompressorMachine", NoMaxLengthZeroDecompressorMachine,
+    _zstd_backend, support.requires_zstd())
 
 
 # ---------------------------------------------------------------------------
@@ -844,51 +881,44 @@ def test_flush(self, data, cut):
                         level=9)
     def test_zdict_round_trip(self, data, zdict, level):
         for wbits in (15, -15):
-            with self.subTest(wbits=wbits):
-                comp = zlib.compressobj(level, wbits=wbits, zdict=zdict)
-                stream = comp.compress(data) + comp.flush()
-                decomp = zlib.decompressobj(wbits, zdict=zdict)
-                self.assertEqual(decomp.decompress(stream), data)
-                self.assertTrue(decomp.eof)
+            comp = zlib.compressobj(level, wbits=wbits, zdict=zdict)
+            stream = comp.compress(data) + comp.flush()
+            decomp = zlib.decompressobj(wbits, zdict=zdict)
+            self.assertEqual(decomp.decompress(stream), data)
+            self.assertTrue(decomp.eof)
 
 
-@support.requires_zlib()
-class ZlibDecompressorZdictTests(unittest.TestCase):
-    def _check_round_trip(self, data, zdict, level, wbits):
-        comp = zlib.compressobj(level, wbits=wbits, zdict=zdict)
-        stream = comp.compress(data) + comp.flush()
-        decomp = zlib._ZlibDecompressor(wbits, zdict=zdict)
-        self.assertEqual(decomp.decompress(stream), data)
-        self.assertTrue(decomp.eof)
+def _zdict_example(data, zdict, level, wbits):
+    example = hypothesis.example(data=data, zdict=zdict, level=level,
+                                 wbits=wbits)
+    if wbits > 0:
+        return _unless_debug_build(
+            example.xfail(reason="gh-160001", raises=SystemError))
+    return example
 
-    @hypothesis.given(
-        data=data_strategy,
-        zdict=st.binary(min_size=1, max_size=256),
-        level=st.integers(0, 9),
-    )
-    @hypothesis.example(data=b"", zdict=b"\x00", level=0)
-    @hypothesis.example(data=b"hello world" * 10, zdict=b"hello world",
-                        level=9)
-    def test_raw_deflate(self, data, zdict, level):
-        self._check_round_trip(data, zdict, level, -15)
 
+@support.requires_zlib()
+class ZlibDecompressorZdictTests(unittest.TestCase):
     # _ZlibDecompressor only applies zdict up front for raw deflate; for a
     # zlib-wrapped stream it hits Z_NEED_DICT and returns NULL without
-    # setting an exception (gh-160001).  That surfaces as SystemError in a
-    # release build and as a fatal C assertion in a debug build, so the test
-    # can only run in the former.
-    @unittest.skipIf(support.Py_DEBUG, "gh-160001 aborts debug builds")
-    @unittest.expectedFailure
+    # setting an exception (gh-160001).  Only raw deflate is generated;
+    # the expected-failure examples pin the wrapper.
     @hypothesis.given(
         data=data_strategy,
         zdict=st.binary(min_size=1, max_size=256),
         level=st.integers(0, 9),
+        wbits=st.just(-15),
     )
-    @hypothesis.example(data=b"", zdict=b"\x00", level=0)
-    @hypothesis.example(data=b"hello world" * 10, zdict=b"hello world",
-                        level=9)
-    def test_zlib_wrapper(self, data, zdict, level):
-        self._check_round_trip(data, zdict, level, 15)
+    @_zdict_example(b"", b"\x00", 0, -15)
+    @_zdict_example(b"hello world" * 10, b"hello world", 9, -15)
+    @_zdict_example(b"", b"\x00", 0, 15)
+    @_zdict_example(b"hello world" * 10, b"hello world", 9, 15)
+    def test_round_trip(self, data, zdict, level, wbits):
+        comp = zlib.compressobj(level, wbits=wbits, zdict=zdict)
+        stream = comp.compress(data) + comp.flush()
+        decomp = zlib._ZlibDecompressor(wbits, zdict=zdict)
+        self.assertEqual(decomp.decompress(stream), data)
+        self.assertTrue(decomp.eof)
 
 
 # ---------------------------------------------------------------------------