Merged revisions 80926 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk
........
r80926 | antoine.pitrou | 2010-05-07 18:50:34 +0200 (ven., 07 mai 2010) | 5 lines
Issue #8571: Fix an internal error when compressing or decompressing a
chunk larger than 1GB with the zlib module's compressor and decompressor
objects.
........
diff --git a/Lib/test/test_zlib.py b/Lib/test/test_zlib.py
index d1e1796..9d17c93 100644
--- a/Lib/test/test_zlib.py
+++ b/Lib/test/test_zlib.py
@@ -3,6 +3,7 @@
import zlib
import binascii
import random
+from test.test_support import precisionbigmemtest, _1G
class ChecksumTestCase(unittest.TestCase):
@@ -89,8 +90,39 @@
self.assertRaises(ValueError, zlib.decompressobj().flush, -1)
+class BaseCompressTestCase(object):
+ def check_big_compress_buffer(self, size, compress_func):
+ _1M = 1024 * 1024
+ fmt = "%%0%dx" % (2 * _1M)
+ # Generate 10MB worth of random, and expand it by repeating it.
+ # The assumption is that zlib's memory is not big enough to exploit
+ # such spread out redundancy.
+ data = ''.join([binascii.a2b_hex(fmt % random.getrandbits(8 * _1M))
+ for i in range(10)])
+ data = data * (size // len(data) + 1)
+ try:
+ compress_func(data)
+ finally:
+ # Release memory
+ data = None
-class CompressTestCase(unittest.TestCase):
+ def check_big_decompress_buffer(self, size, decompress_func):
+ data = 'x' * size
+ try:
+ compressed = zlib.compress(data, 1)
+ finally:
+ # Release memory
+ data = None
+ data = decompress_func(compressed)
+ # Sanity check
+ try:
+ self.assertEqual(len(data), size)
+ self.assertEqual(len(data.strip('x')), 0)
+ finally:
+ data = None
+
+
+class CompressTestCase(BaseCompressTestCase, unittest.TestCase):
# Test compression in one go (whole message compression)
def test_speech(self):
x = zlib.compress(HAMLET_SCENE)
@@ -102,10 +134,19 @@
x = zlib.compress(data)
self.assertEqual(zlib.decompress(x), data)
+ # Memory use of the following functions takes into account overallocation
+
+ @precisionbigmemtest(size=_1G + 1024 * 1024, memuse=3)
+ def test_big_compress_buffer(self, size):
+ compress = lambda s: zlib.compress(s, 1)
+ self.check_big_compress_buffer(size, compress)
+
+ @precisionbigmemtest(size=_1G + 1024 * 1024, memuse=2)
+ def test_big_decompress_buffer(self, size):
+ self.check_big_decompress_buffer(size, zlib.decompress)
-
-class CompressObjectTestCase(unittest.TestCase):
+class CompressObjectTestCase(BaseCompressTestCase, unittest.TestCase):
# Test compression object
def test_pair(self):
# straightforward compress/decompress objects
@@ -379,6 +420,21 @@
d.flush()
self.assertRaises(ValueError, d.copy)
+ # Memory use of the following functions takes into account overallocation
+
+ @precisionbigmemtest(size=_1G + 1024 * 1024, memuse=3)
+ def test_big_compress_buffer(self, size):
+ c = zlib.compressobj(1)
+ compress = lambda s: c.compress(s) + c.flush()
+ self.check_big_compress_buffer(size, compress)
+
+ @precisionbigmemtest(size=_1G + 1024 * 1024, memuse=2)
+ def test_big_decompress_buffer(self, size):
+ d = zlib.decompressobj()
+ decompress = lambda s: d.decompress(s) + d.flush()
+ self.check_big_decompress_buffer(size, decompress)
+
+
def genblock(seed, length, step=1024, generator=random):
"""length-byte stream of random data from a seed (in step-byte blocks)."""
if seed is not None: