| # We can test part of the module without zlib. | |
| try: | |
| import zlib | |
| except ImportError: | |
| zlib = None | |
| import os | |
| import io | |
| import sys | |
| import time | |
| import struct | |
| import zipfile | |
| import unittest | |
| from StringIO import StringIO | |
| from tempfile import TemporaryFile | |
| from random import randint, random, getrandbits | |
| from unittest import skipUnless | |
| from test import script_helper | |
| from test.test_support import TESTFN, TESTFN_UNICODE, TESTFN_ENCODING, \ | |
| run_unittest, findfile, unlink, rmtree, \ | |
| check_warnings, captured_stdout | |
| try: | |
| TESTFN_UNICODE.encode(TESTFN_ENCODING) | |
| except (UnicodeError, TypeError): | |
| # Either the file system encoding is None, or the file name | |
| # cannot be encoded in the file system encoding. | |
| TESTFN_UNICODE = None | |
| TESTFN2 = TESTFN + "2" | |
| TESTFNDIR = TESTFN + "d" | |
| FIXEDTEST_SIZE = 1000 | |
| SMALL_TEST_DATA = [('_ziptest1', '1q2w3e4r5t'), | |
| ('ziptest2dir/_ziptest2', 'qawsedrftg'), | |
| ('ziptest2dir/ziptest3dir/_ziptest3', 'azsxdcfvgb'), | |
| ('ziptest2dir/ziptest3dir/ziptest4dir/_ziptest3', '6y7u8i9o0p')] | |
| def getrandbytes(size): | |
| return bytes(bytearray.fromhex('%0*x' % (2 * size, getrandbits(8 * size)))) | |
| class TestsWithSourceFile(unittest.TestCase): | |
| def setUp(self): | |
| self.line_gen = ["Zipfile test line %d. random float: %f" % (i, random()) | |
| for i in xrange(FIXEDTEST_SIZE)] | |
| self.data = '\n'.join(self.line_gen) + '\n' | |
| # Make a source file with some lines | |
| with open(TESTFN, "wb") as fp: | |
| fp.write(self.data) | |
| def make_test_archive(self, f, compression): | |
| # Create the ZIP archive | |
| with zipfile.ZipFile(f, "w", compression) as zipfp: | |
| zipfp.write(TESTFN, "another.name") | |
| zipfp.write(TESTFN, TESTFN) | |
| zipfp.writestr("strfile", self.data) | |
| def zip_test(self, f, compression): | |
| self.make_test_archive(f, compression) | |
| # 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 = StringIO() | |
| stdout = sys.stdout | |
| try: | |
| sys.stdout = fp | |
| zipfp.printdir() | |
| finally: | |
| sys.stdout = stdout | |
| 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() | |
| def test_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_test(f, zipfile.ZIP_STORED) | |
| 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(''.join(zipdata1), self.data) | |
| self.assertEqual(''.join(zipdata2), self.data) | |
| def test_open_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_open_test(f, zipfile.ZIP_STORED) | |
| 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 check_warnings(('', UserWarning)): | |
| zipfp.writestr("name", "bar") | |
| self.assertEqual(zipfp.namelist(), ["name"] * 2) | |
| with zipfile.ZipFile(TESTFN2, "r") as zipfp: | |
| infos = zipfp.infolist() | |
| data = "" | |
| for info in infos: | |
| with zipfp.open(info) as f: | |
| data += f.read() | |
| self.assertTrue(data == "foobar" or data == "barfoo") | |
| data = "" | |
| for info in infos: | |
| data += zipfp.read(info) | |
| self.assertTrue(data == "foobar" or data == "barfoo") | |
| 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(''.join(zipdata1), self.data) | |
| def test_random_open_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_random_open_test(f, zipfile.ZIP_STORED) | |
| def test_universal_readaheads(self): | |
| f = StringIO() | |
| data = 'a\r\n' * 16 * 1024 | |
| with zipfile.ZipFile(f, 'w', zipfile.ZIP_STORED) as zipfp: | |
| zipfp.writestr(TESTFN, data) | |
| data2 = '' | |
| with zipfile.ZipFile(f, 'r') as zipfp: | |
| with zipfp.open(TESTFN, 'rU') as zipopen: | |
| for line in zipopen: | |
| data2 += line | |
| self.assertEqual(data, data2.replace('\n', '\r\n')) | |
| 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: | |
| with zipfp.open(TESTFN) as zipopen: | |
| data = '' | |
| 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 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 + '\n') | |
| 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 zo: | |
| ziplines = zo.readlines() | |
| for line, zipline in zip(self.line_gen, ziplines): | |
| self.assertEqual(zipline, line + '\n') | |
| def zip_iterlines_test(self, f, compression): | |
| self.make_test_archive(f, compression) | |
| # Read the ZIP archive | |
| with zipfile.ZipFile(f, "r") as zipfp: | |
| for line, zipline in zip(self.line_gen, zipfp.open(TESTFN)): | |
| self.assertEqual(zipline, line + '\n') | |
| def test_readline_read_stored(self): | |
| # Issue #7610: calls to readline() interleaved with calls to read(). | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_readline_read_test(f, zipfile.ZIP_STORED) | |
| def test_readline_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_readline_test(f, zipfile.ZIP_STORED) | |
| def test_readlines_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_readlines_test(f, zipfile.ZIP_STORED) | |
| def test_iterlines_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_iterlines_test(f, zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_open_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_open_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_random_open_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_random_open_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_readline_read_deflated(self): | |
| # Issue #7610: calls to readline() interleaved with calls to read(). | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_readline_read_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_readline_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_readline_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_readlines_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_readlines_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_iterlines_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_iterlines_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_low_compression(self): | |
| """Check for cases where compressed data is larger than original.""" | |
| # Create the ZIP archive | |
| with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_DEFLATED) as zipfp: | |
| zipfp.writestr("strfile", '12') | |
| # Get an open object for strfile | |
| with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_DEFLATED) as zipfp: | |
| with zipfp.open("strfile") as openobj: | |
| self.assertEqual(openobj.read(1), '1') | |
| self.assertEqual(openobj.read(1), '2') | |
| 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 = '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,'r') as fid: | |
| self.assertEqual(zipfp.read(TESTFN), fid.read()) | |
| @skipUnless(zlib, "requires zlib") | |
| 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) | |
| def test_write_to_readonly(self): | |
| """Check that trying to call write() on a readonly ZipFile object | |
| raises a RuntimeError.""" | |
| with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: | |
| zipfp.writestr("somefile.txt", "bogus") | |
| with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: | |
| self.assertRaises(RuntimeError, zipfp.write, TESTFN) | |
| def test_extract(self): | |
| with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: | |
| for fpath, fdata in SMALL_TEST_DATA: | |
| zipfp.writestr(fpath, fdata) | |
| with zipfile.ZipFile(TESTFN2, "r") as zipfp: | |
| for fpath, fdata in SMALL_TEST_DATA: | |
| writtenfile = zipfp.extract(fpath) | |
| # make sure it was written to the right place | |
| correctfile = os.path.join(os.getcwd(), fpath) | |
| correctfile = os.path.normpath(correctfile) | |
| self.assertEqual(writtenfile, correctfile) | |
| # make sure correct data is in correct file | |
| with open(writtenfile, "rb") as fid: | |
| self.assertEqual(fdata, fid.read()) | |
| os.remove(writtenfile) | |
| # remove the test file subdirectories | |
| rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) | |
| def test_extract_all(self): | |
| with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: | |
| for fpath, fdata in SMALL_TEST_DATA: | |
| zipfp.writestr(fpath, fdata) | |
| with zipfile.ZipFile(TESTFN2, "r") as zipfp: | |
| zipfp.extractall() | |
| for fpath, fdata in SMALL_TEST_DATA: | |
| outfile = os.path.join(os.getcwd(), fpath) | |
| with open(outfile, "rb") as fid: | |
| self.assertEqual(fdata, fid.read()) | |
| os.remove(outfile) | |
| # remove the test file subdirectories | |
| rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) | |
| def check_file(self, filename, content): | |
| self.assertTrue(os.path.isfile(filename)) | |
| with open(filename, 'rb') as f: | |
| self.assertEqual(f.read(), content) | |
| @skipUnless(TESTFN_UNICODE, "No Unicode filesystem semantics on this platform.") | |
| def test_extract_unicode_filenames(self): | |
| fnames = [u'foo.txt', os.path.basename(TESTFN_UNICODE)] | |
| content = 'Test for unicode filename' | |
| with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: | |
| for fname in fnames: | |
| zipfp.writestr(fname, content) | |
| with zipfile.ZipFile(TESTFN2, "r") as zipfp: | |
| for fname in fnames: | |
| writtenfile = zipfp.extract(fname) | |
| # make sure it was written to the right place | |
| correctfile = os.path.join(os.getcwd(), fname) | |
| correctfile = os.path.normpath(correctfile) | |
| self.assertEqual(writtenfile, correctfile) | |
| self.check_file(writtenfile, content) | |
| os.remove(writtenfile) | |
| def test_extract_hackers_arcnames(self): | |
| 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'), | |
| ] | |
| if os.path.sep == '\\': | |
| hacknames.extend([ | |
| (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\c<d>e|f"g?h*i', 'b/c_d_e_f_g_h_i'), | |
| ('../../foo../../ba..r', 'foo/ba..r'), | |
| ]) | |
| else: # Unix | |
| hacknames.extend([ | |
| ('//foo/bar', 'foo/bar'), | |
| ('../../foo../../ba..r', 'foo../ba..r'), | |
| (r'foo/..\bar', r'foo/..\bar'), | |
| ]) | |
| 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" % arcname) | |
| 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]) | |
| os.remove(TESTFN2) | |
| def test_writestr_compression(self): | |
| zipfp = zipfile.ZipFile(TESTFN2, "w") | |
| zipfp.writestr("a.txt", "hello world", compress_type=zipfile.ZIP_STORED) | |
| if zlib: | |
| zipfp.writestr("b.txt", "hello world", compress_type=zipfile.ZIP_DEFLATED) | |
| info = zipfp.getinfo('a.txt') | |
| self.assertEqual(info.compress_type, zipfile.ZIP_STORED) | |
| if zlib: | |
| info = zipfp.getinfo('b.txt') | |
| self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED) | |
| def zip_test_writestr_permissions(self, f, compression): | |
| # Make sure that writestr creates files with mode 0600, | |
| # when it is 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, 0600 << 16) | |
| def test_writestr_permissions(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_test_writestr_permissions(f, zipfile.ZIP_STORED) | |
| 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.assertTrue(zipfp.fp is not None, 'zipfp is not open') | |
| self.assertTrue(zipfp.fp is None, 'zipfp is not closed') | |
| with zipfile.ZipFile(TESTFN2, "r") as zipfp: | |
| self.assertTrue(zipfp.fp is not None, 'zipfp is not open') | |
| self.assertTrue(zipfp.fp is None, '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.assertTrue(zipfp2.fp is None, 'zipfp is not closed') | |
| 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) | |
| def tearDown(self): | |
| unlink(TESTFN) | |
| unlink(TESTFN2) | |
| class TestZip64InSmallFiles(unittest.TestCase): | |
| # These tests test the ZIP64 functionality without using large files, | |
| # see test_zipfile64 for proper tests. | |
| def setUp(self): | |
| self._limit = zipfile.ZIP64_LIMIT | |
| self._filecount_limit = zipfile.ZIP_FILECOUNT_LIMIT | |
| zipfile.ZIP64_LIMIT = 1000 | |
| zipfile.ZIP_FILECOUNT_LIMIT = 9 | |
| line_gen = ("Test of zipfile line %d." % i | |
| for i in range(0, FIXEDTEST_SIZE)) | |
| self.data = '\n'.join(line_gen) | |
| # Make a source file with some lines | |
| with open(TESTFN, "wb") as fp: | |
| fp.write(self.data) | |
| def large_file_exception_test(self, f, compression): | |
| with zipfile.ZipFile(f, "w", compression) 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) as zipfp: | |
| self.assertRaises(zipfile.LargeZipFile, | |
| zipfp.writestr, "another.name", self.data) | |
| def test_large_file_exception(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.large_file_exception_test(f, zipfile.ZIP_STORED) | |
| self.large_file_exception_test2(f, zipfile.ZIP_STORED) | |
| 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 = StringIO() | |
| stdout = sys.stdout | |
| try: | |
| sys.stdout = fp | |
| zipfp.printdir() | |
| finally: | |
| sys.stdout = stdout | |
| 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() | |
| def test_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_test(f, zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_test(f, zipfile.ZIP_DEFLATED) | |
| 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_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, mode="w", 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, mode="r") | |
| self.assertEqual(len(zipf2.namelist()), numfiles) | |
| for i in range(numfiles): | |
| content = zipf2.read("foo%08d" % i) | |
| self.assertEqual(content, "%d" % (i**3 % 57)) | |
| zipf2.close() | |
| def test_too_many_files_append(self): | |
| zipf = zipfile.ZipFile(TESTFN, mode="w", 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, mode="a", 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, mode="a", 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, mode="r") | |
| self.assertEqual(len(zipf2.namelist()), numfiles2) | |
| for i in range(numfiles2): | |
| content = zipf2.read("foo%08d" % i) | |
| 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 PyZipFileTests(unittest.TestCase): | |
| def requiresWriteAccess(self, path): | |
| if not os.access(path, os.W_OK): | |
| self.skipTest('requires write access to the installed location') | |
| filename = os.path.join(path, 'test_zipfile.try') | |
| try: | |
| fd = os.open(filename, os.O_WRONLY | os.O_CREAT) | |
| os.close(fd) | |
| except Exception: | |
| self.skipTest('requires write access to the installed location') | |
| unlink(filename) | |
| def test_write_pyfile(self): | |
| self.requiresWriteAccess(os.path.dirname(__file__)) | |
| with zipfile.PyZipFile(TemporaryFile(), "w") as zipfp: | |
| fn = __file__ | |
| if fn.endswith('.pyc') or fn.endswith('.pyo'): | |
| fn = fn[:-1] | |
| zipfp.writepy(fn) | |
| bn = os.path.basename(fn) | |
| self.assertNotIn(bn, zipfp.namelist()) | |
| self.assertTrue(bn + 'o' in zipfp.namelist() or | |
| bn + 'c' in zipfp.namelist()) | |
| with zipfile.PyZipFile(TemporaryFile(), "w") as zipfp: | |
| fn = __file__ | |
| if fn.endswith(('.pyc', '.pyo')): | |
| fn = fn[:-1] | |
| zipfp.writepy(fn, "testpackage") | |
| bn = "%s/%s" % ("testpackage", os.path.basename(fn)) | |
| self.assertNotIn(bn, zipfp.namelist()) | |
| self.assertTrue(bn + 'o' in zipfp.namelist() or | |
| bn + 'c' in zipfp.namelist()) | |
| def test_write_python_package(self): | |
| import email | |
| packagedir = os.path.dirname(email.__file__) | |
| self.requiresWriteAccess(packagedir) | |
| with zipfile.PyZipFile(TemporaryFile(), "w") as zipfp: | |
| zipfp.writepy(packagedir) | |
| # Check for a couple of modules at different levels of the | |
| # hierarchy | |
| names = zipfp.namelist() | |
| self.assertTrue('email/__init__.pyo' in names or | |
| 'email/__init__.pyc' in names) | |
| self.assertTrue('email/mime/text.pyo' in names or | |
| 'email/mime/text.pyc' in names) | |
| def test_write_python_directory(self): | |
| os.mkdir(TESTFN2) | |
| try: | |
| with open(os.path.join(TESTFN2, "mod1.py"), "w") as fp: | |
| fp.write("print(42)\n") | |
| with open(os.path.join(TESTFN2, "mod2.py"), "w") as fp: | |
| fp.write("print(42 * 42)\n") | |
| with open(os.path.join(TESTFN2, "mod2.txt"), "w") as fp: | |
| fp.write("bla bla bla\n") | |
| zipfp = zipfile.PyZipFile(TemporaryFile(), "w") | |
| zipfp.writepy(TESTFN2) | |
| names = zipfp.namelist() | |
| self.assertTrue('mod1.pyc' in names or 'mod1.pyo' in names) | |
| self.assertTrue('mod2.pyc' in names or 'mod2.pyo' in names) | |
| self.assertNotIn('mod2.txt', names) | |
| finally: | |
| rmtree(TESTFN2) | |
| def test_write_non_pyfile(self): | |
| with zipfile.PyZipFile(TemporaryFile(), "w") as zipfp: | |
| with open(TESTFN, 'w') as fid: | |
| fid.write('most definitely not a python file') | |
| self.assertRaises(RuntimeError, zipfp.writepy, TESTFN) | |
| os.remove(TESTFN) | |
| class OtherTests(unittest.TestCase): | |
| zips_with_bad_crc = { | |
| zipfile.ZIP_STORED: ( | |
| 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'), | |
| zipfile.ZIP_DEFLATED: ( | |
| 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'), | |
| } | |
| def test_unicode_filenames(self): | |
| with zipfile.ZipFile(TESTFN, "w") as zf: | |
| zf.writestr(u"foo.txt", "Test for unicode filename") | |
| zf.writestr(u"\xf6.txt", "Test for unicode filename") | |
| self.assertIsInstance(zf.infolist()[0].filename, unicode) | |
| with zipfile.ZipFile(TESTFN, "r") as zf: | |
| self.assertEqual(zf.filelist[0].filename, "foo.txt") | |
| self.assertEqual(zf.filelist[1].filename, u"\xf6.txt") | |
| def test_create_non_existent_file_for_append(self): | |
| if os.path.exists(TESTFN): | |
| os.unlink(TESTFN) | |
| filename = 'testfile.txt' | |
| content = 'hello, world. this is some content.' | |
| try: | |
| with zipfile.ZipFile(TESTFN, 'a') as zf: | |
| zf.writestr(filename, content) | |
| except IOError: | |
| 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") | |
| chk = zipfile.is_zipfile(TESTFN) | |
| self.assertFalse(chk) | |
| # - passing a file object | |
| with open(TESTFN, "rb") as fp: | |
| chk = zipfile.is_zipfile(fp) | |
| self.assertTrue(not chk) | |
| # - passing a file-like object | |
| fp = StringIO() | |
| fp.write("this is not a legal zip file\n") | |
| chk = zipfile.is_zipfile(fp) | |
| self.assertTrue(not chk) | |
| fp.seek(0, 0) | |
| chk = zipfile.is_zipfile(fp) | |
| self.assertTrue(not chk) | |
| 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", "O, for a Muse of Fire!") | |
| chk = zipfile.is_zipfile(TESTFN) | |
| self.assertTrue(chk) | |
| # - passing a file object | |
| with open(TESTFN, "rb") as fp: | |
| chk = zipfile.is_zipfile(fp) | |
| self.assertTrue(chk) | |
| fp.seek(0, 0) | |
| zip_contents = fp.read() | |
| # - passing a file-like object | |
| fp = StringIO() | |
| fp.write(zip_contents) | |
| chk = zipfile.is_zipfile(fp) | |
| self.assertTrue(chk) | |
| fp.seek(0, 0) | |
| chk = zipfile.is_zipfile(fp) | |
| self.assertTrue(chk) | |
| def test_non_existent_file_raises_IOError(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(IOError, zipfile.ZipFile, TESTFN) | |
| def test_empty_file_raises_BadZipFile(self): | |
| with open(TESTFN, 'w') as f: | |
| pass | |
| 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_RuntimeError(self): | |
| """Verify that testzip() doesn't swallow inappropriate exceptions.""" | |
| data = StringIO() | |
| 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 RuntimeError, 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(RuntimeError, zipf.read, "foo.txt") | |
| self.assertRaises(RuntimeError, zipf.open, "foo.txt") | |
| self.assertRaises(RuntimeError, zipf.testzip) | |
| self.assertRaises(RuntimeError, zipf.writestr, "bogus.txt", "bogus") | |
| with open(TESTFN, 'w') as fid: | |
| fid.write('zipfile test data') | |
| self.assertRaises(RuntimeError, zipf.write, TESTFN) | |
| def test_bad_constructor_mode(self): | |
| """Check that bad modes passed to ZipFile constructor are caught.""" | |
| self.assertRaises(RuntimeError, 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(RuntimeError, zipf.open, "foo.txt", "q") | |
| 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 xrange(FIXEDTEST_SIZE): | |
| self.assertEqual(f.read(0), '') | |
| self.assertEqual(f.read(), "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(RuntimeError, 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", "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) | |
| 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, '') | |
| zipf.writestr("foo.txt", "O, for a Muse of Fire!") | |
| with zipfile.ZipFile(TESTFN, mode="r") as zipf: | |
| self.assertEqual(zipf.comment, '') | |
| # check a simple short comment | |
| comment = '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 zipf: | |
| self.assertEqual(zipf.comment, comment) | |
| # check a comment of max length | |
| comment2 = ''.join(['%d' % (i**3 % 10) for i in xrange((1 << 16)-1)]) | |
| 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 zipf: | |
| self.assertEqual(zipf.comment, comment2) | |
| # check a comment that is too long is truncated | |
| with zipfile.ZipFile(TESTFN, mode="w") as zipf: | |
| with check_warnings(('', UserWarning)): | |
| zipf.comment = comment2 + 'oops' | |
| zipf.writestr("foo.txt", "O, for a Muse of Fire!") | |
| with zipfile.ZipFile(TESTFN, mode="r") as zipf: | |
| self.assertEqual(zipf.comment, comment2) | |
| 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 check_testzip_with_bad_crc(self, compression): | |
| """Tests that files with bad CRCs return their name from testzip.""" | |
| zipdata = self.zips_with_bad_crc[compression] | |
| 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_testzip_with_bad_crc_stored(self): | |
| self.check_testzip_with_bad_crc(zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_testzip_with_bad_crc_deflated(self): | |
| self.check_testzip_with_bad_crc(zipfile.ZIP_DEFLATED) | |
| def check_read_with_bad_crc(self, compression): | |
| """Tests that files with bad CRCs raise a BadZipfile exception when read.""" | |
| zipdata = self.zips_with_bad_crc[compression] | |
| # 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 | |
| def test_read_with_bad_crc_stored(self): | |
| self.check_read_with_bad_crc(zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_read_with_bad_crc_deflated(self): | |
| self.check_read_with_bad_crc(zipfile.ZIP_DEFLATED) | |
| def check_read_return_size(self, compression): | |
| # 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", 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_read_return_size_stored(self): | |
| self.check_read_return_size(zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_read_return_size_deflated(self): | |
| self.check_read_return_size(zipfile.ZIP_DEFLATED) | |
| 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 | |
| with zipfile.ZipFile(TESTFN, mode="w") as zipf: | |
| pass | |
| try: | |
| zipf = zipfile.ZipFile(TESTFN, mode="r") | |
| zipf.close() | |
| except zipfile.BadZipfile: | |
| self.fail("Unable to create empty ZIP file in 'w' mode") | |
| with zipfile.ZipFile(TESTFN, mode="a") as zipf: | |
| pass | |
| try: | |
| zipf = zipfile.ZipFile(TESTFN, mode="r") | |
| zipf.close() | |
| 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 | |
| # IOError) | |
| with open(TESTFN, 'w') as f: | |
| pass | |
| 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 tearDown(self): | |
| unlink(TESTFN) | |
| unlink(TESTFN2) | |
| 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 = ( | |
| 'PK\x03\x04\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00\x1a\x00' | |
| '\x00\x00\x08\x00\x00\x00test.txt\xfa\x10\xa0gly|\xfa-\xc5\xc0=\xf9y' | |
| '\x18\xe0\xa8r\xb3Z}Lg\xbc\xae\xf9|\x9b\x19\xe4\x8b\xba\xbb)\x8c\xb0\xdbl' | |
| 'PK\x01\x02\x14\x00\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00' | |
| '\x1a\x00\x00\x00\x08\x00\x00\x00\x00\x00\x00\x00\x01\x00 \x00\xb6\x81' | |
| '\x00\x00\x00\x00test.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x006\x00' | |
| '\x00\x00L\x00\x00\x00\x00\x00' ) | |
| data2 = ( | |
| 'PK\x03\x04\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02' | |
| '\x00\x00\x04\x00\x15\x00zeroUT\t\x00\x03\xd6\x8b\x92G\xda\x8b\x92GUx\x04' | |
| '\x00\xe8\x03\xe8\x03\xc7<M\xb5a\xceX\xa3Y&\x8b{oE\xd7\x9d\x8c\x98\x02\xc0' | |
| 'PK\x07\x08xu\xaa\xb2\x14\x00\x00\x00\x00\x02\x00\x00PK\x01\x02\x17\x03' | |
| '\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02\x00\x00' | |
| '\x04\x00\r\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa4\x81\x00\x00\x00\x00ze' | |
| 'roUT\x05\x00\x03\xd6\x8b\x92GUx\x00\x00PK\x05\x06\x00\x00\x00\x00\x01' | |
| '\x00\x01\x00?\x00\x00\x00[\x00\x00\x00\x00\x00' ) | |
| plain = 'zipfile.py encryption test' | |
| plain2 = '\x00'*512 | |
| def setUp(self): | |
| with open(TESTFN, "wb") as fp: | |
| fp.write(self.data) | |
| self.zip = zipfile.ZipFile(TESTFN, "r") | |
| with open(TESTFN2, "wb") as fp: | |
| fp.write(self.data2) | |
| self.zip2 = zipfile.ZipFile(TESTFN2, "r") | |
| def tearDown(self): | |
| self.zip.close() | |
| os.unlink(TESTFN) | |
| self.zip2.close() | |
| os.unlink(TESTFN2) | |
| def test_no_password(self): | |
| # Reading the encrypted file without password | |
| # must generate a RunTime exception | |
| self.assertRaises(RuntimeError, self.zip.read, "test.txt") | |
| self.assertRaises(RuntimeError, self.zip2.read, "zero") | |
| def test_bad_password(self): | |
| self.zip.setpassword("perl") | |
| self.assertRaises(RuntimeError, self.zip.read, "test.txt") | |
| self.zip2.setpassword("perl") | |
| self.assertRaises(RuntimeError, self.zip2.read, "zero") | |
| @skipUnless(zlib, "requires zlib") | |
| def test_good_password(self): | |
| self.zip.setpassword("python") | |
| self.assertEqual(self.zip.read("test.txt"), self.plain) | |
| self.zip2.setpassword("12345") | |
| self.assertEqual(self.zip2.read("zero"), self.plain2) | |
| class TestsWithRandomBinaryFiles(unittest.TestCase): | |
| def setUp(self): | |
| datacount = randint(16, 64)*1024 + randint(1, 1024) | |
| self.data = ''.join(struct.pack('<f', random()*randint(-1000, 1000)) | |
| for i in xrange(datacount)) | |
| # Make a source file with some lines | |
| with open(TESTFN, "wb") as fp: | |
| fp.write(self.data) | |
| def tearDown(self): | |
| unlink(TESTFN) | |
| unlink(TESTFN2) | |
| def make_test_archive(self, f, compression): | |
| # Create the ZIP archive | |
| with zipfile.ZipFile(f, "w", compression) as zipfp: | |
| zipfp.write(TESTFN, "another.name") | |
| zipfp.write(TESTFN, TESTFN) | |
| def zip_test(self, f, compression): | |
| self.make_test_archive(f, compression) | |
| # Read the ZIP archive | |
| with zipfile.ZipFile(f, "r", compression) as zipfp: | |
| testdata = zipfp.read(TESTFN) | |
| self.assertEqual(len(testdata), len(self.data)) | |
| self.assertEqual(testdata, self.data) | |
| self.assertEqual(zipfp.read("another.name"), self.data) | |
| def test_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_test(f, zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), io.BytesIO()): | |
| self.zip_test(f, zipfile.ZIP_DEFLATED) | |
| 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) | |
| testdata1 = ''.join(zipdata1) | |
| self.assertEqual(len(testdata1), len(self.data)) | |
| self.assertEqual(testdata1, self.data) | |
| testdata2 = ''.join(zipdata2) | |
| self.assertEqual(len(testdata2), len(self.data)) | |
| self.assertEqual(testdata2, self.data) | |
| def test_open_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_open_test(f, zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_open_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), io.BytesIO()): | |
| self.zip_open_test(f, zipfile.ZIP_DEFLATED) | |
| 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) | |
| testdata = ''.join(zipdata1) | |
| self.assertEqual(len(testdata), len(self.data)) | |
| self.assertEqual(testdata, self.data) | |
| def test_random_open_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.zip_random_open_test(f, zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_random_open_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), io.BytesIO()): | |
| self.zip_random_open_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| class TestsWithMultipleOpens(unittest.TestCase): | |
| @classmethod | |
| def setUpClass(cls): | |
| cls.data1 = b'111' + getrandbytes(10000) | |
| cls.data2 = b'222' + getrandbytes(10000) | |
| def make_test_archive(self, f): | |
| # Create the ZIP archive | |
| with zipfile.ZipFile(f, "w", zipfile.ZIP_DEFLATED) as zipfp: | |
| zipfp.writestr('ones', self.data1) | |
| zipfp.writestr('twos', self.data2) | |
| def test_same_file(self): | |
| # Verify that (when the ZipFile is in control of creating file objects) | |
| # multiple open() calls can be made without interfering with each other. | |
| self.make_test_archive(TESTFN2) | |
| with zipfile.ZipFile(TESTFN2, mode="r") as zipf: | |
| with zipf.open('ones') as zopen1, zipf.open('ones') as zopen2: | |
| data1 = zopen1.read(500) | |
| data2 = zopen2.read(500) | |
| data1 += zopen1.read() | |
| data2 += zopen2.read() | |
| self.assertEqual(data1, data2) | |
| self.assertEqual(data1, self.data1) | |
| def test_different_file(self): | |
| # Verify that (when the ZipFile is in control of creating file objects) | |
| # multiple open() calls can be made without interfering with each other. | |
| self.make_test_archive(TESTFN2) | |
| with zipfile.ZipFile(TESTFN2, mode="r") as zipf: | |
| with zipf.open('ones') as zopen1, zipf.open('twos') as zopen2: | |
| data1 = zopen1.read(500) | |
| data2 = zopen2.read(500) | |
| data1 += zopen1.read() | |
| data2 += zopen2.read() | |
| self.assertEqual(data1, self.data1) | |
| self.assertEqual(data2, self.data2) | |
| def test_interleaved(self): | |
| # Verify that (when the ZipFile is in control of creating file objects) | |
| # multiple open() calls can be made without interfering with each other. | |
| self.make_test_archive(TESTFN2) | |
| with zipfile.ZipFile(TESTFN2, mode="r") as zipf: | |
| with zipf.open('ones') as zopen1: | |
| data1 = zopen1.read(500) | |
| with zipf.open('twos') as zopen2: | |
| data2 = zopen2.read(500) | |
| data1 += zopen1.read() | |
| data2 += zopen2.read() | |
| self.assertEqual(data1, self.data1) | |
| self.assertEqual(data2, self.data2) | |
| def test_read_after_close(self): | |
| self.make_test_archive(TESTFN2) | |
| zopen1 = zopen2 = None | |
| try: | |
| with zipfile.ZipFile(TESTFN2, 'r') as zipf: | |
| zopen1 = zipf.open('ones') | |
| zopen2 = zipf.open('twos') | |
| data1 = zopen1.read(500) | |
| data2 = zopen2.read(500) | |
| data1 += zopen1.read() | |
| data2 += zopen2.read() | |
| finally: | |
| if zopen1: | |
| zopen1.close() | |
| if zopen2: | |
| zopen2.close() | |
| self.assertEqual(data1, self.data1) | |
| self.assertEqual(data2, self.data2) | |
| def test_read_after_write(self): | |
| with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_DEFLATED) as zipf: | |
| zipf.writestr('ones', self.data1) | |
| zipf.writestr('twos', self.data2) | |
| with zipf.open('ones') as zopen1: | |
| data1 = zopen1.read(500) | |
| self.assertEqual(data1, self.data1[:500]) | |
| with zipfile.ZipFile(TESTFN2, 'r') as zipf: | |
| data1 = zipf.read('ones') | |
| data2 = zipf.read('twos') | |
| self.assertEqual(data1, self.data1) | |
| self.assertEqual(data2, self.data2) | |
| def test_write_after_read(self): | |
| with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_DEFLATED) as zipf: | |
| zipf.writestr('ones', self.data1) | |
| with zipf.open('ones') as zopen1: | |
| zopen1.read(500) | |
| zipf.writestr('twos', self.data2) | |
| with zipfile.ZipFile(TESTFN2, 'r') as zipf: | |
| data1 = zipf.read('ones') | |
| data2 = zipf.read('twos') | |
| self.assertEqual(data1, self.data1) | |
| self.assertEqual(data2, self.data2) | |
| def test_many_opens(self): | |
| # Verify that read() and open() promptly close the file descriptor, | |
| # and don't rely on the garbage collector to free resources. | |
| self.make_test_archive(TESTFN2) | |
| with zipfile.ZipFile(TESTFN2, mode="r") as zipf: | |
| for x in range(100): | |
| zipf.read('ones') | |
| with zipf.open('ones') as zopen1: | |
| pass | |
| with open(os.devnull) as f: | |
| self.assertLess(f.fileno(), 100) | |
| def tearDown(self): | |
| unlink(TESTFN2) | |
| class TestWithDirectory(unittest.TestCase): | |
| def setUp(self): | |
| os.mkdir(TESTFN2) | |
| def test_extract_dir(self): | |
| with zipfile.ZipFile(findfile("zipdir.zip")) as zipf: | |
| zipf.extractall(TESTFN2) | |
| self.assertTrue(os.path.isdir(os.path.join(TESTFN2, "a"))) | |
| self.assertTrue(os.path.isdir(os.path.join(TESTFN2, "a", "b"))) | |
| self.assertTrue(os.path.exists(os.path.join(TESTFN2, "a", "b", "c"))) | |
| def test_bug_6050(self): | |
| # Extraction should succeed if directories already exist | |
| os.mkdir(os.path.join(TESTFN2, "a")) | |
| self.test_extract_dir() | |
| def test_write_dir(self): | |
| dirpath = os.path.join(TESTFN2, "x") | |
| os.mkdir(dirpath) | |
| mode = os.stat(dirpath).st_mode & 0xFFFF | |
| with zipfile.ZipFile(TESTFN, "w") as zipf: | |
| zipf.write(dirpath) | |
| zinfo = zipf.filelist[0] | |
| self.assertTrue(zinfo.filename.endswith("/x/")) | |
| self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10) | |
| zipf.write(dirpath, "y") | |
| zinfo = zipf.filelist[1] | |
| self.assertTrue(zinfo.filename, "y/") | |
| self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10) | |
| with zipfile.ZipFile(TESTFN, "r") as zipf: | |
| zinfo = zipf.filelist[0] | |
| self.assertTrue(zinfo.filename.endswith("/x/")) | |
| self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10) | |
| zinfo = zipf.filelist[1] | |
| self.assertTrue(zinfo.filename, "y/") | |
| self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10) | |
| target = os.path.join(TESTFN2, "target") | |
| os.mkdir(target) | |
| zipf.extractall(target) | |
| self.assertTrue(os.path.isdir(os.path.join(target, "y"))) | |
| self.assertEqual(len(os.listdir(target)), 2) | |
| def test_writestr_dir(self): | |
| os.mkdir(os.path.join(TESTFN2, "x")) | |
| with zipfile.ZipFile(TESTFN, "w") as zipf: | |
| zipf.writestr("x/", b'') | |
| zinfo = zipf.filelist[0] | |
| self.assertEqual(zinfo.filename, "x/") | |
| self.assertEqual(zinfo.external_attr, (0o40775 << 16) | 0x10) | |
| with zipfile.ZipFile(TESTFN, "r") as zipf: | |
| zinfo = zipf.filelist[0] | |
| self.assertTrue(zinfo.filename.endswith("x/")) | |
| self.assertEqual(zinfo.external_attr, (0o40775 << 16) | 0x10) | |
| target = os.path.join(TESTFN2, "target") | |
| os.mkdir(target) | |
| zipf.extractall(target) | |
| self.assertTrue(os.path.isdir(os.path.join(target, "x"))) | |
| self.assertEqual(os.listdir(target), ["x"]) | |
| def tearDown(self): | |
| rmtree(TESTFN2) | |
| if os.path.exists(TESTFN): | |
| unlink(TESTFN) | |
| class UniversalNewlineTests(unittest.TestCase): | |
| def setUp(self): | |
| self.line_gen = ["Test of zipfile line %d." % i | |
| for i in xrange(FIXEDTEST_SIZE)] | |
| self.seps = ('\r', '\r\n', '\n') | |
| self.arcdata, self.arcfiles = {}, {} | |
| for n, s in enumerate(self.seps): | |
| self.arcdata[s] = s.join(self.line_gen) + s | |
| self.arcfiles[s] = '%s-%d' % (TESTFN, n) | |
| with open(self.arcfiles[s], "wb") as fid: | |
| fid.write(self.arcdata[s]) | |
| def make_test_archive(self, f, compression): | |
| # Create the ZIP archive | |
| with zipfile.ZipFile(f, "w", compression) as zipfp: | |
| for fn in self.arcfiles.values(): | |
| zipfp.write(fn, fn) | |
| def read_test(self, f, compression): | |
| self.make_test_archive(f, compression) | |
| # Read the ZIP archive | |
| with zipfile.ZipFile(f, "r") as zipfp: | |
| for sep, fn in self.arcfiles.items(): | |
| with zipfp.open(fn, "rU") as fp: | |
| zipdata = fp.read() | |
| self.assertEqual(self.arcdata[sep], zipdata) | |
| def readline_read_test(self, f, compression): | |
| self.make_test_archive(f, compression) | |
| # Read the ZIP archive | |
| zipfp = zipfile.ZipFile(f, "r") | |
| for sep, fn in self.arcfiles.items(): | |
| with zipfp.open(fn, "rU") as zipopen: | |
| data = '' | |
| while True: | |
| read = zipopen.readline() | |
| if not read: | |
| break | |
| data += read | |
| read = zipopen.read(5) | |
| if not read: | |
| break | |
| data += read | |
| self.assertEqual(data, self.arcdata['\n']) | |
| zipfp.close() | |
| def readline_test(self, f, compression): | |
| self.make_test_archive(f, compression) | |
| # Read the ZIP archive | |
| with zipfile.ZipFile(f, "r") as zipfp: | |
| for sep, fn in self.arcfiles.items(): | |
| with zipfp.open(fn, "rU") as zipopen: | |
| for line in self.line_gen: | |
| linedata = zipopen.readline() | |
| self.assertEqual(linedata, line + '\n') | |
| def readlines_test(self, f, compression): | |
| self.make_test_archive(f, compression) | |
| # Read the ZIP archive | |
| with zipfile.ZipFile(f, "r") as zipfp: | |
| for sep, fn in self.arcfiles.items(): | |
| with zipfp.open(fn, "rU") as fp: | |
| ziplines = fp.readlines() | |
| for line, zipline in zip(self.line_gen, ziplines): | |
| self.assertEqual(zipline, line + '\n') | |
| def iterlines_test(self, f, compression): | |
| self.make_test_archive(f, compression) | |
| # Read the ZIP archive | |
| with zipfile.ZipFile(f, "r") as zipfp: | |
| for sep, fn in self.arcfiles.items(): | |
| with zipfp.open(fn, "rU") as fid: | |
| for line, zipline in zip(self.line_gen, fid): | |
| self.assertEqual(zipline, line + '\n') | |
| def test_read_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.read_test(f, zipfile.ZIP_STORED) | |
| def test_readline_read_stored(self): | |
| # Issue #7610: calls to readline() interleaved with calls to read(). | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.readline_read_test(f, zipfile.ZIP_STORED) | |
| def test_readline_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.readline_test(f, zipfile.ZIP_STORED) | |
| def test_readlines_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.readlines_test(f, zipfile.ZIP_STORED) | |
| def test_iterlines_stored(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.iterlines_test(f, zipfile.ZIP_STORED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_read_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.read_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_readline_read_deflated(self): | |
| # Issue #7610: calls to readline() interleaved with calls to read(). | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.readline_read_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_readline_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.readline_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_readlines_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.readlines_test(f, zipfile.ZIP_DEFLATED) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_iterlines_deflated(self): | |
| for f in (TESTFN2, TemporaryFile(), StringIO()): | |
| self.iterlines_test(f, zipfile.ZIP_DEFLATED) | |
| def tearDown(self): | |
| for sep, fn in self.arcfiles.items(): | |
| os.remove(fn) | |
| unlink(TESTFN) | |
| unlink(TESTFN2) | |
| class CommandLineTest(unittest.TestCase): | |
| def zipfilecmd(self, *args, **kwargs): | |
| rc, out, err = script_helper.assert_python_ok('-m', 'zipfile', *args, | |
| **kwargs) | |
| return out.replace(os.linesep.encode(), b'\n') | |
| def zipfilecmd_failure(self, *args): | |
| return script_helper.assert_python_failure('-m', 'zipfile', *args) | |
| def test_test_command(self): | |
| zip_name = findfile('zipdir.zip') | |
| out = self.zipfilecmd('-t', zip_name) | |
| self.assertEqual(out.rstrip(), b'Done testing') | |
| zip_name = findfile('testtar.tar') | |
| rc, out, err = self.zipfilecmd_failure('-t', zip_name) | |
| self.assertEqual(out, b'') | |
| def test_list_command(self): | |
| zip_name = findfile('zipdir.zip') | |
| with captured_stdout() as t, zipfile.ZipFile(zip_name, 'r') as tf: | |
| tf.printdir() | |
| expected = t.getvalue().encode('ascii', 'backslashreplace') | |
| out = self.zipfilecmd('-l', zip_name, | |
| PYTHONIOENCODING='ascii:backslashreplace') | |
| self.assertEqual(out, expected) | |
| @skipUnless(zlib, "requires zlib") | |
| def test_create_command(self): | |
| self.addCleanup(unlink, TESTFN) | |
| with open(TESTFN, 'w') as f: | |
| f.write('test 1') | |
| os.mkdir(TESTFNDIR) | |
| self.addCleanup(rmtree, TESTFNDIR) | |
| with open(os.path.join(TESTFNDIR, 'file.txt'), 'w') as f: | |
| f.write('test 2') | |
| files = [TESTFN, TESTFNDIR] | |
| namelist = [TESTFN, TESTFNDIR + '/', TESTFNDIR + '/file.txt'] | |
| try: | |
| out = self.zipfilecmd('-c', TESTFN2, *files) | |
| self.assertEqual(out, b'') | |
| with zipfile.ZipFile(TESTFN2) as zf: | |
| self.assertEqual(zf.namelist(), namelist) | |
| self.assertEqual(zf.read(namelist[0]), b'test 1') | |
| self.assertEqual(zf.read(namelist[2]), b'test 2') | |
| finally: | |
| unlink(TESTFN2) | |
| def test_extract_command(self): | |
| zip_name = findfile('zipdir.zip') | |
| extdir = TESTFNDIR | |
| os.mkdir(extdir) | |
| try: | |
| out = self.zipfilecmd('-e', zip_name, extdir) | |
| self.assertEqual(out, b'') | |
| with zipfile.ZipFile(zip_name) as zf: | |
| for zi in zf.infolist(): | |
| path = os.path.join(extdir, | |
| zi.filename.replace('/', os.sep)) | |
| if zi.filename.endswith('/'): | |
| self.assertTrue(os.path.isdir(path)) | |
| else: | |
| self.assertTrue(os.path.isfile(path)) | |
| with open(path, 'rb') as f: | |
| self.assertEqual(f.read(), zf.read(zi)) | |
| finally: | |
| rmtree(extdir) | |
| def test_main(): | |
| run_unittest(TestsWithSourceFile, TestZip64InSmallFiles, OtherTests, | |
| PyZipFileTests, DecryptionTests, TestsWithMultipleOpens, | |
| TestWithDirectory, UniversalNewlineTests, | |
| TestsWithRandomBinaryFiles, CommandLineTest) | |
| if __name__ == "__main__": | |
| test_main() |