diff --git a/Lib/test/test_zipfile.py b/Lib/test/test_zipfile.py new file mode 100644 index 000000000..ae9a7dae2 --- /dev/null +++ b/Lib/test/test_zipfile.py @@ -0,0 +1,2977 @@ +import contextlib +import importlib.util +import io +import itertools +import os +import pathlib +import posixpath +import string +import struct +import subprocess +import sys +import time +import unittest +import unittest.mock as mock +import zipfile + + +from tempfile import TemporaryFile +from random import randint, random, getrandbits + +from test.support import script_helper +from test.support import (TESTFN, findfile, unlink, rmtree, temp_dir, temp_cwd, + requires_zlib, requires_bz2, requires_lzma, + captured_stdout) + +TESTFN2 = TESTFN + "2" +TESTFNDIR = TESTFN + "d" +FIXEDTEST_SIZE = 1000 +DATAFILES_DIR = 'zipfile_datafiles' + +SMALL_TEST_DATA = [('_ziptest1', '1q2w3e4r5t'), + ('ziptest2dir/_ziptest2', 'qawsedrftg'), + ('ziptest2dir/ziptest3dir/_ziptest3', 'azsxdcfvgb'), + ('ziptest2dir/ziptest3dir/ziptest4dir/_ziptest3', '6y7u8i9o0p')] + +def getrandbytes(size): + return getrandbits(8 * size).to_bytes(size, 'little') + +def get_files(test): + yield TESTFN2 + with TemporaryFile() as f: + yield f + test.assertFalse(f.closed) + with io.BytesIO() as f: + yield f + test.assertFalse(f.closed) + +class AbstractTestsWithSourceFile: + @classmethod + def setUpClass(cls): + cls.line_gen = [bytes("Zipfile test line %d. random float: %f\n" % + (i, random()), "ascii") + for i in range(FIXEDTEST_SIZE)] + cls.data = b''.join(cls.line_gen) + + def setUp(self): + # Make a source file with some lines + with open(TESTFN, "wb") as fp: + fp.write(self.data) + + def make_test_archive(self, f, compression, compresslevel=None): + kwargs = {'compression': compression, 'compresslevel': compresslevel} + # Create the ZIP archive + with zipfile.ZipFile(f, "w", **kwargs) as zipfp: + zipfp.write(TESTFN, "another.name") + zipfp.write(TESTFN, TESTFN) + zipfp.writestr("strfile", self.data) + with zipfp.open('written-open-w', mode='w') as f: + for line in self.line_gen: + f.write(line) + + def zip_test(self, f, compression, compresslevel=None): + self.make_test_archive(f, compression, compresslevel) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r", compression) as zipfp: + self.assertEqual(zipfp.read(TESTFN), self.data) + self.assertEqual(zipfp.read("another.name"), self.data) + self.assertEqual(zipfp.read("strfile"), self.data) + + # Print the ZIP directory + fp = io.StringIO() + zipfp.printdir(file=fp) + directory = fp.getvalue() + lines = directory.splitlines() + self.assertEqual(len(lines), 5) # Number of files + header + + self.assertIn('File Name', lines[0]) + self.assertIn('Modified', lines[0]) + self.assertIn('Size', lines[0]) + + fn, date, time_, size = lines[1].split() + self.assertEqual(fn, 'another.name') + self.assertTrue(time.strptime(date, '%Y-%m-%d')) + self.assertTrue(time.strptime(time_, '%H:%M:%S')) + self.assertEqual(size, str(len(self.data))) + + # Check the namelist + names = zipfp.namelist() + self.assertEqual(len(names), 4) + self.assertIn(TESTFN, names) + self.assertIn("another.name", names) + self.assertIn("strfile", names) + self.assertIn("written-open-w", names) + + # Check infolist + infos = zipfp.infolist() + names = [i.filename for i in infos] + self.assertEqual(len(names), 4) + self.assertIn(TESTFN, names) + self.assertIn("another.name", names) + self.assertIn("strfile", names) + self.assertIn("written-open-w", names) + for i in infos: + self.assertEqual(i.file_size, len(self.data)) + + # check getinfo + for nm in (TESTFN, "another.name", "strfile", "written-open-w"): + info = zipfp.getinfo(nm) + self.assertEqual(info.filename, nm) + self.assertEqual(info.file_size, len(self.data)) + + # Check that testzip doesn't raise an exception + zipfp.testzip() + + # TODO: RUSTPYTHON + @unittest.expectedFailure + def test_basic(self): + for f in get_files(self): + self.zip_test(f, self.compression) + + def zip_open_test(self, f, compression): + self.make_test_archive(f, compression) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r", compression) as zipfp: + zipdata1 = [] + with zipfp.open(TESTFN) as zipopen1: + while True: + read_data = zipopen1.read(256) + if not read_data: + break + zipdata1.append(read_data) + + zipdata2 = [] + with zipfp.open("another.name") as zipopen2: + while True: + read_data = zipopen2.read(256) + if not read_data: + break + zipdata2.append(read_data) + + self.assertEqual(b''.join(zipdata1), self.data) + self.assertEqual(b''.join(zipdata2), self.data) + + def test_open(self): + for f in get_files(self): + self.zip_open_test(f, self.compression) + + def test_open_with_pathlike(self): + path = pathlib.Path(TESTFN2) + self.zip_open_test(path, self.compression) + with zipfile.ZipFile(path, "r", self.compression) as zipfp: + self.assertIsInstance(zipfp.filename, str) + + def zip_random_open_test(self, f, compression): + self.make_test_archive(f, compression) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r", compression) as zipfp: + zipdata1 = [] + with zipfp.open(TESTFN) as zipopen1: + while True: + read_data = zipopen1.read(randint(1, 1024)) + if not read_data: + break + zipdata1.append(read_data) + + self.assertEqual(b''.join(zipdata1), self.data) + + def test_random_open(self): + for f in get_files(self): + self.zip_random_open_test(f, self.compression) + + def zip_read1_test(self, f, compression): + self.make_test_archive(f, compression) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r") as zipfp, \ + zipfp.open(TESTFN) as zipopen: + zipdata = [] + while True: + read_data = zipopen.read1(-1) + if not read_data: + break + zipdata.append(read_data) + + self.assertEqual(b''.join(zipdata), self.data) + + def test_read1(self): + for f in get_files(self): + self.zip_read1_test(f, self.compression) + + def zip_read1_10_test(self, f, compression): + self.make_test_archive(f, compression) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r") as zipfp, \ + zipfp.open(TESTFN) as zipopen: + zipdata = [] + while True: + read_data = zipopen.read1(10) + self.assertLessEqual(len(read_data), 10) + if not read_data: + break + zipdata.append(read_data) + + self.assertEqual(b''.join(zipdata), self.data) + + def test_read1_10(self): + for f in get_files(self): + self.zip_read1_10_test(f, self.compression) + + def zip_readline_read_test(self, f, compression): + self.make_test_archive(f, compression) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r") as zipfp, \ + zipfp.open(TESTFN) as zipopen: + data = b'' + while True: + read = zipopen.readline() + if not read: + break + data += read + + read = zipopen.read(100) + if not read: + break + data += read + + self.assertEqual(data, self.data) + + def test_readline_read(self): + # Issue #7610: calls to readline() interleaved with calls to read(). + for f in get_files(self): + self.zip_readline_read_test(f, self.compression) + + def zip_readline_test(self, f, compression): + self.make_test_archive(f, compression) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r") as zipfp: + with zipfp.open(TESTFN) as zipopen: + for line in self.line_gen: + linedata = zipopen.readline() + self.assertEqual(linedata, line) + + def test_readline(self): + for f in get_files(self): + self.zip_readline_test(f, self.compression) + + def zip_readlines_test(self, f, compression): + self.make_test_archive(f, compression) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r") as zipfp: + with zipfp.open(TESTFN) as zipopen: + ziplines = zipopen.readlines() + for line, zipline in zip(self.line_gen, ziplines): + self.assertEqual(zipline, line) + + def test_readlines(self): + for f in get_files(self): + self.zip_readlines_test(f, self.compression) + + def zip_iterlines_test(self, f, compression): + self.make_test_archive(f, compression) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r") as zipfp: + with zipfp.open(TESTFN) as zipopen: + for line, zipline in zip(self.line_gen, zipopen): + self.assertEqual(zipline, line) + + def test_iterlines(self): + for f in get_files(self): + self.zip_iterlines_test(f, self.compression) + + def test_low_compression(self): + """Check for cases where compressed data is larger than original.""" + # Create the ZIP archive + with zipfile.ZipFile(TESTFN2, "w", self.compression) as zipfp: + zipfp.writestr("strfile", '12') + + # Get an open object for strfile + with zipfile.ZipFile(TESTFN2, "r", self.compression) as zipfp: + with zipfp.open("strfile") as openobj: + self.assertEqual(openobj.read(1), b'1') + self.assertEqual(openobj.read(1), b'2') + + def test_writestr_compression(self): + zipfp = zipfile.ZipFile(TESTFN2, "w") + zipfp.writestr("b.txt", "hello world", compress_type=self.compression) + info = zipfp.getinfo('b.txt') + self.assertEqual(info.compress_type, self.compression) + + def test_writestr_compresslevel(self): + zipfp = zipfile.ZipFile(TESTFN2, "w", compresslevel=1) + zipfp.writestr("a.txt", "hello world", compress_type=self.compression) + zipfp.writestr("b.txt", "hello world", compress_type=self.compression, + compresslevel=2) + + # Compression level follows the constructor. + a_info = zipfp.getinfo('a.txt') + self.assertEqual(a_info.compress_type, self.compression) + self.assertEqual(a_info._compresslevel, 1) + + # Compression level is overridden. + b_info = zipfp.getinfo('b.txt') + self.assertEqual(b_info.compress_type, self.compression) + self.assertEqual(b_info._compresslevel, 2) + + def test_read_return_size(self): + # Issue #9837: ZipExtFile.read() shouldn't return more bytes + # than requested. + for test_size in (1, 4095, 4096, 4097, 16384): + file_size = test_size + 1 + junk = getrandbytes(file_size) + with zipfile.ZipFile(io.BytesIO(), "w", self.compression) as zipf: + zipf.writestr('foo', junk) + with zipf.open('foo', 'r') as fp: + buf = fp.read(test_size) + self.assertEqual(len(buf), test_size) + + def test_truncated_zipfile(self): + fp = io.BytesIO() + with zipfile.ZipFile(fp, mode='w') as zipf: + zipf.writestr('strfile', self.data, compress_type=self.compression) + end_offset = fp.tell() + zipfiledata = fp.getvalue() + + fp = io.BytesIO(zipfiledata) + with zipfile.ZipFile(fp) as zipf: + with zipf.open('strfile') as zipopen: + fp.truncate(end_offset - 20) + with self.assertRaises(EOFError): + zipopen.read() + + fp = io.BytesIO(zipfiledata) + with zipfile.ZipFile(fp) as zipf: + with zipf.open('strfile') as zipopen: + fp.truncate(end_offset - 20) + with self.assertRaises(EOFError): + while zipopen.read(100): + pass + + fp = io.BytesIO(zipfiledata) + with zipfile.ZipFile(fp) as zipf: + with zipf.open('strfile') as zipopen: + fp.truncate(end_offset - 20) + with self.assertRaises(EOFError): + while zipopen.read1(100): + pass + + def test_repr(self): + fname = 'file.name' + for f in get_files(self): + with zipfile.ZipFile(f, 'w', self.compression) as zipfp: + zipfp.write(TESTFN, fname) + r = repr(zipfp) + self.assertIn("mode='w'", r) + + with zipfile.ZipFile(f, 'r') as zipfp: + r = repr(zipfp) + if isinstance(f, str): + self.assertIn('filename=%r' % f, r) + else: + self.assertIn('file=%r' % f, r) + self.assertIn("mode='r'", r) + r = repr(zipfp.getinfo(fname)) + self.assertIn('filename=%r' % fname, r) + self.assertIn('filemode=', r) + self.assertIn('file_size=', r) + if self.compression != zipfile.ZIP_STORED: + self.assertIn('compress_type=', r) + self.assertIn('compress_size=', r) + with zipfp.open(fname) as zipopen: + r = repr(zipopen) + self.assertIn('name=%r' % fname, r) + self.assertIn("mode='r'", r) + if self.compression != zipfile.ZIP_STORED: + self.assertIn('compress_type=', r) + self.assertIn('[closed]', repr(zipopen)) + self.assertIn('[closed]', repr(zipfp)) + + # TODO: RUSTPYTHON + @unittest.expectedFailure + def test_compresslevel_basic(self): + for f in get_files(self): + self.zip_test(f, self.compression, compresslevel=9) + + def test_per_file_compresslevel(self): + """Check that files within a Zip archive can have different + compression levels.""" + with zipfile.ZipFile(TESTFN2, "w", compresslevel=1) as zipfp: + zipfp.write(TESTFN, 'compress_1') + zipfp.write(TESTFN, 'compress_9', compresslevel=9) + one_info = zipfp.getinfo('compress_1') + nine_info = zipfp.getinfo('compress_9') + self.assertEqual(one_info._compresslevel, 1) + self.assertEqual(nine_info._compresslevel, 9) + + def test_writing_errors(self): + class BrokenFile(io.BytesIO): + def write(self, data): + nonlocal count + if count is not None: + if count == stop: + raise OSError + count += 1 + super().write(data) + + stop = 0 + while True: + testfile = BrokenFile() + count = None + with zipfile.ZipFile(testfile, 'w', self.compression) as zipfp: + with zipfp.open('file1', 'w') as f: + f.write(b'data1') + count = 0 + try: + with zipfp.open('file2', 'w') as f: + f.write(b'data2') + except OSError: + stop += 1 + else: + break + finally: + count = None + with zipfile.ZipFile(io.BytesIO(testfile.getvalue())) as zipfp: + self.assertEqual(zipfp.namelist(), ['file1']) + self.assertEqual(zipfp.read('file1'), b'data1') + + with zipfile.ZipFile(io.BytesIO(testfile.getvalue())) as zipfp: + self.assertEqual(zipfp.namelist(), ['file1', 'file2']) + self.assertEqual(zipfp.read('file1'), b'data1') + self.assertEqual(zipfp.read('file2'), b'data2') + + + def tearDown(self): + unlink(TESTFN) + unlink(TESTFN2) + + +class StoredTestsWithSourceFile(AbstractTestsWithSourceFile, + unittest.TestCase): + compression = zipfile.ZIP_STORED + test_low_compression = None + + def zip_test_writestr_permissions(self, f, compression): + # Make sure that writestr and open(... mode='w') create files with + # mode 0600, when they are passed a name rather than a ZipInfo + # instance. + + self.make_test_archive(f, compression) + with zipfile.ZipFile(f, "r") as zipfp: + zinfo = zipfp.getinfo('strfile') + self.assertEqual(zinfo.external_attr, 0o600 << 16) + + zinfo2 = zipfp.getinfo('written-open-w') + self.assertEqual(zinfo2.external_attr, 0o600 << 16) + + def test_writestr_permissions(self): + for f in get_files(self): + self.zip_test_writestr_permissions(f, zipfile.ZIP_STORED) + + def test_absolute_arcnames(self): + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: + zipfp.write(TESTFN, "/absolute") + + with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: + self.assertEqual(zipfp.namelist(), ["absolute"]) + + def test_append_to_zip_file(self): + """Test appending to an existing zipfile.""" + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: + zipfp.write(TESTFN, TESTFN) + + with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp: + zipfp.writestr("strfile", self.data) + self.assertEqual(zipfp.namelist(), [TESTFN, "strfile"]) + + def test_append_to_non_zip_file(self): + """Test appending to an existing file that is not a zipfile.""" + # NOTE: this test fails if len(d) < 22 because of the first + # line "fpin.seek(-22, 2)" in _EndRecData + data = b'I am not a ZipFile!'*10 + with open(TESTFN2, 'wb') as f: + f.write(data) + + with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp: + zipfp.write(TESTFN, TESTFN) + + with open(TESTFN2, 'rb') as f: + f.seek(len(data)) + with zipfile.ZipFile(f, "r") as zipfp: + self.assertEqual(zipfp.namelist(), [TESTFN]) + self.assertEqual(zipfp.read(TESTFN), self.data) + with open(TESTFN2, 'rb') as f: + self.assertEqual(f.read(len(data)), data) + zipfiledata = f.read() + with io.BytesIO(zipfiledata) as bio, zipfile.ZipFile(bio) as zipfp: + self.assertEqual(zipfp.namelist(), [TESTFN]) + self.assertEqual(zipfp.read(TESTFN), self.data) + + def test_read_concatenated_zip_file(self): + with io.BytesIO() as bio: + with zipfile.ZipFile(bio, 'w', zipfile.ZIP_STORED) as zipfp: + zipfp.write(TESTFN, TESTFN) + zipfiledata = bio.getvalue() + data = b'I am not a ZipFile!'*10 + with open(TESTFN2, 'wb') as f: + f.write(data) + f.write(zipfiledata) + + with zipfile.ZipFile(TESTFN2) as zipfp: + self.assertEqual(zipfp.namelist(), [TESTFN]) + self.assertEqual(zipfp.read(TESTFN), self.data) + + def test_append_to_concatenated_zip_file(self): + with io.BytesIO() as bio: + with zipfile.ZipFile(bio, 'w', zipfile.ZIP_STORED) as zipfp: + zipfp.write(TESTFN, TESTFN) + zipfiledata = bio.getvalue() + data = b'I am not a ZipFile!'*1000000 + with open(TESTFN2, 'wb') as f: + f.write(data) + f.write(zipfiledata) + + with zipfile.ZipFile(TESTFN2, 'a') as zipfp: + self.assertEqual(zipfp.namelist(), [TESTFN]) + zipfp.writestr('strfile', self.data) + + with open(TESTFN2, 'rb') as f: + self.assertEqual(f.read(len(data)), data) + zipfiledata = f.read() + with io.BytesIO(zipfiledata) as bio, zipfile.ZipFile(bio) as zipfp: + self.assertEqual(zipfp.namelist(), [TESTFN, 'strfile']) + self.assertEqual(zipfp.read(TESTFN), self.data) + self.assertEqual(zipfp.read('strfile'), self.data) + + def test_ignores_newline_at_end(self): + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: + zipfp.write(TESTFN, TESTFN) + with open(TESTFN2, 'a') as f: + f.write("\r\n\00\00\00") + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + self.assertIsInstance(zipfp, zipfile.ZipFile) + + def test_ignores_stuff_appended_past_comments(self): + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: + zipfp.comment = b"this is a comment" + zipfp.write(TESTFN, TESTFN) + with open(TESTFN2, 'a') as f: + f.write("abcdef\r\n") + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + self.assertIsInstance(zipfp, zipfile.ZipFile) + self.assertEqual(zipfp.comment, b"this is a comment") + + def test_write_default_name(self): + """Check that calling ZipFile.write without arcname specified + produces the expected result.""" + with zipfile.ZipFile(TESTFN2, "w") as zipfp: + zipfp.write(TESTFN) + with open(TESTFN, "rb") as f: + self.assertEqual(zipfp.read(TESTFN), f.read()) + + def test_write_to_readonly(self): + """Check that trying to call write() on a readonly ZipFile object + raises a ValueError.""" + with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: + zipfp.writestr("somefile.txt", "bogus") + + with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: + self.assertRaises(ValueError, zipfp.write, TESTFN) + + with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: + with self.assertRaises(ValueError): + zipfp.open(TESTFN, mode='w') + + def test_add_file_before_1980(self): + # Set atime and mtime to 1970-01-01 + os.utime(TESTFN, (0, 0)) + with zipfile.ZipFile(TESTFN2, "w") as zipfp: + self.assertRaises(ValueError, zipfp.write, TESTFN) + + with zipfile.ZipFile(TESTFN2, "w", strict_timestamps=False) as zipfp: + zipfp.write(TESTFN) + zinfo = zipfp.getinfo(TESTFN) + self.assertEqual(zinfo.date_time, (1980, 1, 1, 0, 0, 0)) + + # TODO: RUSTPYTHON + @unittest.expectedFailure + def test_add_file_after_2107(self): + # Set atime and mtime to 2108-12-30 + ts = 4386268800 + try: + time.localtime(ts) + except OverflowError: + self.skipTest(f'time.localtime({ts}) raises OverflowError') + try: + os.utime(TESTFN, (ts, ts)) + except OverflowError: + self.skipTest('Host fs cannot set timestamp to required value.') + + mtime_ns = os.stat(TESTFN).st_mtime_ns + if mtime_ns != (4386268800 * 10**9): + # XFS filesystem is limited to 32-bit timestamp, but the syscall + # didn't fail. Moreover, there is a VFS bug which returns + # a cached timestamp which is different than the value on disk. + # + # Test st_mtime_ns rather than st_mtime to avoid rounding issues. + # + # https://bugzilla.redhat.com/show_bug.cgi?id=1795576 + # https://bugs.python.org/issue39460#msg360952 + self.skipTest(f"Linux VFS/XFS kernel bug detected: {mtime_ns=}") + + with zipfile.ZipFile(TESTFN2, "w") as zipfp: + self.assertRaises(struct.error, zipfp.write, TESTFN) + + with zipfile.ZipFile(TESTFN2, "w", strict_timestamps=False) as zipfp: + zipfp.write(TESTFN) + zinfo = zipfp.getinfo(TESTFN) + self.assertEqual(zinfo.date_time, (2107, 12, 31, 23, 59, 59)) + + +@requires_zlib +class DeflateTestsWithSourceFile(AbstractTestsWithSourceFile, + unittest.TestCase): + compression = zipfile.ZIP_DEFLATED + + def test_per_file_compression(self): + """Check that files within a Zip archive can have different + compression options.""" + with zipfile.ZipFile(TESTFN2, "w") as zipfp: + zipfp.write(TESTFN, 'storeme', zipfile.ZIP_STORED) + zipfp.write(TESTFN, 'deflateme', zipfile.ZIP_DEFLATED) + sinfo = zipfp.getinfo('storeme') + dinfo = zipfp.getinfo('deflateme') + self.assertEqual(sinfo.compress_type, zipfile.ZIP_STORED) + self.assertEqual(dinfo.compress_type, zipfile.ZIP_DEFLATED) + +@requires_bz2 +class Bzip2TestsWithSourceFile(AbstractTestsWithSourceFile, + unittest.TestCase): + compression = zipfile.ZIP_BZIP2 + +@requires_lzma +class LzmaTestsWithSourceFile(AbstractTestsWithSourceFile, + unittest.TestCase): + compression = zipfile.ZIP_LZMA + + +class AbstractTestZip64InSmallFiles: + # These tests test the ZIP64 functionality without using large files, + # see test_zipfile64 for proper tests. + + @classmethod + def setUpClass(cls): + line_gen = (bytes("Test of zipfile line %d." % i, "ascii") + for i in range(0, FIXEDTEST_SIZE)) + cls.data = b'\n'.join(line_gen) + + def setUp(self): + self._limit = zipfile.ZIP64_LIMIT + self._filecount_limit = zipfile.ZIP_FILECOUNT_LIMIT + zipfile.ZIP64_LIMIT = 1000 + zipfile.ZIP_FILECOUNT_LIMIT = 9 + + # Make a source file with some lines + with open(TESTFN, "wb") as fp: + fp.write(self.data) + + def zip_test(self, f, compression): + # Create the ZIP archive + with zipfile.ZipFile(f, "w", compression, allowZip64=True) as zipfp: + zipfp.write(TESTFN, "another.name") + zipfp.write(TESTFN, TESTFN) + zipfp.writestr("strfile", self.data) + + # Read the ZIP archive + with zipfile.ZipFile(f, "r", compression) as zipfp: + self.assertEqual(zipfp.read(TESTFN), self.data) + self.assertEqual(zipfp.read("another.name"), self.data) + self.assertEqual(zipfp.read("strfile"), self.data) + + # Print the ZIP directory + fp = io.StringIO() + zipfp.printdir(fp) + + directory = fp.getvalue() + lines = directory.splitlines() + self.assertEqual(len(lines), 4) # Number of files + header + + self.assertIn('File Name', lines[0]) + self.assertIn('Modified', lines[0]) + self.assertIn('Size', lines[0]) + + fn, date, time_, size = lines[1].split() + self.assertEqual(fn, 'another.name') + self.assertTrue(time.strptime(date, '%Y-%m-%d')) + self.assertTrue(time.strptime(time_, '%H:%M:%S')) + self.assertEqual(size, str(len(self.data))) + + # Check the namelist + names = zipfp.namelist() + self.assertEqual(len(names), 3) + self.assertIn(TESTFN, names) + self.assertIn("another.name", names) + self.assertIn("strfile", names) + + # Check infolist + infos = zipfp.infolist() + names = [i.filename for i in infos] + self.assertEqual(len(names), 3) + self.assertIn(TESTFN, names) + self.assertIn("another.name", names) + self.assertIn("strfile", names) + for i in infos: + self.assertEqual(i.file_size, len(self.data)) + + # check getinfo + for nm in (TESTFN, "another.name", "strfile"): + info = zipfp.getinfo(nm) + self.assertEqual(info.filename, nm) + self.assertEqual(info.file_size, len(self.data)) + + # Check that testzip doesn't raise an exception + zipfp.testzip() + + # TODO: RUSTPYTHON + @unittest.expectedFailure + def test_basic(self): + for f in get_files(self): + self.zip_test(f, self.compression) + + def test_too_many_files(self): + # This test checks that more than 64k files can be added to an archive, + # and that the resulting archive can be read properly by ZipFile + zipf = zipfile.ZipFile(TESTFN, "w", self.compression, + allowZip64=True) + zipf.debug = 100 + numfiles = 15 + for i in range(numfiles): + zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) + self.assertEqual(len(zipf.namelist()), numfiles) + zipf.close() + + zipf2 = zipfile.ZipFile(TESTFN, "r", self.compression) + self.assertEqual(len(zipf2.namelist()), numfiles) + for i in range(numfiles): + content = zipf2.read("foo%08d" % i).decode('ascii') + self.assertEqual(content, "%d" % (i**3 % 57)) + zipf2.close() + + def test_too_many_files_append(self): + zipf = zipfile.ZipFile(TESTFN, "w", self.compression, + allowZip64=False) + zipf.debug = 100 + numfiles = 9 + for i in range(numfiles): + zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) + self.assertEqual(len(zipf.namelist()), numfiles) + with self.assertRaises(zipfile.LargeZipFile): + zipf.writestr("foo%08d" % numfiles, b'') + self.assertEqual(len(zipf.namelist()), numfiles) + zipf.close() + + zipf = zipfile.ZipFile(TESTFN, "a", self.compression, + allowZip64=False) + zipf.debug = 100 + self.assertEqual(len(zipf.namelist()), numfiles) + with self.assertRaises(zipfile.LargeZipFile): + zipf.writestr("foo%08d" % numfiles, b'') + self.assertEqual(len(zipf.namelist()), numfiles) + zipf.close() + + zipf = zipfile.ZipFile(TESTFN, "a", self.compression, + allowZip64=True) + zipf.debug = 100 + self.assertEqual(len(zipf.namelist()), numfiles) + numfiles2 = 15 + for i in range(numfiles, numfiles2): + zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) + self.assertEqual(len(zipf.namelist()), numfiles2) + zipf.close() + + zipf2 = zipfile.ZipFile(TESTFN, "r", self.compression) + self.assertEqual(len(zipf2.namelist()), numfiles2) + for i in range(numfiles2): + content = zipf2.read("foo%08d" % i).decode('ascii') + self.assertEqual(content, "%d" % (i**3 % 57)) + zipf2.close() + + def tearDown(self): + zipfile.ZIP64_LIMIT = self._limit + zipfile.ZIP_FILECOUNT_LIMIT = self._filecount_limit + unlink(TESTFN) + unlink(TESTFN2) + + +class StoredTestZip64InSmallFiles(AbstractTestZip64InSmallFiles, + unittest.TestCase): + compression = zipfile.ZIP_STORED + + def large_file_exception_test(self, f, compression): + with zipfile.ZipFile(f, "w", compression, allowZip64=False) as zipfp: + self.assertRaises(zipfile.LargeZipFile, + zipfp.write, TESTFN, "another.name") + + def large_file_exception_test2(self, f, compression): + with zipfile.ZipFile(f, "w", compression, allowZip64=False) as zipfp: + self.assertRaises(zipfile.LargeZipFile, + zipfp.writestr, "another.name", self.data) + + def test_large_file_exception(self): + for f in get_files(self): + self.large_file_exception_test(f, zipfile.ZIP_STORED) + self.large_file_exception_test2(f, zipfile.ZIP_STORED) + + def test_absolute_arcnames(self): + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED, + allowZip64=True) as zipfp: + zipfp.write(TESTFN, "/absolute") + + with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: + self.assertEqual(zipfp.namelist(), ["absolute"]) + + def test_append(self): + # Test that appending to the Zip64 archive doesn't change + # extra fields of existing entries. + with zipfile.ZipFile(TESTFN2, "w", allowZip64=True) as zipfp: + zipfp.writestr("strfile", self.data) + with zipfile.ZipFile(TESTFN2, "r", allowZip64=True) as zipfp: + zinfo = zipfp.getinfo("strfile") + extra = zinfo.extra + with zipfile.ZipFile(TESTFN2, "a", allowZip64=True) as zipfp: + zipfp.writestr("strfile2", self.data) + with zipfile.ZipFile(TESTFN2, "r", allowZip64=True) as zipfp: + zinfo = zipfp.getinfo("strfile") + self.assertEqual(zinfo.extra, extra) + + def make_zip64_file( + self, file_size_64_set=False, file_size_extra=False, + compress_size_64_set=False, compress_size_extra=False, + header_offset_64_set=False, header_offset_extra=False, + ): + """Generate bytes sequence for a zip with (incomplete) zip64 data. + + The actual values (not the zip 64 0xffffffff values) stored in the file + are: + file_size: 8 + compress_size: 8 + header_offset: 0 + """ + actual_size = 8 + actual_header_offset = 0 + local_zip64_fields = [] + central_zip64_fields = [] + + file_size = actual_size + if file_size_64_set: + file_size = 0xffffffff + if file_size_extra: + local_zip64_fields.append(actual_size) + central_zip64_fields.append(actual_size) + file_size = struct.pack("e|f"g?h*i', ','), r'a\b,c_d_e_f_g_h_i') + self.assertEqual(san('../../foo../../ba..r', '/'), r'foo/ba..r') + + def test_extract_hackers_arcnames_common_cases(self): + common_hacknames = [ + ('../foo/bar', 'foo/bar'), + ('foo/../bar', 'foo/bar'), + ('foo/../../bar', 'foo/bar'), + ('foo/bar/..', 'foo/bar'), + ('./../foo/bar', 'foo/bar'), + ('/foo/bar', 'foo/bar'), + ('/foo/../bar', 'foo/bar'), + ('/foo/../../bar', 'foo/bar'), + ] + self._test_extract_hackers_arcnames(common_hacknames) + + @unittest.skipIf(os.path.sep != '\\', 'Requires \\ as path separator.') + def test_extract_hackers_arcnames_windows_only(self): + """Test combination of path fixing and windows name sanitization.""" + windows_hacknames = [ + (r'..\foo\bar', 'foo/bar'), + (r'..\/foo\/bar', 'foo/bar'), + (r'foo/\..\/bar', 'foo/bar'), + (r'foo\/../\bar', 'foo/bar'), + (r'C:foo/bar', 'foo/bar'), + (r'C:/foo/bar', 'foo/bar'), + (r'C://foo/bar', 'foo/bar'), + (r'C:\foo\bar', 'foo/bar'), + (r'//conky/mountpoint/foo/bar', 'foo/bar'), + (r'\\conky\mountpoint\foo\bar', 'foo/bar'), + (r'///conky/mountpoint/foo/bar', 'conky/mountpoint/foo/bar'), + (r'\\\conky\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'), + (r'//conky//mountpoint/foo/bar', 'conky/mountpoint/foo/bar'), + (r'\\conky\\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'), + (r'//?/C:/foo/bar', 'foo/bar'), + (r'\\?\C:\foo\bar', 'foo/bar'), + (r'C:/../C:/foo/bar', 'C_/foo/bar'), + (r'a:b\ce|f"g?h*i', 'b/c_d_e_f_g_h_i'), + ('../../foo../../ba..r', 'foo/ba..r'), + ] + self._test_extract_hackers_arcnames(windows_hacknames) + + @unittest.skipIf(os.path.sep != '/', r'Requires / as path separator.') + def test_extract_hackers_arcnames_posix_only(self): + posix_hacknames = [ + ('//foo/bar', 'foo/bar'), + ('../../foo../../ba..r', 'foo../ba..r'), + (r'foo/..\bar', r'foo/..\bar'), + ] + self._test_extract_hackers_arcnames(posix_hacknames) + + def _test_extract_hackers_arcnames(self, hacknames): + for arcname, fixedname in hacknames: + content = b'foobar' + arcname.encode() + with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipfp: + zinfo = zipfile.ZipInfo() + # preserve backslashes + zinfo.filename = arcname + zinfo.external_attr = 0o600 << 16 + zipfp.writestr(zinfo, content) + + arcname = arcname.replace(os.sep, "/") + targetpath = os.path.join('target', 'subdir', 'subsub') + correctfile = os.path.join(targetpath, *fixedname.split('/')) + + with zipfile.ZipFile(TESTFN2, 'r') as zipfp: + writtenfile = zipfp.extract(arcname, targetpath) + self.assertEqual(writtenfile, correctfile, + msg='extract %r: %r != %r' % + (arcname, writtenfile, correctfile)) + self.check_file(correctfile, content) + rmtree('target') + + with zipfile.ZipFile(TESTFN2, 'r') as zipfp: + zipfp.extractall(targetpath) + self.check_file(correctfile, content) + rmtree('target') + + correctfile = os.path.join(os.getcwd(), *fixedname.split('/')) + + with zipfile.ZipFile(TESTFN2, 'r') as zipfp: + writtenfile = zipfp.extract(arcname) + self.assertEqual(writtenfile, correctfile, + msg="extract %r" % arcname) + self.check_file(correctfile, content) + rmtree(fixedname.split('/')[0]) + + with zipfile.ZipFile(TESTFN2, 'r') as zipfp: + zipfp.extractall() + self.check_file(correctfile, content) + rmtree(fixedname.split('/')[0]) + + unlink(TESTFN2) + + +class OtherTests(unittest.TestCase): + def test_open_via_zip_info(self): + # Create the ZIP archive + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: + zipfp.writestr("name", "foo") + with self.assertWarns(UserWarning): + zipfp.writestr("name", "bar") + self.assertEqual(zipfp.namelist(), ["name"] * 2) + + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + infos = zipfp.infolist() + data = b"" + for info in infos: + with zipfp.open(info) as zipopen: + data += zipopen.read() + self.assertIn(data, {b"foobar", b"barfoo"}) + data = b"" + for info in infos: + data += zipfp.read(info) + self.assertIn(data, {b"foobar", b"barfoo"}) + + def test_writestr_extended_local_header_issue1202(self): + with zipfile.ZipFile(TESTFN2, 'w') as orig_zip: + for data in 'abcdefghijklmnop': + zinfo = zipfile.ZipInfo(data) + zinfo.flag_bits |= 0x08 # Include an extended local header. + orig_zip.writestr(zinfo, data) + + def test_close(self): + """Check that the zipfile is closed after the 'with' block.""" + with zipfile.ZipFile(TESTFN2, "w") as zipfp: + for fpath, fdata in SMALL_TEST_DATA: + zipfp.writestr(fpath, fdata) + self.assertIsNotNone(zipfp.fp, 'zipfp is not open') + self.assertIsNone(zipfp.fp, 'zipfp is not closed') + + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + self.assertIsNotNone(zipfp.fp, 'zipfp is not open') + self.assertIsNone(zipfp.fp, 'zipfp is not closed') + + def test_close_on_exception(self): + """Check that the zipfile is closed if an exception is raised in the + 'with' block.""" + with zipfile.ZipFile(TESTFN2, "w") as zipfp: + for fpath, fdata in SMALL_TEST_DATA: + zipfp.writestr(fpath, fdata) + + try: + with zipfile.ZipFile(TESTFN2, "r") as zipfp2: + raise zipfile.BadZipFile() + except zipfile.BadZipFile: + self.assertIsNone(zipfp2.fp, 'zipfp is not closed') + + def test_unsupported_version(self): + # File has an extract_version of 120 + data = (b'PK\x03\x04x\x00\x00\x00\x00\x00!p\xa1@\x00\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00xPK\x01\x02x\x03x\x00\x00\x00\x00' + b'\x00!p\xa1@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00\x00xPK\x05\x06' + b'\x00\x00\x00\x00\x01\x00\x01\x00/\x00\x00\x00\x1f\x00\x00\x00\x00\x00') + + self.assertRaises(NotImplementedError, zipfile.ZipFile, + io.BytesIO(data), 'r') + + @requires_zlib + def test_read_unicode_filenames(self): + # bug #10801 + fname = findfile('zip_cp437_header.zip') + with zipfile.ZipFile(fname) as zipfp: + for name in zipfp.namelist(): + zipfp.open(name).close() + + def test_write_unicode_filenames(self): + with zipfile.ZipFile(TESTFN, "w") as zf: + zf.writestr("foo.txt", "Test for unicode filename") + zf.writestr("\xf6.txt", "Test for unicode filename") + self.assertIsInstance(zf.infolist()[0].filename, str) + + with zipfile.ZipFile(TESTFN, "r") as zf: + self.assertEqual(zf.filelist[0].filename, "foo.txt") + self.assertEqual(zf.filelist[1].filename, "\xf6.txt") + + # TODO: RUSTPYTHON + @unittest.expectedFailure + def test_read_after_write_unicode_filenames(self): + with zipfile.ZipFile(TESTFN2, 'w') as zipfp: + zipfp.writestr('приклад', b'sample') + self.assertEqual(zipfp.read('приклад'), b'sample') + + def test_exclusive_create_zip_file(self): + """Test exclusive creating a new zipfile.""" + unlink(TESTFN2) + filename = 'testfile.txt' + content = b'hello, world. this is some content.' + with zipfile.ZipFile(TESTFN2, "x", zipfile.ZIP_STORED) as zipfp: + zipfp.writestr(filename, content) + with self.assertRaises(FileExistsError): + zipfile.ZipFile(TESTFN2, "x", zipfile.ZIP_STORED) + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + self.assertEqual(zipfp.namelist(), [filename]) + self.assertEqual(zipfp.read(filename), content) + + def test_create_non_existent_file_for_append(self): + if os.path.exists(TESTFN): + os.unlink(TESTFN) + + filename = 'testfile.txt' + content = b'hello, world. this is some content.' + + try: + with zipfile.ZipFile(TESTFN, 'a') as zf: + zf.writestr(filename, content) + except OSError: + self.fail('Could not append data to a non-existent zip file.') + + self.assertTrue(os.path.exists(TESTFN)) + + with zipfile.ZipFile(TESTFN, 'r') as zf: + self.assertEqual(zf.read(filename), content) + + def test_close_erroneous_file(self): + # This test checks that the ZipFile constructor closes the file object + # it opens if there's an error in the file. If it doesn't, the + # traceback holds a reference to the ZipFile object and, indirectly, + # the file object. + # On Windows, this causes the os.unlink() call to fail because the + # underlying file is still open. This is SF bug #412214. + # + with open(TESTFN, "w") as fp: + fp.write("this is not a legal zip file\n") + try: + zf = zipfile.ZipFile(TESTFN) + except zipfile.BadZipFile: + pass + + def test_is_zip_erroneous_file(self): + """Check that is_zipfile() correctly identifies non-zip files.""" + # - passing a filename + with open(TESTFN, "w") as fp: + fp.write("this is not a legal zip file\n") + self.assertFalse(zipfile.is_zipfile(TESTFN)) + # - passing a path-like object + self.assertFalse(zipfile.is_zipfile(pathlib.Path(TESTFN))) + # - passing a file object + with open(TESTFN, "rb") as fp: + self.assertFalse(zipfile.is_zipfile(fp)) + # - passing a file-like object + fp = io.BytesIO() + fp.write(b"this is not a legal zip file\n") + self.assertFalse(zipfile.is_zipfile(fp)) + fp.seek(0, 0) + self.assertFalse(zipfile.is_zipfile(fp)) + + def test_damaged_zipfile(self): + """Check that zipfiles with missing bytes at the end raise BadZipFile.""" + # - Create a valid zip file + fp = io.BytesIO() + with zipfile.ZipFile(fp, mode="w") as zipf: + zipf.writestr("foo.txt", b"O, for a Muse of Fire!") + zipfiledata = fp.getvalue() + + # - Now create copies of it missing the last N bytes and make sure + # a BadZipFile exception is raised when we try to open it + for N in range(len(zipfiledata)): + fp = io.BytesIO(zipfiledata[:N]) + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, fp) + + def test_is_zip_valid_file(self): + """Check that is_zipfile() correctly identifies zip files.""" + # - passing a filename + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + zipf.writestr("foo.txt", b"O, for a Muse of Fire!") + + self.assertTrue(zipfile.is_zipfile(TESTFN)) + # - passing a file object + with open(TESTFN, "rb") as fp: + self.assertTrue(zipfile.is_zipfile(fp)) + fp.seek(0, 0) + zip_contents = fp.read() + # - passing a file-like object + fp = io.BytesIO() + fp.write(zip_contents) + self.assertTrue(zipfile.is_zipfile(fp)) + fp.seek(0, 0) + self.assertTrue(zipfile.is_zipfile(fp)) + + def test_non_existent_file_raises_OSError(self): + # make sure we don't raise an AttributeError when a partially-constructed + # ZipFile instance is finalized; this tests for regression on SF tracker + # bug #403871. + + # The bug we're testing for caused an AttributeError to be raised + # when a ZipFile instance was created for a file that did not + # exist; the .fp member was not initialized but was needed by the + # __del__() method. Since the AttributeError is in the __del__(), + # it is ignored, but the user should be sufficiently annoyed by + # the message on the output that regression will be noticed + # quickly. + self.assertRaises(OSError, zipfile.ZipFile, TESTFN) + + def test_empty_file_raises_BadZipFile(self): + f = open(TESTFN, 'w') + f.close() + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) + + with open(TESTFN, 'w') as fp: + fp.write("short file") + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) + + def test_closed_zip_raises_ValueError(self): + """Verify that testzip() doesn't swallow inappropriate exceptions.""" + data = io.BytesIO() + with zipfile.ZipFile(data, mode="w") as zipf: + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + + # This is correct; calling .read on a closed ZipFile should raise + # a ValueError, and so should calling .testzip. An earlier + # version of .testzip would swallow this exception (and any other) + # and report that the first file in the archive was corrupt. + self.assertRaises(ValueError, zipf.read, "foo.txt") + self.assertRaises(ValueError, zipf.open, "foo.txt") + self.assertRaises(ValueError, zipf.testzip) + self.assertRaises(ValueError, zipf.writestr, "bogus.txt", "bogus") + with open(TESTFN, 'w') as f: + f.write('zipfile test data') + self.assertRaises(ValueError, zipf.write, TESTFN) + + def test_bad_constructor_mode(self): + """Check that bad modes passed to ZipFile constructor are caught.""" + self.assertRaises(ValueError, zipfile.ZipFile, TESTFN, "q") + + def test_bad_open_mode(self): + """Check that bad modes passed to ZipFile.open are caught.""" + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + + with zipfile.ZipFile(TESTFN, mode="r") as zipf: + # read the data to make sure the file is there + zipf.read("foo.txt") + self.assertRaises(ValueError, zipf.open, "foo.txt", "q") + # universal newlines support is removed + self.assertRaises(ValueError, zipf.open, "foo.txt", "U") + self.assertRaises(ValueError, zipf.open, "foo.txt", "rU") + + def test_read0(self): + """Check that calling read(0) on a ZipExtFile object returns an empty + string and doesn't advance file pointer.""" + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + # read the data to make sure the file is there + with zipf.open("foo.txt") as f: + for i in range(FIXEDTEST_SIZE): + self.assertEqual(f.read(0), b'') + + self.assertEqual(f.read(), b"O, for a Muse of Fire!") + + def test_open_non_existent_item(self): + """Check that attempting to call open() for an item that doesn't + exist in the archive raises a RuntimeError.""" + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + self.assertRaises(KeyError, zipf.open, "foo.txt", "r") + + def test_bad_compression_mode(self): + """Check that bad compression methods passed to ZipFile.open are + caught.""" + self.assertRaises(NotImplementedError, zipfile.ZipFile, TESTFN, "w", -1) + + def test_unsupported_compression(self): + # data is declared as shrunk, but actually deflated + data = (b'PK\x03\x04.\x00\x00\x00\x01\x00\xe4C\xa1@\x00\x00\x00' + b'\x00\x02\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00x\x03\x00PK\x01' + b'\x02.\x03.\x00\x00\x00\x01\x00\xe4C\xa1@\x00\x00\x00\x00\x02\x00\x00' + b'\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x80\x01\x00\x00\x00\x00xPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x00' + b'/\x00\x00\x00!\x00\x00\x00\x00\x00') + with zipfile.ZipFile(io.BytesIO(data), 'r') as zipf: + self.assertRaises(NotImplementedError, zipf.open, 'x') + + def test_null_byte_in_filename(self): + """Check that a filename containing a null byte is properly + terminated.""" + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + zipf.writestr("foo.txt\x00qqq", b"O, for a Muse of Fire!") + self.assertEqual(zipf.namelist(), ['foo.txt']) + + def test_struct_sizes(self): + """Check that ZIP internal structure sizes are calculated correctly.""" + self.assertEqual(zipfile.sizeEndCentDir, 22) + self.assertEqual(zipfile.sizeCentralDir, 46) + self.assertEqual(zipfile.sizeEndCentDir64, 56) + self.assertEqual(zipfile.sizeEndCentDir64Locator, 20) + + # TODO: RUSTPYTHON + @unittest.expectedFailure + def test_comments(self): + """Check that comments on the archive are handled properly.""" + + # check default comment is empty + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + self.assertEqual(zipf.comment, b'') + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + + with zipfile.ZipFile(TESTFN, mode="r") as zipfr: + self.assertEqual(zipfr.comment, b'') + + # check a simple short comment + comment = b'Bravely taking to his feet, he beat a very brave retreat.' + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + zipf.comment = comment + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + with zipfile.ZipFile(TESTFN, mode="r") as zipfr: + self.assertEqual(zipf.comment, comment) + + # check a comment of max length + comment2 = ''.join(['%d' % (i**3 % 10) for i in range((1 << 16)-1)]) + comment2 = comment2.encode("ascii") + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + zipf.comment = comment2 + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + + with zipfile.ZipFile(TESTFN, mode="r") as zipfr: + self.assertEqual(zipfr.comment, comment2) + + # check a comment that is too long is truncated + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + with self.assertWarns(UserWarning): + zipf.comment = comment2 + b'oops' + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + with zipfile.ZipFile(TESTFN, mode="r") as zipfr: + self.assertEqual(zipfr.comment, comment2) + + # check that comments are correctly modified in append mode + with zipfile.ZipFile(TESTFN,mode="w") as zipf: + zipf.comment = b"original comment" + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + with zipfile.ZipFile(TESTFN,mode="a") as zipf: + zipf.comment = b"an updated comment" + with zipfile.ZipFile(TESTFN,mode="r") as zipf: + self.assertEqual(zipf.comment, b"an updated comment") + + # check that comments are correctly shortened in append mode + # and the file is indeed truncated + with zipfile.ZipFile(TESTFN,mode="w") as zipf: + zipf.comment = b"original comment that's longer" + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + original_zip_size = os.path.getsize(TESTFN) + with zipfile.ZipFile(TESTFN,mode="a") as zipf: + zipf.comment = b"shorter comment" + self.assertTrue(original_zip_size > os.path.getsize(TESTFN)) + with zipfile.ZipFile(TESTFN,mode="r") as zipf: + self.assertEqual(zipf.comment, b"shorter comment") + + def test_unicode_comment(self): + with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf: + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + with self.assertRaises(TypeError): + zipf.comment = "this is an error" + + def test_change_comment_in_empty_archive(self): + with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf: + self.assertFalse(zipf.filelist) + zipf.comment = b"this is a comment" + with zipfile.ZipFile(TESTFN, "r") as zipf: + self.assertEqual(zipf.comment, b"this is a comment") + + def test_change_comment_in_nonempty_archive(self): + with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf: + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf: + self.assertTrue(zipf.filelist) + zipf.comment = b"this is a comment" + with zipfile.ZipFile(TESTFN, "r") as zipf: + self.assertEqual(zipf.comment, b"this is a comment") + + def test_empty_zipfile(self): + # Check that creating a file in 'w' or 'a' mode and closing without + # adding any files to the archives creates a valid empty ZIP file + zipf = zipfile.ZipFile(TESTFN, mode="w") + zipf.close() + try: + zipf = zipfile.ZipFile(TESTFN, mode="r") + except zipfile.BadZipFile: + self.fail("Unable to create empty ZIP file in 'w' mode") + + zipf = zipfile.ZipFile(TESTFN, mode="a") + zipf.close() + try: + zipf = zipfile.ZipFile(TESTFN, mode="r") + except: + self.fail("Unable to create empty ZIP file in 'a' mode") + + def test_open_empty_file(self): + # Issue 1710703: Check that opening a file with less than 22 bytes + # raises a BadZipFile exception (rather than the previously unhelpful + # OSError) + f = open(TESTFN, 'w') + f.close() + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN, 'r') + + def test_create_zipinfo_before_1980(self): + self.assertRaises(ValueError, + zipfile.ZipInfo, 'seventies', (1979, 1, 1, 0, 0, 0)) + + def test_zipfile_with_short_extra_field(self): + """If an extra field in the header is less than 4 bytes, skip it.""" + zipdata = ( + b'PK\x03\x04\x14\x00\x00\x00\x00\x00\x93\x9b\xad@\x8b\x9e' + b'\xd9\xd3\x01\x00\x00\x00\x01\x00\x00\x00\x03\x00\x03\x00ab' + b'c\x00\x00\x00APK\x01\x02\x14\x03\x14\x00\x00\x00\x00' + b'\x00\x93\x9b\xad@\x8b\x9e\xd9\xd3\x01\x00\x00\x00\x01\x00\x00' + b'\x00\x03\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa4\x81\x00' + b'\x00\x00\x00abc\x00\x00PK\x05\x06\x00\x00\x00\x00' + b'\x01\x00\x01\x003\x00\x00\x00%\x00\x00\x00\x00\x00' + ) + with zipfile.ZipFile(io.BytesIO(zipdata), 'r') as zipf: + # testzip returns the name of the first corrupt file, or None + self.assertIsNone(zipf.testzip()) + + def test_open_conflicting_handles(self): + # It's only possible to open one writable file handle at a time + msg1 = b"It's fun to charter an accountant!" + msg2 = b"And sail the wide accountant sea" + msg3 = b"To find, explore the funds offshore" + with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipf: + with zipf.open('foo', mode='w') as w2: + w2.write(msg1) + with zipf.open('bar', mode='w') as w1: + with self.assertRaises(ValueError): + zipf.open('handle', mode='w') + with self.assertRaises(ValueError): + zipf.open('foo', mode='r') + with self.assertRaises(ValueError): + zipf.writestr('str', 'abcde') + with self.assertRaises(ValueError): + zipf.write(__file__, 'file') + with self.assertRaises(ValueError): + zipf.close() + w1.write(msg2) + with zipf.open('baz', mode='w') as w2: + w2.write(msg3) + + with zipfile.ZipFile(TESTFN2, 'r') as zipf: + self.assertEqual(zipf.read('foo'), msg1) + self.assertEqual(zipf.read('bar'), msg2) + self.assertEqual(zipf.read('baz'), msg3) + self.assertEqual(zipf.namelist(), ['foo', 'bar', 'baz']) + + def test_seek_tell(self): + # Test seek functionality + txt = b"Where's Bruce?" + bloc = txt.find(b"Bruce") + # Check seek on a file + with zipfile.ZipFile(TESTFN, "w") as zipf: + zipf.writestr("foo.txt", txt) + with zipfile.ZipFile(TESTFN, "r") as zipf: + with zipf.open("foo.txt", "r") as fp: + fp.seek(bloc, os.SEEK_SET) + self.assertEqual(fp.tell(), bloc) + fp.seek(-bloc, os.SEEK_CUR) + self.assertEqual(fp.tell(), 0) + fp.seek(bloc, os.SEEK_CUR) + self.assertEqual(fp.tell(), bloc) + self.assertEqual(fp.read(5), txt[bloc:bloc+5]) + fp.seek(0, os.SEEK_END) + self.assertEqual(fp.tell(), len(txt)) + fp.seek(0, os.SEEK_SET) + self.assertEqual(fp.tell(), 0) + # Check seek on memory file + data = io.BytesIO() + with zipfile.ZipFile(data, mode="w") as zipf: + zipf.writestr("foo.txt", txt) + with zipfile.ZipFile(data, mode="r") as zipf: + with zipf.open("foo.txt", "r") as fp: + fp.seek(bloc, os.SEEK_SET) + self.assertEqual(fp.tell(), bloc) + fp.seek(-bloc, os.SEEK_CUR) + self.assertEqual(fp.tell(), 0) + fp.seek(bloc, os.SEEK_CUR) + self.assertEqual(fp.tell(), bloc) + self.assertEqual(fp.read(5), txt[bloc:bloc+5]) + fp.seek(0, os.SEEK_END) + self.assertEqual(fp.tell(), len(txt)) + fp.seek(0, os.SEEK_SET) + self.assertEqual(fp.tell(), 0) + + @requires_bz2 + def test_decompress_without_3rd_party_library(self): + data = b'PK\x05\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' + zip_file = io.BytesIO(data) + with zipfile.ZipFile(zip_file, 'w', compression=zipfile.ZIP_BZIP2) as zf: + zf.writestr('a.txt', b'a') + with mock.patch('zipfile.bz2', None): + with zipfile.ZipFile(zip_file) as zf: + self.assertRaises(RuntimeError, zf.extract, 'a.txt') + + def tearDown(self): + unlink(TESTFN) + unlink(TESTFN2) + + +class AbstractBadCrcTests: + def test_testzip_with_bad_crc(self): + """Tests that files with bad CRCs return their name from testzip.""" + zipdata = self.zip_with_bad_crc + + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + # testzip returns the name of the first corrupt file, or None + self.assertEqual('afile', zipf.testzip()) + + def test_read_with_bad_crc(self): + """Tests that files with bad CRCs raise a BadZipFile exception when read.""" + zipdata = self.zip_with_bad_crc + + # Using ZipFile.read() + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + self.assertRaises(zipfile.BadZipFile, zipf.read, 'afile') + + # Using ZipExtFile.read() + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + with zipf.open('afile', 'r') as corrupt_file: + self.assertRaises(zipfile.BadZipFile, corrupt_file.read) + + # Same with small reads (in order to exercise the buffering logic) + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + with zipf.open('afile', 'r') as corrupt_file: + corrupt_file.MIN_READ_SIZE = 2 + with self.assertRaises(zipfile.BadZipFile): + while corrupt_file.read(2): + pass + + +class StoredBadCrcTests(AbstractBadCrcTests, unittest.TestCase): + compression = zipfile.ZIP_STORED + zip_with_bad_crc = ( + b'PK\003\004\024\0\0\0\0\0 \213\212;:r' + b'\253\377\f\0\0\0\f\0\0\0\005\0\0\000af' + b'ilehello,AworldP' + b'K\001\002\024\003\024\0\0\0\0\0 \213\212;:' + b'r\253\377\f\0\0\0\f\0\0\0\005\0\0\0\0' + b'\0\0\0\0\0\0\0\200\001\0\0\0\000afi' + b'lePK\005\006\0\0\0\0\001\0\001\0003\000' + b'\0\0/\0\0\0\0\0') + +@requires_zlib +class DeflateBadCrcTests(AbstractBadCrcTests, unittest.TestCase): + compression = zipfile.ZIP_DEFLATED + zip_with_bad_crc = ( + b'PK\x03\x04\x14\x00\x00\x00\x08\x00n}\x0c=FA' + b'KE\x10\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' + b'ile\xcbH\xcd\xc9\xc9W(\xcf/\xcaI\xc9\xa0' + b'=\x13\x00PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00n' + b'}\x0c=FAKE\x10\x00\x00\x00n\x00\x00\x00\x05' + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00' + b'\x00afilePK\x05\x06\x00\x00\x00\x00\x01\x00' + b'\x01\x003\x00\x00\x003\x00\x00\x00\x00\x00') + +@requires_bz2 +class Bzip2BadCrcTests(AbstractBadCrcTests, unittest.TestCase): + compression = zipfile.ZIP_BZIP2 + zip_with_bad_crc = ( + b'PK\x03\x04\x14\x03\x00\x00\x0c\x00nu\x0c=FA' + b'KE8\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' + b'ileBZh91AY&SY\xd4\xa8\xca' + b'\x7f\x00\x00\x0f\x11\x80@\x00\x06D\x90\x80 \x00 \xa5' + b'P\xd9!\x03\x03\x13\x13\x13\x89\xa9\xa9\xc2u5:\x9f' + b'\x8b\xb9"\x9c(HjTe?\x80PK\x01\x02\x14' + b'\x03\x14\x03\x00\x00\x0c\x00nu\x0c=FAKE8' + b'\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x00 \x80\x80\x81\x00\x00\x00\x00afilePK' + b'\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00\x00[\x00' + b'\x00\x00\x00\x00') + +@requires_lzma +class LzmaBadCrcTests(AbstractBadCrcTests, unittest.TestCase): + compression = zipfile.ZIP_LZMA + zip_with_bad_crc = ( + b'PK\x03\x04\x14\x03\x00\x00\x0e\x00nu\x0c=FA' + b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' + b'ile\t\x04\x05\x00]\x00\x00\x00\x04\x004\x19I' + b'\xee\x8d\xe9\x17\x89:3`\tq!.8\x00PK' + b'\x01\x02\x14\x03\x14\x03\x00\x00\x0e\x00nu\x0c=FA' + b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00 \x80\x80\x81\x00\x00\x00\x00afil' + b'ePK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00' + b'\x00>\x00\x00\x00\x00\x00') + + +class DecryptionTests(unittest.TestCase): + """Check that ZIP decryption works. Since the library does not + support encryption at the moment, we use a pre-generated encrypted + ZIP file.""" + + data = ( + b'PK\x03\x04\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00\x1a\x00' + b'\x00\x00\x08\x00\x00\x00test.txt\xfa\x10\xa0gly|\xfa-\xc5\xc0=\xf9y' + b'\x18\xe0\xa8r\xb3Z}Lg\xbc\xae\xf9|\x9b\x19\xe4\x8b\xba\xbb)\x8c\xb0\xdbl' + b'PK\x01\x02\x14\x00\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00' + b'\x1a\x00\x00\x00\x08\x00\x00\x00\x00\x00\x00\x00\x01\x00 \x00\xb6\x81' + b'\x00\x00\x00\x00test.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x006\x00' + b'\x00\x00L\x00\x00\x00\x00\x00' ) + data2 = ( + b'PK\x03\x04\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02' + b'\x00\x00\x04\x00\x15\x00zeroUT\t\x00\x03\xd6\x8b\x92G\xda\x8b\x92GUx\x04' + b'\x00\xe8\x03\xe8\x03\xc7exe_with_zip +rm zip2.zip +``` + +### Modern format (4.5) zip64 file + +Redirecting from stdin forces infozip's zip tool to create a zip64. + +``` +zip -0 zip64.zip +cat header.sh zip64.zip >exe_with_z64 +rm zip64.zip +``` + diff --git a/Lib/test/ziptestdata/exe_with_z64 b/Lib/test/ziptestdata/exe_with_z64 new file mode 100755 index 000000000..82b03cf39 Binary files /dev/null and b/Lib/test/ziptestdata/exe_with_z64 differ diff --git a/Lib/test/ziptestdata/exe_with_zip b/Lib/test/ziptestdata/exe_with_zip new file mode 100755 index 000000000..c833cdf9f Binary files /dev/null and b/Lib/test/ziptestdata/exe_with_zip differ diff --git a/Lib/test/ziptestdata/header.sh b/Lib/test/ziptestdata/header.sh new file mode 100755 index 000000000..52dc91acf --- /dev/null +++ b/Lib/test/ziptestdata/header.sh @@ -0,0 +1,24 @@ +#!/bin/bash +INTERPRETER_UNDER_TEST="$1" +if [[ ! -x "${INTERPRETER_UNDER_TEST}" ]]; then + echo "Interpreter must be the command line argument." + exit 4 +fi +EXECUTABLE="$0" exec "${INTERPRETER_UNDER_TEST}" -E - <