https://github.com/python/cpython/commit/e5ed2ad9be8e0014a3bf4dc9f89c4ad2695500cd commit: e5ed2ad9be8e0014a3bf4dc9f89c4ad2695500cd branch: main author: Danny Lin <[email protected]> committer: gpshead <[email protected]> date: 2026-08-24T09:54:55-07:00 summary:
gh-152190: Fix memory checking failure in `test_zipfile64.py` (GH-152203) * gh-152190: Fix memory checking failure in `test_strip_removed_large_file_with_dd_no_sig` Remove the overly restrictive `allowed_memory` override (200 KiB) in `test_strip_removed_large_file_with_dd_no_sig` to prevent a memory checking failure. * gh-152190: Revise comment about the empirical memory threshold * gh-152190: Improve memory checking accuracy for `test_zipfile64` Introduce the `assert_memory_usage` context manager to narrow the scope of tracemalloc tracking down exclusively to the file-repacking phase. This prevents previous file-writing side effects from interfering with the baseline, improves tracing accuracy, and eliminates redundant inner sub-function wrappers. * gh-152190: Improve coding style and docstrings * gh-152190: Remove unneeded comments and checks Remove redundant "TESTFN2" disk space warnings from TestRepack, as these tests exclusively use TemporaryFile(). Additionally, remove the repetitive `self.assertFalse(f.closed)` checks from `TestRepack` methods since it's already verified in `TestsWithSourceFile`. * gh-152190: Further optimize tests and tidy code Rename `TestRepack` to `TestRepacker` to better reflect its coverage of `zipfile._Repacker`. Move heavy text chunk generation and common constants from `setUp` to `setUpClass` to ensure single initialization across tests. Clean up repetitive local definitions of filenames and lorem text by promoting them to class properties. Reduce the test file size from 8 GiB to 4.1 GiB, which is large enough to trigger ZIP64 extension and sufficient for the testing purpose. --------- Co-authored-by: Zachary Ware <[email protected]> files: M Lib/test/test_zipfile64.py diff --git a/Lib/test/test_zipfile64.py b/Lib/test/test_zipfile64.py index 7d802d59849ce1..e13f064f2ac4fd 100644 --- a/Lib/test/test_zipfile64.py +++ b/Lib/test/test_zipfile64.py @@ -17,6 +17,7 @@ import sys import unittest.mock as mock +from contextlib import contextmanager from tempfile import TemporaryFile from test.support import os_helper @@ -91,176 +92,133 @@ def tearDown(self): os_helper.unlink(TESTFN2) -class TestRepack(unittest.TestCase): - def setUp(self): - # Create test data. - line_gen = ("Test of zipfile line %d." % i for i in range(1000000)) - self.data = '\n'.join(line_gen).encode('ascii') - - # It will contain enough copies of self.data to reach about 8 GiB. - self.datacount = 8*1024**3 // len(self.data) +class TestRepacker(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.largefilename = 'largefile.txt' - # memory usage should not exceed 10 MiB - self.allowed_memory = 10*1024**2 + line_gen = ("Test of zipfile line %d." % i for i in range(1000000)) + cls.chunk = '\n'.join(line_gen).encode('ascii') + + # It will contain enough copies of cls.chunk to reach about 4.1 GiB. + cls.chunkcount = int(4.1*1024**3 / len(cls.chunk)) + + cls.filename = 'file.txt' + cls.lorem = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' + + # Memory usage should not exceed 10 MiB during repacking. + # This empirical threshold ensures that the internal processing + # like signature scanning, compressed block end tracing, and + # data copying are properly buffered without loading the entire + # large file into memory. + cls.allowed_memory = 10*1024**2 + + @contextmanager + def assert_memory_usage(self, threshold): + tracemalloc.start() + try: + yield + finally: + current, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + self.assertLess(peak, threshold) def _write_large_file(self, fh): next_time = time.monotonic() + _PRINT_WORKING_MSG_INTERVAL - for num in range(self.datacount): - fh.write(self.data) + for num in range(self.chunkcount): + fh.write(self.chunk) # Print still working message since this test can be really slow if next_time <= time.monotonic(): next_time = time.monotonic() + _PRINT_WORKING_MSG_INTERVAL print(( ' writing %d of %d, be patient...' % - (num, self.datacount)), file=sys.__stdout__) + (num, self.chunkcount)), file=sys.__stdout__) sys.__stdout__.flush() def test_strip_removed_large_file(self): """Should move the physical data of a file positioned after a large removed file without causing a memory issue.""" - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_large_file(f) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_large_file(self, f): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with zipfile.ZipFile(f, 'w') as zh: - with zh.open(file1, 'w', force_zip64=True) as fh: - self._write_large_file(fh) - zh.writestr(file, data) - - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file1) - zh.repack() - self.assertIsNone(zh.testzip()) + with zipfile.ZipFile(f, 'w') as zh: + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: + self._write_large_file(fh) + zh.writestr(self.filename, self.lorem) + + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.largefilename) + zh.repack() + self.assertIsNone(zh.testzip()) def test_strip_removed_file_before_large_file(self): """Should move the physical data of a large file positioned after a removed file without causing a memory issue.""" - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_file_before_large_file(f) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_file_before_large_file(self, f): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with zipfile.ZipFile(f, 'w') as zh: - zh.writestr(file, data) - with zh.open(file1, 'w', force_zip64=True) as fh: - self._write_large_file(fh) - - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file) - zh.repack() - self.assertIsNone(zh.testzip()) + with zipfile.ZipFile(f, 'w') as zh: + zh.writestr(self.filename, self.lorem) + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: + self._write_large_file(fh) + + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.filename) + zh.repack() + self.assertIsNone(zh.testzip()) def test_strip_removed_large_file_with_dd(self): """Should scan for the data descriptor of a removed large file without causing a memory issue.""" - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_large_file_with_dd(f) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_large_file_with_dd(self, f): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with zipfile.ZipFile(Unseekable(f), 'w') as zh: - with zh.open(file1, 'w', force_zip64=True) as fh: - self._write_large_file(fh) - zh.writestr(file, data) - - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file1) - zh.repack() - self.assertIsNone(zh.testzip()) + with zipfile.ZipFile(Unseekable(f), 'w') as zh: + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: + self._write_large_file(fh) + zh.writestr(self.filename, self.lorem) + + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.largefilename) + zh.repack() + self.assertIsNone(zh.testzip()) def test_strip_removed_large_file_with_dd_no_sig(self): - """Should scan for the data descriptor (without signature) of a removed - large file without causing a memory issue.""" + """Should scan for the unsigned data descriptor of a removed large file + without causing a memory issue.""" # Reduce data scale for this test, as it's especially slow... - self.datacount = 30*1024**2 // len(self.data) - self.allowed_memory = 200*1024 + self.chunkcount = int(30*1024**2 / len(self.chunk)) - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_large_file_with_dd_no_sig(f) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_large_file_with_dd_no_sig(self, f): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with mock.patch('zipfile.struct.pack', side_effect=struct_pack_no_dd_sig): - with zipfile.ZipFile(Unseekable(f), 'w') as zh: - with zh.open(file1, 'w', force_zip64=True) as fh: + with mock.patch('zipfile.struct.pack', side_effect=struct_pack_no_dd_sig), \ + zipfile.ZipFile(Unseekable(f), 'w') as zh: + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: self._write_large_file(fh) - zh.writestr(file, data) + zh.writestr(self.filename, self.lorem) - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file1) - # strict_descriptor=False to scan the unsigned data descriptor - # (scanning is disabled under the strict_descriptor=True default) - zh.repack(strict_descriptor=False) - self.assertIsNone(zh.testzip()) + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.largefilename) + # strict_descriptor=False to scan the unsigned data descriptor + # (scanning is disabled under the strict_descriptor=True default) + zh.repack(strict_descriptor=False) + self.assertIsNone(zh.testzip()) @requires_zlib() def test_strip_removed_large_file_with_dd_no_sig_by_decompression(self): - """Should scan for the data descriptor (without signature) of a removed - large file without causing a memory issue.""" - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. + """Should scan for the unsigned data descriptor (via tracing compressed + block end) of a removed large file without causing a memory issue.""" with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_large_file_with_dd_no_sig_by_decompression( - f, zipfile.ZIP_DEFLATED) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_large_file_with_dd_no_sig_by_decompression(self, f, method): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with mock.patch('zipfile.struct.pack', side_effect=struct_pack_no_dd_sig): - with zipfile.ZipFile(Unseekable(f), 'w', compression=method) as zh: - with zh.open(file1, 'w', force_zip64=True) as fh: + with mock.patch('zipfile.struct.pack', side_effect=struct_pack_no_dd_sig), \ + zipfile.ZipFile(Unseekable(f), 'w', compression=zipfile.ZIP_DEFLATED) as zh: + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: self._write_large_file(fh) - zh.writestr(file, data) - - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file1) - # strict_descriptor=False to detect the unsigned data descriptor - # (scanning is disabled under the strict_descriptor=True default) - zh.repack(strict_descriptor=False) - self.assertIsNone(zh.testzip()) + zh.writestr(self.filename, self.lorem) + + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.largefilename) + # strict_descriptor=False to detect the unsigned data descriptor + # (scanning is disabled under the strict_descriptor=True default) + zh.repack(strict_descriptor=False) + self.assertIsNone(zh.testzip()) class OtherTests(unittest.TestCase): _______________________________________________ Python-checkins mailing list -- [email protected] To unsubscribe send an email to [email protected] https://mail.python.org/mailman3//lists/python-checkins.python.org Member address: [email protected]
