diff options
Diffstat (limited to 'Lib/test/test_zipfile.py')
| -rw-r--r-- | Lib/test/test_zipfile.py | 1805 |
1 files changed, 1011 insertions, 794 deletions
diff --git a/Lib/test/test_zipfile.py b/Lib/test/test_zipfile.py index ee7524e..e07380d 100644 --- a/Lib/test/test_zipfile.py +++ b/Lib/test/test_zipfile.py @@ -3,19 +3,23 @@ try: import zlib except ImportError: zlib = None + import io import os +import sys +import imp +import time import shutil import struct -import sys -import unittest import zipfile +import unittest + from tempfile import TemporaryFile from random import randint, random +from unittest import skipUnless -import test.support as support -from test.support import TESTFN, run_unittest, findfile +from test.support import TESTFN, run_unittest, findfile, unlink TESTFN2 = TESTFN + "2" TESTFNDIR = TESTFN + "d" @@ -27,6 +31,7 @@ SMALL_TEST_DATA = [('_ziptest1', '1q2w3e4r5t'), ('/ziptest2dir/ziptest3dir/_ziptest3', 'azsxdcfvgb'), ('ziptest2dir/ziptest3dir/ziptest4dir/_ziptest3', '6y7u8i9o0p')] + class TestsWithSourceFile(unittest.TestCase): def setUp(self): self.line_gen = (bytes("Zipfile test line %d. random float: %f" % @@ -35,384 +40,492 @@ class TestsWithSourceFile(unittest.TestCase): self.data = b'\n'.join(self.line_gen) + b'\n' # Make a source file with some lines - fp = open(TESTFN, "wb") - fp.write(self.data) - fp.close() + with open(TESTFN, "wb") as fp: + fp.write(self.data) - def makeTestArchive(self, f, compression): + def make_test_archive(self, f, compression): # Create the ZIP archive - zipfp = zipfile.ZipFile(f, "w", compression) - zipfp.write(TESTFN, "another.name") - zipfp.write(TESTFN, TESTFN) - zipfp.writestr("strfile", self.data) - zipfp.close() + with zipfile.ZipFile(f, "w", compression) as zipfp: + zipfp.write(TESTFN, "another.name") + zipfp.write(TESTFN, TESTFN) + zipfp.writestr("strfile", self.data) - def zipTest(self, f, compression): - self.makeTestArchive(f, compression) + def zip_test(self, f, compression): + self.make_test_archive(f, compression) # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r", compression) - 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), 4) # Number of files + header - - self.assertTrue('File Name' in lines[0]) - self.assertTrue('Modified' in lines[0]) - self.assertTrue('Size' in lines[0]) - - fn, date, time, size = lines[1].split() - self.assertEqual(fn, 'another.name') - # XXX: timestamp is not tested - self.assertEqual(size, str(len(self.data))) - - # Check the namelist - names = zipfp.namelist() - self.assertEqual(len(names), 3) - self.assertTrue(TESTFN in names) - self.assertTrue("another.name" in names) - self.assertTrue("strfile" in names) - - # Check infolist - infos = zipfp.infolist() - names = [ i.filename for i in infos ] - self.assertEqual(len(names), 3) - self.assertTrue(TESTFN in names) - self.assertTrue("another.name" in names) - self.assertTrue("strfile" in 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() - zipfp.close() - - def testStored(self): + 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), 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() + if not isinstance(f, str): + f.close() + + def test_stored(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipTest(f, zipfile.ZIP_STORED) + self.zip_test(f, zipfile.ZIP_STORED) - def zipOpenTest(self, f, compression): - self.makeTestArchive(f, compression) + def zip_open_test(self, f, compression): + self.make_test_archive(f, compression) # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r", compression) - zipdata1 = [] - zipopen1 = zipfp.open(TESTFN) - while 1: - read_data = zipopen1.read(256) - if not read_data: - break - zipdata1.append(read_data) - - zipdata2 = [] - zipopen2 = zipfp.open("another.name") - while 1: - 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) - zipfp.close() - - def testOpenStored(self): + 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) + if not isinstance(f, str): + f.close() + + def test_open_stored(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipOpenTest(f, zipfile.ZIP_STORED) + self.zip_open_test(f, zipfile.ZIP_STORED) - def testOpenViaZipInfo(self): + def test_open_via_zip_info(self): # Create the ZIP archive - zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) - zipfp.writestr("name", "foo") - zipfp.writestr("name", "bar") - zipfp.close() + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: + zipfp.writestr("name", "foo") + zipfp.writestr("name", "bar") + + 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.assertTrue(data == b"foobar" or data == b"barfoo") + data = b"" + for info in infos: + data += zipfp.read(info) + self.assertTrue(data == b"foobar" or data == b"barfoo") + + def zip_random_open_test(self, f, compression): + self.make_test_archive(f, compression) - zipfp = zipfile.ZipFile(TESTFN2, "r") - infos = zipfp.infolist() - data = b"" - for info in infos: - data += zipfp.open(info).read() - self.assertTrue(data == b"foobar" or data == b"barfoo") - data = b"" - for info in infos: - data += zipfp.read(info) - self.assertTrue(data == b"foobar" or data == b"barfoo") - zipfp.close() + # 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) + if not isinstance(f, str): + f.close() + + def test_random_open_stored(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_random_open_test(f, zipfile.ZIP_STORED) - def zipRandomOpenTest(self, f, compression): - self.makeTestArchive(f, compression) + def test_univeral_readaheads(self): + f = io.BytesIO() - # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r", compression) - zipdata1 = [] - zipopen1 = zipfp.open(TESTFN) - while 1: - read_data = zipopen1.read(randint(1, 1024)) - if not read_data: - break - zipdata1.append(read_data) - - self.assertEqual(b''.join(zipdata1), self.data) + data = b'a\r\n' * 16 * 1024 + zipfp = zipfile.ZipFile(f, 'w', zipfile.ZIP_STORED) + zipfp.writestr(TESTFN, data) zipfp.close() - def testRandomOpenStored(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipRandomOpenTest(f, zipfile.ZIP_STORED) + data2 = b'' + zipfp = zipfile.ZipFile(f, 'r') + with zipfp.open(TESTFN, 'rU') as zipopen: + for line in zipopen: + data2 += line + zipfp.close() + + self.assertEqual(data, data2.replace(b'\n', b'\r\n')) - def zipReadlineTest(self, f, compression): - self.makeTestArchive(f, compression) + def zip_readline_read_test(self, f, compression): + self.make_test_archive(f, compression) # Read the ZIP archive zipfp = zipfile.ZipFile(f, "r") - zipopen = zipfp.open(TESTFN) - for line in self.line_gen: - linedata = zipopen.readline() - self.assertEqual(linedata, line + '\n') - + with 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) zipfp.close() + if not isinstance(f, str): + f.close() - def zipReadlinesTest(self, f, compression): - self.makeTestArchive(f, compression) + def zip_readline_test(self, f, compression): + self.make_test_archive(f, compression) # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r") - ziplines = zipfp.open(TESTFN).readlines() - for line, zipline in zip(self.line_gen, ziplines): - self.assertEqual(zipline, line + '\n') - - zipfp.close() + 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') + if not isinstance(f, str): + f.close() - def zipIterlinesTest(self, f, compression): - self.makeTestArchive(f, compression) + def zip_readlines_test(self, f, compression): + self.make_test_archive(f, compression) # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r") - for line, zipline in zip(self.line_gen, zipfp.open(TESTFN)): - self.assertEqual(zipline, line + '\n') + 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 + '\n') + if not isinstance(f, str): + f.close() - zipfp.close() + def zip_iterlines_test(self, f, compression): + self.make_test_archive(f, compression) - def testReadlineStored(self): + # 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 + '\n') + if not isinstance(f, str): + f.close() + + def test_readline_read_stored(self): + # Issue #7610: calls to readline() interleaved with calls to read(). for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipReadlineTest(f, zipfile.ZIP_STORED) + self.zip_readline_read_test(f, zipfile.ZIP_STORED) - def testReadlinesStored(self): + def test_readline_stored(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipReadlinesTest(f, zipfile.ZIP_STORED) + self.zip_readline_test(f, zipfile.ZIP_STORED) - def testIterlinesStored(self): + def test_readlines_stored(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipIterlinesTest(f, zipfile.ZIP_STORED) - - if zlib: - def testDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipTest(f, zipfile.ZIP_DEFLATED) - - def testOpenDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipOpenTest(f, zipfile.ZIP_DEFLATED) + self.zip_readlines_test(f, zipfile.ZIP_STORED) - def testRandomOpenDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipRandomOpenTest(f, zipfile.ZIP_DEFLATED) + def test_iterlines_stored(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_iterlines_test(f, zipfile.ZIP_STORED) - def testReadlineDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipReadlineTest(f, zipfile.ZIP_DEFLATED) + @skipUnless(zlib, "requires zlib") + def test_deflated(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_test(f, zipfile.ZIP_DEFLATED) - def testReadlinesDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipReadlinesTest(f, zipfile.ZIP_DEFLATED) - def testIterlinesDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipIterlinesTest(f, zipfile.ZIP_DEFLATED) + @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 testLowCompression(self): - # Checks for cases where compressed data is larger than original - # Create the ZIP archive - zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_DEFLATED) - zipfp.writestr("strfile", '12') - zipfp.close() + @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) - # Get an open object for strfile - zipfp = zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_DEFLATED) - openobj = zipfp.open("strfile") - self.assertEqual(openobj.read(1), b'1') - self.assertEqual(openobj.read(1), b'2') + @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(), io.BytesIO()): + self.zip_readline_read_test(f, zipfile.ZIP_DEFLATED) - def testAbsoluteArcnames(self): - zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) - zipfp.write(TESTFN, "/absolute") - zipfp.close() + @skipUnless(zlib, "requires zlib") + def test_readline_deflated(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_readline_test(f, zipfile.ZIP_DEFLATED) - zipfp = zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) - self.assertEqual(zipfp.namelist(), ["absolute"]) - zipfp.close() + @skipUnless(zlib, "requires zlib") + def test_readlines_deflated(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_readlines_test(f, zipfile.ZIP_DEFLATED) - def testAppendToZipFile(self): - # Test appending to an existing zipfile - zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) - zipfp.write(TESTFN, TESTFN) - zipfp.close() - zipfp = zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) - zipfp.writestr("strfile", self.data) - self.assertEqual(zipfp.namelist(), [TESTFN, "strfile"]) - zipfp.close() + @skipUnless(zlib, "requires zlib") + def test_iterlines_deflated(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_iterlines_test(f, zipfile.ZIP_DEFLATED) - def testAppendToNonZipFile(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 - d = b'I am not a ZipFile!'*10 - f = open(TESTFN2, 'wb') - f.write(d) - f.close() - zipfp = zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) - zipfp.write(TESTFN, TESTFN) - zipfp.close() + @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') - f = open(TESTFN2, 'rb') - f.seek(len(d)) - zipfp = zipfile.ZipFile(f, "r") - self.assertEqual(zipfp.namelist(), [TESTFN]) - zipfp.close() - f.close() + # 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), b'1') + self.assertEqual(openobj.read(1), b'2') - def test_WriteDefaultName(self): - # Check that calling ZipFile.write without arcname specified produces the expected result - zipfp = zipfile.ZipFile(TESTFN2, "w") - zipfp.write(TESTFN) - self.assertEqual(zipfp.read(TESTFN), open(TESTFN, "rb").read()) - zipfp.close() + def test_absolute_arcnames(self): + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: + zipfp.write(TESTFN, "/absolute") - def test_PerFileCompression(self): - # Check that files within a Zip archive can have different compression options - zipfp = zipfile.ZipFile(TESTFN2, "w") - 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) - zipfp.close() + with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: + self.assertEqual(zipfp.namelist(), ["absolute"]) - def test_WriteToReadonly(self): - # Check that trying to call write() on a readonly ZipFile object - # raises a RuntimeError - zipf = zipfile.ZipFile(TESTFN2, mode="w") - zipf.writestr("somefile.txt", "bogus") - zipf.close() - zipf = zipfile.ZipFile(TESTFN2, mode="r") - self.assertRaises(RuntimeError, zipf.write, TESTFN) - zipf.close() + 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) - def testExtract(self): - zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) - for fpath, fdata in SMALL_TEST_DATA: - zipfp.writestr(fpath, fdata) - zipfp.close() + with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp: + zipfp.writestr("strfile", self.data) + self.assertEqual(zipfp.namelist(), [TESTFN, "strfile"]) - zipfp = zipfile.ZipFile(TESTFN2, "r") - for fpath, fdata in SMALL_TEST_DATA: - writtenfile = zipfp.extract(fpath) + 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]) + + 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()) + + @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 + if os.path.isabs(fpath): + correctfile = os.path.join(os.getcwd(), fpath[1:]) + else: + 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 f: + self.assertEqual(fdata.encode(), f.read()) + + os.remove(writtenfile) - # make sure it was written to the right place - if os.path.isabs(fpath): - correctfile = os.path.join(os.getcwd(), fpath[1:]) - else: - correctfile = os.path.join(os.getcwd(), fpath) - correctfile = os.path.normpath(correctfile) + # remove the test file subdirectories + shutil.rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) - self.assertEqual(writtenfile, correctfile) + 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) - # make sure correct data is in correct file - self.assertEqual(fdata.encode(), open(writtenfile, "rb").read()) + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + zipfp.extractall() + for fpath, fdata in SMALL_TEST_DATA: + if os.path.isabs(fpath): + outfile = os.path.join(os.getcwd(), fpath[1:]) + else: + outfile = os.path.join(os.getcwd(), fpath) - os.remove(writtenfile) + with open(outfile, "rb") as f: + self.assertEqual(fdata.encode(), f.read()) - zipfp.close() + os.remove(outfile) # remove the test file subdirectories shutil.rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) - def testExtractAll(self): - zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) - for fpath, fdata in SMALL_TEST_DATA: - zipfp.writestr(fpath, fdata) - zipfp.close() - - zipfp = zipfile.ZipFile(TESTFN2, "r") - zipfp.extractall() - for fpath, fdata in SMALL_TEST_DATA: - if os.path.isabs(fpath): - outfile = os.path.join(os.getcwd(), fpath[1:]) - else: - outfile = os.path.join(os.getcwd(), fpath) - - self.assertEqual(fdata.encode(), open(outfile, "rb").read()) + 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) - os.remove(outfile) + info = zipfp.getinfo('a.txt') + self.assertEqual(info.compress_type, zipfile.ZIP_STORED) - zipfp.close() + if zlib: + info = zipfp.getinfo('b.txt') + self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED) - # remove the test file subdirectories - shutil.rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) 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.makeTestArchive(f, compression) - zipfp = zipfile.ZipFile(f, "r") - zinfo = zipfp.getinfo('strfile') - self.assertEqual(zinfo.external_attr, 0o600 << 16) + self.make_test_archive(f, compression) + with zipfile.ZipFile(f, "r") as zipfp: + zinfo = zipfp.getinfo('strfile') + self.assertEqual(zinfo.external_attr, 0o600 << 16) + if not isinstance(f, str): + f.close() def test_writestr_permissions(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): self.zip_test_writestr_permissions(f, zipfile.ZIP_STORED) def test_writestr_extended_local_header_issue1202(self): - orig_zip = zipfile.ZipFile(TESTFN2, 'w') - for data in 'abcdefghijklmnop': - zinfo = zipfile.ZipInfo(data) - zinfo.flag_bits |= 0x08 # Include an extended local header. - orig_zip.writestr(zinfo, data) - orig_zip.close() + 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.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) - def test_unicode_filenames(self): - if __name__ == '__main__': - myfile = sys.argv[0] - else: - myfile = __file__ + try: + with zipfile.ZipFile(TESTFN2, "r") as zipfp2: + raise zipfile.BadZipFile() + except zipfile.BadZipFile: + self.assertTrue(zipfp2.fp is None, 'zipfp is not closed') - mydir = os.path.dirname(myfile) or os.curdir - fname = os.path.join(mydir, 'zip_cp437_header.zip') + 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) - zipfp = zipfile.ZipFile(fname) - try: + + @skipUnless(zlib, "requires zlib") + def test_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() - finally: - zipfp.close() def tearDown(self): - support.unlink(TESTFN) - support.unlink(TESTFN2) + unlink(TESTFN) + unlink(TESTFN2) + class TestZip64InSmallFiles(unittest.TestCase): # These tests test the ZIP64 functionality without using large files, @@ -427,181 +540,200 @@ class TestZip64InSmallFiles(unittest.TestCase): self.data = b'\n'.join(line_gen) # Make a source file with some lines - fp = open(TESTFN, "wb") - fp.write(self.data) - fp.close() - - def largeFileExceptionTest(self, f, compression): - zipfp = zipfile.ZipFile(f, "w", compression) - self.assertRaises(zipfile.LargeZipFile, - zipfp.write, TESTFN, "another.name") - zipfp.close() - - def largeFileExceptionTest2(self, f, compression): - zipfp = zipfile.ZipFile(f, "w", compression) - self.assertRaises(zipfile.LargeZipFile, - zipfp.writestr, "another.name", self.data) - zipfp.close() - - def testLargeFileException(self): + 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) + if not isinstance(f, str): + f.close() + + def test_large_file_exception(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.largeFileExceptionTest(f, zipfile.ZIP_STORED) - self.largeFileExceptionTest2(f, zipfile.ZIP_STORED) + self.large_file_exception_test(f, zipfile.ZIP_STORED) + self.large_file_exception_test2(f, zipfile.ZIP_STORED) - def zipTest(self, f, compression): + def zip_test(self, f, compression): # Create the ZIP archive - zipfp = zipfile.ZipFile(f, "w", compression, allowZip64=True) - zipfp.write(TESTFN, "another.name") - zipfp.write(TESTFN, TESTFN) - zipfp.writestr("strfile", self.data) - zipfp.close() + 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 - zipfp = zipfile.ZipFile(f, "r", compression) - 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.assertTrue('File Name' in lines[0]) - self.assertTrue('Modified' in lines[0]) - self.assertTrue('Size' in lines[0]) - - fn, date, time, size = lines[1].split() - self.assertEqual(fn, 'another.name') - # XXX: timestamp is not tested - self.assertEqual(size, str(len(self.data))) - - # Check the namelist - names = zipfp.namelist() - self.assertEqual(len(names), 3) - self.assertTrue(TESTFN in names) - self.assertTrue("another.name" in names) - self.assertTrue("strfile" in names) - - # Check infolist - infos = zipfp.infolist() - names = [ i.filename for i in infos ] - self.assertEqual(len(names), 3) - self.assertTrue(TESTFN in names) - self.assertTrue("another.name" in names) - self.assertTrue("strfile" in 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() - - - zipfp.close() - - def testStored(self): + 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() + if not isinstance(f, str): + f.close() + + def test_stored(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipTest(f, zipfile.ZIP_STORED) + 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) - if zlib: - def testDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipTest(f, zipfile.ZIP_DEFLATED) - - def testAbsoluteArcnames(self): - zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED, allowZip64=True) - zipfp.write(TESTFN, "/absolute") - zipfp.close() + def test_absolute_arcnames(self): + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED, + allowZip64=True) as zipfp: + zipfp.write(TESTFN, "/absolute") - zipfp = zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) - self.assertEqual(zipfp.namelist(), ["absolute"]) - zipfp.close() + with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: + self.assertEqual(zipfp.namelist(), ["absolute"]) def tearDown(self): zipfile.ZIP64_LIMIT = self._limit - os.remove(TESTFN) - os.remove(TESTFN2) - -class PyZipFileTests(unittest.TestCase): - def testWritePyfile(self): - zipfp = zipfile.PyZipFile(TemporaryFile(), "w") - fn = __file__ - if fn.endswith('.pyc') or fn.endswith('.pyo'): - fn = fn[:-1] + unlink(TESTFN) + unlink(TESTFN2) - zipfp.writepy(fn) - bn = os.path.basename(fn) - self.assertTrue(bn not in zipfp.namelist()) - self.assertTrue(bn + 'o' in zipfp.namelist() or bn + 'c' in zipfp.namelist()) - zipfp.close() - - - zipfp = zipfile.PyZipFile(TemporaryFile(), "w") - fn = __file__ - if fn.endswith('.pyc') or fn.endswith('.pyo'): - fn = fn[:-1] +class PyZipFileTests(unittest.TestCase): + def test_write_pyfile(self): + with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: + fn = __file__ + if fn.endswith('.pyc') or fn.endswith('.pyo'): + path_split = fn.split(os.sep) + if os.altsep is not None: + path_split.extend(fn.split(os.altsep)) + if '__pycache__' in path_split: + fn = imp.source_from_cache(fn) + else: + 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 TemporaryFile() as t, zipfile.PyZipFile(t, "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__) - zipfp.writepy(fn, "testpackage") + with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: + zipfp.writepy(packagedir) - bn = "%s/%s"%("testpackage", os.path.basename(fn)) - self.assertTrue(bn not in zipfp.namelist()) - self.assertTrue(bn + 'o' in zipfp.namelist() or bn + 'c' in zipfp.namelist()) - zipfp.close() + # 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 testWritePythonPackage(self): + def test_write_with_optimization(self): import email packagedir = os.path.dirname(email.__file__) + # use .pyc if running test in optimization mode, + # use .pyo if running test in debug mode + optlevel = 1 if __debug__ else 0 + ext = '.pyo' if optlevel == 1 else '.pyc' - zipfp = zipfile.PyZipFile(TemporaryFile(), "w") - zipfp.writepy(packagedir) + with TemporaryFile() as t, \ + zipfile.PyZipFile(t, "w", optimize=optlevel) as zipfp: + zipfp.writepy(packagedir) - # Check for a couple of modules at different levels of the hieararchy - 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) + names = zipfp.namelist() + self.assertIn('email/__init__' + ext, names) + self.assertIn('email/mime/text' + ext, names) - def testWritePythonDirectory(self): + def test_write_python_directory(self): os.mkdir(TESTFN2) try: - fp = open(os.path.join(TESTFN2, "mod1.py"), "w") - fp.write("print(42)\n") - fp.close() + with open(os.path.join(TESTFN2, "mod1.py"), "w") as fp: + fp.write("print(42)\n") - fp = open(os.path.join(TESTFN2, "mod2.py"), "w") - fp.write("print(42 * 42)\n") - fp.close() + with open(os.path.join(TESTFN2, "mod2.py"), "w") as fp: + fp.write("print(42 * 42)\n") - fp = open(os.path.join(TESTFN2, "mod2.txt"), "w") - fp.write("bla bla bla\n") - fp.close() + 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) + with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: + 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.assertTrue('mod2.txt' not in names) + 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: shutil.rmtree(TESTFN2) - def testWriteNonPyfile(self): - zipfp = zipfile.PyZipFile(TemporaryFile(), "w") - open(TESTFN, 'w').write('most definitely not a python file') - self.assertRaises(RuntimeError, zipfp.writepy, TESTFN) - os.remove(TESTFN) + def test_write_non_pyfile(self): + with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: + with open(TESTFN, 'w') as f: + f.write('most definitely not a python file') + self.assertRaises(RuntimeError, zipfp.writepy, TESTFN) + os.remove(TESTFN) class OtherTests(unittest.TestCase): @@ -626,17 +758,17 @@ class OtherTests(unittest.TestCase): b'\x01\x003\x00\x00\x003\x00\x00\x00\x00\x00'), } - def testUnicodeFilenames(self): - zf = zipfile.ZipFile(TESTFN, "w") - zf.writestr("foo.txt", "Test for unicode filename") - zf.writestr("\xf6.txt", "Test for unicode filename") - zf.close() - zf = zipfile.ZipFile(TESTFN, "r") - self.assertEqual(zf.filelist[0].filename, "foo.txt") - self.assertEqual(zf.filelist[1].filename, "\xf6.txt") - zf.close() - - def testCreateNonExistentFileForAppend(self): + def test_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") + + def test_create_non_existent_file_for_append(self): if os.path.exists(TESTFN): os.unlink(TESTFN) @@ -644,42 +776,38 @@ class OtherTests(unittest.TestCase): content = b'hello, world. this is some content.' try: - zf = zipfile.ZipFile(TESTFN, 'a') - zf.writestr(filename, content) - zf.close() + 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)) - zf = zipfile.ZipFile(TESTFN, 'r') - self.assertEqual(zf.read(filename), content) - zf.close() + with zipfile.ZipFile(TESTFN, 'r') as zf: + self.assertEqual(zf.read(filename), content) - def testCloseErroneousFile(self): + 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. + # 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. # - fp = open(TESTFN, "w") - fp.write("this is not a legal zip file\n") - fp.close() + with open(TESTFN, "w") as fp: + fp.write("this is not a legal zip file\n") try: zf = zipfile.ZipFile(TESTFN) - except zipfile.BadZipfile: + except zipfile.BadZipFile: pass - def testIsZipErroneousFile(self): - # This test checks that the is_zipfile function correctly identifies - # a file that is not a zip file - + 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.assertTrue(not chk) + self.assertFalse(chk) # - passing a file object with open(TESTFN, "rb") as fp: chk = zipfile.is_zipfile(fp) @@ -689,36 +817,34 @@ class OtherTests(unittest.TestCase): fp.write(b"this is not a legal zip file\n") chk = zipfile.is_zipfile(fp) self.assertTrue(not chk) - fp.seek(0,0) + fp.seek(0, 0) chk = zipfile.is_zipfile(fp) self.assertTrue(not chk) - def testIsZipValidFile(self): - # This test checks that the is_zipfile function correctly identifies - # a file that is a zip file - + def test_is_zip_valid_file(self): + """Check that is_zipfile() correctly identifies zip files.""" # - passing a filename - zipf = zipfile.ZipFile(TESTFN, mode="w") - zipf.writestr("foo.txt", b"O, for a Muse of Fire!") - zipf.close() + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + zipf.writestr("foo.txt", b"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) + fp.seek(0, 0) zip_contents = fp.read() # - passing a file-like object fp = io.BytesIO() fp.write(zip_contents) chk = zipfile.is_zipfile(fp) self.assertTrue(chk) - fp.seek(0,0) + fp.seek(0, 0) chk = zipfile.is_zipfile(fp) self.assertTrue(chk) - def testNonExistentFileRaisesIOError(self): + 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. @@ -735,21 +861,19 @@ class OtherTests(unittest.TestCase): def test_empty_file_raises_BadZipFile(self): f = open(TESTFN, 'w') f.close() - self.assertRaises(zipfile.BadZipfile, zipfile.ZipFile, TESTFN) + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) - f = open(TESTFN, 'w') - f.write("short file") - 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 testClosedZipRaisesRuntimeError(self): - # Verify that testzip() doesn't swallow inappropriate exceptions. + def test_closed_zip_raises_RuntimeError(self): + """Verify that testzip() doesn't swallow inappropriate exceptions.""" data = io.BytesIO() - zipf = zipfile.ZipFile(data, mode="w") - zipf.writestr("foo.txt", "O, for a Muse of Fire!") - zipf.close() + 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 throw + # 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. @@ -757,150 +881,190 @@ class OtherTests(unittest.TestCase): self.assertRaises(RuntimeError, zipf.open, "foo.txt") self.assertRaises(RuntimeError, zipf.testzip) self.assertRaises(RuntimeError, zipf.writestr, "bogus.txt", "bogus") - open(TESTFN, 'w').write('zipfile test data') + with open(TESTFN, 'w') as f: + f.write('zipfile test data') self.assertRaises(RuntimeError, zipf.write, TESTFN) - def test_BadConstructorMode(self): - # Check that bad modes passed to ZipFile constructor are caught + 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_BadOpenMode(self): - # Check that bad modes passed to ZipFile.open are caught - zipf = zipfile.ZipFile(TESTFN, mode="w") - zipf.writestr("foo.txt", "O, for a Muse of Fire!") - zipf.close() - zipf = zipfile.ZipFile(TESTFN, mode="r") - # read the data to make sure the file is there - zipf.read("foo.txt") - self.assertRaises(RuntimeError, zipf.open, "foo.txt", "q") - zipf.close() + 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!") - def test_Read0(self): - # Check that calling read(0) on a ZipExtFile object returns an empty - # string and doesn't advance file pointer - zipf = zipfile.ZipFile(TESTFN, mode="w") - 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 - f = zipf.open("foo.txt") - for i in range(FIXEDTEST_SIZE): - self.assertEqual(f.read(0), b'') - - self.assertEqual(f.read(), b"O, for a Muse of Fire!") - zipf.close() - - def test_OpenNonexistentItem(self): - # Check that attempting to call open() for an item that doesn't - # exist in the archive raises a RuntimeError - zipf = zipfile.ZipFile(TESTFN, mode="w") - self.assertRaises(KeyError, zipf.open, "foo.txt", "r") - - def test_BadCompressionMode(self): - # Check that bad compression methods passed to ZipFile.open are caught + 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 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(RuntimeError, zipfile.ZipFile, TESTFN, "w", -1) - def test_NullByteInFilename(self): - # Check that a filename containing a null byte is properly terminated - zipf = zipfile.ZipFile(TESTFN, mode="w") - zipf.writestr("foo.txt\x00qqq", b"O, for a Muse of Fire!") - self.assertEqual(zipf.namelist(), ['foo.txt']) - - def test_StructSizes(self): - # check that ZIP internal structure sizes are calculated correctly + 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) - def testComments(self): - # This test checks that comments on the archive are handled properly + def test_comments(self): + """Check that comments on the archive are handled properly.""" # check default comment is empty - zipf = zipfile.ZipFile(TESTFN, mode="w") - self.assertEqual(zipf.comment, b'') - zipf.writestr("foo.txt", "O, for a Muse of Fire!") - zipf.close() - zipfr = zipfile.ZipFile(TESTFN, mode="r") - self.assertEqual(zipfr.comment, b'') - zipfr.close() + 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.' - zipf = zipfile.ZipFile(TESTFN, mode="w") - zipf.comment = comment - zipf.writestr("foo.txt", "O, for a Muse of Fire!") - zipf.close() - zipfr = zipfile.ZipFile(TESTFN, mode="r") - self.assertEqual(zipfr.comment, comment) - zipfr.close() + 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") - zipf = zipfile.ZipFile(TESTFN, mode="w") - zipf.comment = comment2 - zipf.writestr("foo.txt", "O, for a Muse of Fire!") - zipf.close() - zipfr = zipfile.ZipFile(TESTFN, mode="r") - self.assertEqual(zipfr.comment, comment2) - zipfr.close() + 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 - zipf = zipfile.ZipFile(TESTFN, mode="w") - zipf.comment = comment2 + b'oops' - zipf.writestr("foo.txt", "O, for a Muse of Fire!") - zipf.close() - zipfr = zipfile.ZipFile(TESTFN, mode="r") - self.assertEqual(zipfr.comment, comment2) - zipfr.close() + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + 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 + 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!") + with zipfile.ZipFile(TESTFN,mode="a") as zipf: + zipf.comment = b"shorter comment" + 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 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] - zipf = zipfile.ZipFile(io.BytesIO(zipdata), mode="r") - # testzip returns the name of the first corrupt file, or None - self.assertEqual('afile', zipf.testzip()) - zipf.close() + 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) - if zlib: - def test_testzip_with_bad_crc_deflated(self): - self.check_testzip_with_bad_crc(zipfile.ZIP_DEFLATED) + @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.""" + """Tests that files with bad CRCs raise a BadZipFile exception when read.""" zipdata = self.zips_with_bad_crc[compression] # Using ZipFile.read() - zipf = zipfile.ZipFile(io.BytesIO(zipdata), mode="r") - self.assertRaises(zipfile.BadZipfile, zipf.read, 'afile') - zipf.close() + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + self.assertRaises(zipfile.BadZipFile, zipf.read, 'afile') # Using ZipExtFile.read() - zipf = zipfile.ZipFile(io.BytesIO(zipdata), mode="r") - corrupt_file = zipf.open('afile', 'r') - self.assertRaises(zipfile.BadZipfile, corrupt_file.read) - corrupt_file.close() - zipf.close() + 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) - zipf = zipfile.ZipFile(io.BytesIO(zipdata), mode="r") - corrupt_file = zipf.open('afile', 'r') - corrupt_file.MIN_READ_SIZE = 2 - with self.assertRaises(zipfile.BadZipfile): - while corrupt_file.read(2): - pass - corrupt_file.close() - zipf.close() + 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) - if zlib: - def test_read_with_bad_crc_deflated(self): - self.check_read_with_bad_crc(zipfile.ZIP_DEFLATED) + @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 @@ -909,21 +1073,18 @@ class OtherTests(unittest.TestCase): file_size = test_size + 1 junk = b''.join(struct.pack('B', randint(0, 255)) for x in range(file_size)) - zipf = zipfile.ZipFile(io.BytesIO(), "w", compression) - try: + with zipfile.ZipFile(io.BytesIO(), "w", compression) as zipf: zipf.writestr('foo', junk) - fp = zipf.open('foo', 'r') - buf = fp.read(test_size) - self.assertEqual(len(buf), test_size) - finally: - zipf.close() + 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) - if zlib: - def test_read_return_size_deflated(self): - self.check_read_return_size(zipfile.ZIP_DEFLATED) + @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 @@ -932,7 +1093,7 @@ class OtherTests(unittest.TestCase): zipf.close() try: zipf = zipfile.ZipFile(TESTFN, mode="r") - except zipfile.BadZipfile: + except zipfile.BadZipFile: self.fail("Unable to create empty ZIP file in 'w' mode") zipf = zipfile.ZipFile(TESTFN, mode="a") @@ -944,20 +1105,25 @@ class OtherTests(unittest.TestCase): 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 + # raises a BadZipFile exception (rather than the previously unhelpful # IOError) f = open(TESTFN, 'w') f.close() - self.assertRaises(zipfile.BadZipfile, zipfile.ZipFile, TESTFN, 'r') + 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 tearDown(self): - support.unlink(TESTFN) - support.unlink(TESTFN2) + unlink(TESTFN) + unlink(TESTFN2) + class DecryptionTests(unittest.TestCase): - # This test checks that ZIP decryption works. Since the library does not - # support encryption at the moment, we use a pre-generated encrypted - # ZIP file + """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' @@ -981,13 +1147,11 @@ class DecryptionTests(unittest.TestCase): plain2 = b'\x00'*512 def setUp(self): - fp = open(TESTFN, "wb") - fp.write(self.data) - fp.close() + with open(TESTFN, "wb") as fp: + fp.write(self.data) self.zip = zipfile.ZipFile(TESTFN, "r") - fp = open(TESTFN2, "wb") - fp.write(self.data2) - fp.close() + with open(TESTFN2, "wb") as fp: + fp.write(self.data2) self.zip2 = zipfile.ZipFile(TESTFN2, "r") def tearDown(self): @@ -996,19 +1160,20 @@ class DecryptionTests(unittest.TestCase): self.zip2.close() os.unlink(TESTFN2) - def testNoPassword(self): + 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 testBadPassword(self): + def test_bad_password(self): self.zip.setpassword(b"perl") self.assertRaises(RuntimeError, self.zip.read, "test.txt") self.zip2.setpassword(b"perl") self.assertRaises(RuntimeError, self.zip2.read, "zero") - def testGoodPassword(self): + @skipUnless(zlib, "requires zlib") + def test_good_password(self): self.zip.setpassword(b"python") self.assertEqual(self.zip.read("test.txt"), self.plain) self.zip2.setpassword(b"12345") @@ -1028,151 +1193,163 @@ class TestsWithRandomBinaryFiles(unittest.TestCase): for i in range(datacount)) # Make a source file with some lines - fp = open(TESTFN, "wb") - fp.write(self.data) - fp.close() + with open(TESTFN, "wb") as fp: + fp.write(self.data) def tearDown(self): - support.unlink(TESTFN) - support.unlink(TESTFN2) + unlink(TESTFN) + unlink(TESTFN2) - def makeTestArchive(self, f, compression): + def make_test_archive(self, f, compression): # Create the ZIP archive - zipfp = zipfile.ZipFile(f, "w", compression) - zipfp.write(TESTFN, "another.name") - zipfp.write(TESTFN, TESTFN) - zipfp.close() + with zipfile.ZipFile(f, "w", compression) as zipfp: + zipfp.write(TESTFN, "another.name") + zipfp.write(TESTFN, TESTFN) - def zipTest(self, f, compression): - self.makeTestArchive(f, compression) + def zip_test(self, f, compression): + self.make_test_archive(f, compression) # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r", compression) - testdata = zipfp.read(TESTFN) - self.assertEqual(len(testdata), len(self.data)) - self.assertEqual(testdata, self.data) - self.assertEqual(zipfp.read("another.name"), self.data) - zipfp.close() + 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) + if not isinstance(f, str): + f.close() + + def test_stored(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_test(f, zipfile.ZIP_STORED) - def testStored(self): + @skipUnless(zlib, "requires zlib") + def test_deflated(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipTest(f, zipfile.ZIP_STORED) + self.zip_test(f, zipfile.ZIP_DEFLATED) - def zipOpenTest(self, f, compression): - self.makeTestArchive(f, compression) + def zip_open_test(self, f, compression): + self.make_test_archive(f, compression) # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r", compression) - zipdata1 = [] - zipopen1 = zipfp.open(TESTFN) - while 1: - read_data = zipopen1.read(256) - if not read_data: - break - zipdata1.append(read_data) - - zipdata2 = [] - zipopen2 = zipfp.open("another.name") - while 1: - read_data = zipopen2.read(256) - if not read_data: - break - zipdata2.append(read_data) - - testdata1 = b''.join(zipdata1) - self.assertEqual(len(testdata1), len(self.data)) - self.assertEqual(testdata1, self.data) - - testdata2 = b''.join(zipdata2) - self.assertEqual(len(testdata1), len(self.data)) - self.assertEqual(testdata1, self.data) - zipfp.close() + 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 = b''.join(zipdata1) + self.assertEqual(len(testdata1), len(self.data)) + self.assertEqual(testdata1, self.data) + + testdata2 = b''.join(zipdata2) + self.assertEqual(len(testdata2), len(self.data)) + self.assertEqual(testdata2, self.data) + if not isinstance(f, str): + f.close() + + def test_open_stored(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_open_test(f, zipfile.ZIP_STORED) - def testOpenStored(self): + @skipUnless(zlib, "requires zlib") + def test_open_deflated(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipOpenTest(f, zipfile.ZIP_STORED) + self.zip_open_test(f, zipfile.ZIP_DEFLATED) - def zipRandomOpenTest(self, f, compression): - self.makeTestArchive(f, compression) + def zip_random_open_test(self, f, compression): + self.make_test_archive(f, compression) # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r", compression) - zipdata1 = [] - zipopen1 = zipfp.open(TESTFN) - while 1: - read_data = zipopen1.read(randint(1, 1024)) - if not read_data: - break - zipdata1.append(read_data) - - testdata = b''.join(zipdata1) - self.assertEqual(len(testdata), len(self.data)) - self.assertEqual(testdata, self.data) - zipfp.close() + 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 = b''.join(zipdata1) + self.assertEqual(len(testdata), len(self.data)) + self.assertEqual(testdata, self.data) + if not isinstance(f, str): + f.close() + + def test_random_open_stored(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.zip_random_open_test(f, zipfile.ZIP_STORED) - def testRandomOpenStored(self): + @skipUnless(zlib, "requires zlib") + def test_random_open_deflated(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.zipRandomOpenTest(f, zipfile.ZIP_STORED) + self.zip_random_open_test(f, zipfile.ZIP_DEFLATED) + +@skipUnless(zlib, "requires zlib") class TestsWithMultipleOpens(unittest.TestCase): def setUp(self): # Create the ZIP archive - zipfp = zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_DEFLATED) - zipfp.writestr('ones', '1'*FIXEDTEST_SIZE) - zipfp.writestr('twos', '2'*FIXEDTEST_SIZE) - zipfp.close() + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_DEFLATED) as zipfp: + zipfp.writestr('ones', '1'*FIXEDTEST_SIZE) + zipfp.writestr('twos', '2'*FIXEDTEST_SIZE) - def testSameFile(self): + 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. - zipf = zipfile.ZipFile(TESTFN2, mode="r") - zopen1 = zipf.open('ones') - zopen2 = zipf.open('ones') - data1 = zopen1.read(500) - data2 = zopen2.read(500) - data1 += zopen1.read(500) - data2 += zopen2.read(500) - self.assertEqual(data1, data2) - zipf.close() - - def testDifferentFile(self): + 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(500) + data2 += zopen2.read(500) + self.assertEqual(data1, data2) + + 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. - zipf = zipfile.ZipFile(TESTFN2, mode="r") - zopen1 = zipf.open('ones') - zopen2 = zipf.open('twos') - data1 = zopen1.read(500) - data2 = zopen2.read(500) - data1 += zopen1.read(500) - data2 += zopen2.read(500) - self.assertEqual(data1, b'1'*FIXEDTEST_SIZE) - self.assertEqual(data2, b'2'*FIXEDTEST_SIZE) - zipf.close() - - def testInterleaved(self): + 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(500) + data2 += zopen2.read(500) + self.assertEqual(data1, b'1'*FIXEDTEST_SIZE) + self.assertEqual(data2, b'2'*FIXEDTEST_SIZE) + + 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. - zipf = zipfile.ZipFile(TESTFN2, mode="r") - zopen1 = zipf.open('ones') - data1 = zopen1.read(500) - zopen2 = zipf.open('twos') - data2 = zopen2.read(500) - data1 += zopen1.read(500) - data2 += zopen2.read(500) - self.assertEqual(data1, b'1'*FIXEDTEST_SIZE) - self.assertEqual(data2, b'2'*FIXEDTEST_SIZE) - zipf.close() + 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(500) + data2 += zopen2.read(500) + self.assertEqual(data1, b'1'*FIXEDTEST_SIZE) + self.assertEqual(data2, b'2'*FIXEDTEST_SIZE) def tearDown(self): - os.remove(TESTFN2) + unlink(TESTFN2) + class TestWithDirectory(unittest.TestCase): def setUp(self): os.mkdir(TESTFN2) - def testExtractDir(self): - zipf = zipfile.ZipFile(findfile("zipdir.zip")) - zipf.extractall(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"))) @@ -1180,9 +1357,9 @@ class TestWithDirectory(unittest.TestCase): def test_bug_6050(self): # Extraction should succeed if directories already exist os.mkdir(os.path.join(TESTFN2, "a")) - self.testExtractDir() + self.test_extract_dir() - def testStoreDir(self): + def test_store_dir(self): os.mkdir(os.path.join(TESTFN2, "x")) zipf = zipfile.ZipFile(TESTFN, "w") zipf.write(os.path.join(TESTFN2, "x"), "x") @@ -1191,7 +1368,7 @@ class TestWithDirectory(unittest.TestCase): def tearDown(self): shutil.rmtree(TESTFN2) if os.path.exists(TESTFN): - os.remove(TESTFN) + unlink(TESTFN) class UniversalNewlineTests(unittest.TestCase): @@ -1210,105 +1387,145 @@ class UniversalNewlineTests(unittest.TestCase): finally: f.close() - def makeTestArchive(self, f, compression): + def make_test_archive(self, f, compression): # Create the ZIP archive - zipfp = zipfile.ZipFile(f, "w", compression) - for fn in self.arcfiles.values(): - zipfp.write(fn, fn) - zipfp.close() + with zipfile.ZipFile(f, "w", compression) as zipfp: + for fn in self.arcfiles.values(): + zipfp.write(fn, fn) - def readTest(self, f, compression): - self.makeTestArchive(f, compression) + def 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(): - zipdata = zipfp.open(fn, "rU").read() - self.assertEqual(self.arcdata[sep], zipdata) - - zipfp.close() + 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) + if not isinstance(f, str): + f.close() - def readlineTest(self, f, compression): - self.makeTestArchive(f, compression) + 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(): - zipopen = zipfp.open(fn, "rU") - for line in self.line_gen: - linedata = zipopen.readline() - self.assertEqual(linedata, line + b'\n') + with zipfile.ZipFile(f, "r") as zipfp: + for sep, fn in self.arcfiles.items(): + with zipfp.open(fn, "rU") as zipopen: + data = b'' + while True: + read = zipopen.readline() + if not read: + break + data += read - zipfp.close() + read = zipopen.read(5) + if not read: + break + data += read - def readlinesTest(self, f, compression): - self.makeTestArchive(f, compression) + self.assertEqual(data, self.arcdata['\n']) - # Read the ZIP archive - zipfp = zipfile.ZipFile(f, "r") - for sep, fn in self.arcfiles.items(): - ziplines = zipfp.open(fn, "rU").readlines() - for line, zipline in zip(self.line_gen, ziplines): - self.assertEqual(zipline, line + b'\n') + if not isinstance(f, str): + f.close() - zipfp.close() + def readline_test(self, f, compression): + self.make_test_archive(f, compression) - def iterlinesTest(self, f, compression): - self.makeTestArchive(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 + b'\n') + if not isinstance(f, str): + f.close() + + def readlines_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(): - for line, zipline in zip(self.line_gen, zipfp.open(fn, "rU")): - self.assertEqual(zipline, line + b'\n') + 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 + b'\n') + if not isinstance(f, str): + f.close() + + def iterlines_test(self, f, compression): + self.make_test_archive(f, compression) - zipfp.close() + # 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: + for line, zipline in zip(self.line_gen, fp): + self.assertEqual(zipline, line + b'\n') + if not isinstance(f, str): + f.close() + + def test_read_stored(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.read_test(f, zipfile.ZIP_STORED) - def testReadStored(self): + def test_readline_read_stored(self): + # Issue #7610: calls to readline() interleaved with calls to read(). for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.readTest(f, zipfile.ZIP_STORED) + self.readline_read_test(f, zipfile.ZIP_STORED) - def testReadlineStored(self): + def test_readline_stored(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.readlineTest(f, zipfile.ZIP_STORED) + self.readline_test(f, zipfile.ZIP_STORED) - def testReadlinesStored(self): + def test_readlines_stored(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.readlinesTest(f, zipfile.ZIP_STORED) + self.readlines_test(f, zipfile.ZIP_STORED) - def testIterlinesStored(self): + def test_iterlines_stored(self): for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.iterlinesTest(f, zipfile.ZIP_STORED) + self.iterlines_test(f, zipfile.ZIP_STORED) - if zlib: - def testReadDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.readTest(f, zipfile.ZIP_DEFLATED) + @skipUnless(zlib, "requires zlib") + def test_read_deflated(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + 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(), io.BytesIO()): + self.readline_read_test(f, zipfile.ZIP_DEFLATED) - def testReadlineDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.readlineTest(f, zipfile.ZIP_DEFLATED) + @skipUnless(zlib, "requires zlib") + def test_readline_deflated(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.readline_test(f, zipfile.ZIP_DEFLATED) - def testReadlinesDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.readlinesTest(f, zipfile.ZIP_DEFLATED) + @skipUnless(zlib, "requires zlib") + def test_readlines_deflated(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.readlines_test(f, zipfile.ZIP_DEFLATED) - def testIterlinesDeflated(self): - for f in (TESTFN2, TemporaryFile(), io.BytesIO()): - self.iterlinesTest(f, zipfile.ZIP_DEFLATED) + @skipUnless(zlib, "requires zlib") + def test_iterlines_deflated(self): + for f in (TESTFN2, TemporaryFile(), io.BytesIO()): + self.iterlines_test(f, zipfile.ZIP_DEFLATED) def tearDown(self): for sep, fn in self.arcfiles.items(): os.remove(fn) - support.unlink(TESTFN) - support.unlink(TESTFN2) + unlink(TESTFN) + unlink(TESTFN2) def test_main(): run_unittest(TestsWithSourceFile, TestZip64InSmallFiles, OtherTests, PyZipFileTests, DecryptionTests, TestsWithMultipleOpens, - TestWithDirectory, - UniversalNewlineTests, TestsWithRandomBinaryFiles) + TestWithDirectory, UniversalNewlineTests, + TestsWithRandomBinaryFiles) if __name__ == "__main__": test_main() |