From eaf2d972633ea30ac764edf618251bc2ecc5ad19 Mon Sep 17 00:00:00 2001 From: ptweezy Date: Tue, 27 Oct 2026 16:09:31 -0400 Subject: [PATCH 1/2] Make the input buffer tests of IgzipDecompressor reach the input buffer isal_inflate decompresses up to about 64 KiB ahead into its own buffer, so it read the 100 to 300 byte inputs of these tests in full, and the input buffer was never used. Use 512 KiB of random data and multiply the offsets by 1,000. --- tests/test_igzip_lib.py | 37 +++++++++++++++++++++++-------------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/tests/test_igzip_lib.py b/tests/test_igzip_lib.py index a336e7f7..6dbb7463 100644 --- a/tests/test_igzip_lib.py +++ b/tests/test_igzip_lib.py @@ -12,6 +12,7 @@ import itertools import os import pickle +import random import subprocess import sys import textwrap @@ -72,6 +73,12 @@ class TestIgzipDecompressor(): TEXT = DATA DATA = igzip_lib.compress(DATA) BAD_DATA = b"Not a valid deflate block" + # isal_inflate decompresses up to about 64 KiB ahead into its own buffer + # and reads the input it needs for that. For the input buffer tests to + # leave input over, they use inputs of 100 KB or more of incompressible + # data, which needs about as much input as output. + RANDOM_TEXT = random.Random(0).randbytes(512 * 1024) + RANDOM_DATA = igzip_lib.compress(RANDOM_TEXT) def test_decompress(self): decomp = IgzipDecompressor() @@ -168,19 +175,20 @@ def test_decompressor_inputbuf_1(self): out = [] # Create input buffer and fill it - assert igzd.decompress(self.DATA[:100], max_length=0) == b'' + assert igzd.decompress(self.RANDOM_DATA[:100_000], + max_length=0) == b'' # Retrieve some results, freeing capacity at beginning # of input buffer - out.append(igzd.decompress(b'', 2)) + out.append(igzd.decompress(b'', 80_000)) # Add more data that fits into input buffer after # moving existing data to beginning - out.append(igzd.decompress(self.DATA[100:105], 15)) + out.append(igzd.decompress(self.RANDOM_DATA[100_000:105_000], 15)) # Decompress rest of data - out.append(igzd.decompress(self.DATA[105:])) - assert b''.join(out) == self.TEXT + out.append(igzd.decompress(self.RANDOM_DATA[105_000:])) + assert b''.join(out) == self.RANDOM_TEXT def test_decompressor_inputbuf_2(self): # Test reusing input buffer by appending data at the @@ -189,18 +197,19 @@ def test_decompressor_inputbuf_2(self): out = [] # Create input buffer and empty it - assert igzd.decompress(self.DATA[:200], max_length=0) == b'' + assert igzd.decompress(self.RANDOM_DATA[:200_000], + max_length=0) == b'' out.append(igzd.decompress(b'')) # Fill buffer with new data - out.append(igzd.decompress(self.DATA[200:280], 2)) + out.append(igzd.decompress(self.RANDOM_DATA[200_000:280_000], 2)) # Append some more data, not enough to require resize - out.append(igzd.decompress(self.DATA[280:300], 2)) + out.append(igzd.decompress(self.RANDOM_DATA[280_000:300_000], 2)) # Decompress rest of data - out.append(igzd.decompress(self.DATA[300:])) - assert b''.join(out) == self.TEXT + out.append(igzd.decompress(self.RANDOM_DATA[300_000:])) + assert b''.join(out) == self.RANDOM_TEXT def test_decompressor_inputbuf_3(self): # Test reusing input buffer after extending it @@ -209,14 +218,14 @@ def test_decompressor_inputbuf_3(self): out = [] # Create almost full input buffer - out.append(igzd.decompress(self.DATA[:200], 5)) + out.append(igzd.decompress(self.RANDOM_DATA[:200_000], 5)) # Add even more data to it, requiring resize - out.append(igzd.decompress(self.DATA[200:300], 5)) + out.append(igzd.decompress(self.RANDOM_DATA[200_000:300_000], 5)) # Decompress rest of data - out.append(igzd.decompress(self.DATA[300:])) - assert b''.join(out) == self.TEXT + out.append(igzd.decompress(self.RANDOM_DATA[300_000:])) + assert b''.join(out) == self.RANDOM_TEXT def test_failure(self): igzd = IgzipDecompressor() From 5ae2de73bc5daba0919d62d8b9fb67fbcc1d35aa Mon Sep 17 00:00:00 2001 From: ptweezy Date: Wed, 28 Oct 2026 09:11:44 -0400 Subject: [PATCH 2/2] Rename the random data attributes of TestIgzipDecompressor The uncompressed data is random bytes, not text, so RANDOM_TEXT becomes RANDOM_DATA and the compressed data becomes COMPRESSED_RANDOM_DATA. --- tests/test_igzip_lib.py | 34 +++++++++++++++++++--------------- 1 file changed, 19 insertions(+), 15 deletions(-) diff --git a/tests/test_igzip_lib.py b/tests/test_igzip_lib.py index 6dbb7463..ad832503 100644 --- a/tests/test_igzip_lib.py +++ b/tests/test_igzip_lib.py @@ -77,8 +77,8 @@ class TestIgzipDecompressor(): # and reads the input it needs for that. For the input buffer tests to # leave input over, they use inputs of 100 KB or more of incompressible # data, which needs about as much input as output. - RANDOM_TEXT = random.Random(0).randbytes(512 * 1024) - RANDOM_DATA = igzip_lib.compress(RANDOM_TEXT) + RANDOM_DATA = random.Random(0).randbytes(512 * 1024) + COMPRESSED_RANDOM_DATA = igzip_lib.compress(RANDOM_DATA) def test_decompress(self): decomp = IgzipDecompressor() @@ -175,7 +175,7 @@ def test_decompressor_inputbuf_1(self): out = [] # Create input buffer and fill it - assert igzd.decompress(self.RANDOM_DATA[:100_000], + assert igzd.decompress(self.COMPRESSED_RANDOM_DATA[:100_000], max_length=0) == b'' # Retrieve some results, freeing capacity at beginning @@ -184,11 +184,12 @@ def test_decompressor_inputbuf_1(self): # Add more data that fits into input buffer after # moving existing data to beginning - out.append(igzd.decompress(self.RANDOM_DATA[100_000:105_000], 15)) + out.append(igzd.decompress( + self.COMPRESSED_RANDOM_DATA[100_000:105_000], 15)) # Decompress rest of data - out.append(igzd.decompress(self.RANDOM_DATA[105_000:])) - assert b''.join(out) == self.RANDOM_TEXT + out.append(igzd.decompress(self.COMPRESSED_RANDOM_DATA[105_000:])) + assert b''.join(out) == self.RANDOM_DATA def test_decompressor_inputbuf_2(self): # Test reusing input buffer by appending data at the @@ -197,19 +198,21 @@ def test_decompressor_inputbuf_2(self): out = [] # Create input buffer and empty it - assert igzd.decompress(self.RANDOM_DATA[:200_000], + assert igzd.decompress(self.COMPRESSED_RANDOM_DATA[:200_000], max_length=0) == b'' out.append(igzd.decompress(b'')) # Fill buffer with new data - out.append(igzd.decompress(self.RANDOM_DATA[200_000:280_000], 2)) + out.append(igzd.decompress( + self.COMPRESSED_RANDOM_DATA[200_000:280_000], 2)) # Append some more data, not enough to require resize - out.append(igzd.decompress(self.RANDOM_DATA[280_000:300_000], 2)) + out.append(igzd.decompress( + self.COMPRESSED_RANDOM_DATA[280_000:300_000], 2)) # Decompress rest of data - out.append(igzd.decompress(self.RANDOM_DATA[300_000:])) - assert b''.join(out) == self.RANDOM_TEXT + out.append(igzd.decompress(self.COMPRESSED_RANDOM_DATA[300_000:])) + assert b''.join(out) == self.RANDOM_DATA def test_decompressor_inputbuf_3(self): # Test reusing input buffer after extending it @@ -218,14 +221,15 @@ def test_decompressor_inputbuf_3(self): out = [] # Create almost full input buffer - out.append(igzd.decompress(self.RANDOM_DATA[:200_000], 5)) + out.append(igzd.decompress(self.COMPRESSED_RANDOM_DATA[:200_000], 5)) # Add even more data to it, requiring resize - out.append(igzd.decompress(self.RANDOM_DATA[200_000:300_000], 5)) + out.append(igzd.decompress( + self.COMPRESSED_RANDOM_DATA[200_000:300_000], 5)) # Decompress rest of data - out.append(igzd.decompress(self.RANDOM_DATA[300_000:])) - assert b''.join(out) == self.RANDOM_TEXT + out.append(igzd.decompress(self.COMPRESSED_RANDOM_DATA[300_000:])) + assert b''.join(out) == self.RANDOM_DATA def test_failure(self): igzd = IgzipDecompressor()