Skip to content

C
cpython
Commit fa6bc299 authored by Serhiy Storchaka's avatar Serhiy Storchaka
Browse files
Options

Issue #17944: test_zipfile now discoverable and uses subclassing to 

generate tests for different compression types.  Fixed a bug with skipping
some tests due to use of exhausted iterators.
parent 1eb0cb12
Show whitespace changes
Inline Side-by-side
Showing with 638 additions and 921 deletions
Lib/test/test_zipfile.py
View file @ fa6bc299
...@@ -10,10 +10,9 @@ import unittest ...@@ -10,10 +10,9 @@ import unittest
from tempfile import TemporaryFile from tempfile import TemporaryFile
from random import randint, random from random import randint, random, getrandbits
from unittest import skipUnless
from test.support import (TESTFN, run_unittest, findfile, unlink, from test.support import (TESTFN, findfile, unlink,
requires_zlib, requires_bz2, requires_lzma, requires_zlib, requires_bz2, requires_lzma,
captured_stdout) captured_stdout)
...@@ -27,14 +26,24 @@ SMALL_TEST_DATA = [('_ziptest1', '1q2w3e4r5t'), ...@@ -27,14 +26,24 @@ SMALL_TEST_DATA = [('_ziptest1', '1q2w3e4r5t'),
('ziptest2dir/ziptest3dir/_ziptest3', 'azsxdcfvgb'), ('ziptest2dir/ziptest3dir/_ziptest3', 'azsxdcfvgb'),
('ziptest2dir/ziptest3dir/ziptest4dir/_ziptest3', '6y7u8i9o0p')] ('ziptest2dir/ziptest3dir/ziptest4dir/_ziptest3', '6y7u8i9o0p')]
def get_files(test):
class TestsWithSourceFile(unittest.TestCase): yield TESTFN2
def setUp(self): with TemporaryFile() as f:
self.line_gen = (bytes("Zipfile test line %d. random float: %f" % yield f
test.assertFalse(f.closed)
with io.BytesIO() as f:
yield f
test.assertFalse(f.closed)
class AbstractTestsWithSourceFile:
@classmethod
def setUpClass(cls):
cls.line_gen = [bytes("Zipfile test line %d. random float: %f\n" %
(i, random()), "ascii") (i, random()), "ascii")
for i in range(FIXEDTEST_SIZE)) for i in range(FIXEDTEST_SIZE)]
self.data = b'\n'.join(self.line_gen) + b'\n' cls.data = b''.join(cls.line_gen)
def setUp(self):
# Make a source file with some lines # Make a source file with some lines
with open(TESTFN, "wb") as fp: with open(TESTFN, "wb") as fp:
fp.write(self.data) fp.write(self.data)
...@@ -97,12 +106,10 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -97,12 +106,10 @@ class TestsWithSourceFile(unittest.TestCase):
# Check that testzip doesn't raise an exception # Check that testzip doesn't raise an exception
zipfp.testzip() zipfp.testzip()
if not isinstance(f, str):
f.close()
def test_stored(self): def test_basic(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): for f in get_files(self):
self.zip_test(f, zipfile.ZIP_STORED) self.zip_test(f, self.compression)
def zip_open_test(self, f, compression): def zip_open_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -127,30 +134,10 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -127,30 +134,10 @@ class TestsWithSourceFile(unittest.TestCase):
self.assertEqual(b''.join(zipdata1), self.data) self.assertEqual(b''.join(zipdata1), self.data)
self.assertEqual(b''.join(zipdata2), 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.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")
zipfp.writestr("name", "bar")
with zipfile.ZipFile(TESTFN2, "r") as zipfp: def test_open(self):
infos = zipfp.infolist() for f in get_files(self):
data = b"" self.zip_open_test(f, self.compression)
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): def zip_random_open_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -166,36 +153,17 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -166,36 +153,17 @@ class TestsWithSourceFile(unittest.TestCase):
zipdata1.append(read_data) zipdata1.append(read_data)
self.assertEqual(b''.join(zipdata1), self.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 test_univeral_readaheads(self):
f = io.BytesIO()
data = b'a\r\n' * 16 * 1024
zipfp = zipfile.ZipFile(f, 'w', zipfile.ZIP_STORED)
zipfp.writestr(TESTFN, data)
zipfp.close()
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 test_random_open(self):
for f in get_files(self):
self.zip_random_open_test(f, self.compression)
def zip_readline_read_test(self, f, compression): def zip_readline_read_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
# Read the ZIP archive # Read the ZIP archive
zipfp = zipfile.ZipFile(f, "r") with zipfile.ZipFile(f, "r") as zipfp, \
with zipfp.open(TESTFN) as zipopen: zipfp.open(TESTFN) as zipopen:
data = b'' data = b''
while True: while True:
read = zipopen.readline() read = zipopen.readline()
...@@ -209,9 +177,11 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -209,9 +177,11 @@ class TestsWithSourceFile(unittest.TestCase):
data += read data += read
self.assertEqual(data, self.data) self.assertEqual(data, self.data)
zipfp.close()
if not isinstance(f, str): def test_readline_read(self):
f.close() # Issue #7610: calls to readline() interleaved with calls to read().
for f in get_files(self):
self.zip_readline_read_test(f, self.compression)
def zip_readline_test(self, f, compression): def zip_readline_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -221,9 +191,11 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -221,9 +191,11 @@ class TestsWithSourceFile(unittest.TestCase):
with zipfp.open(TESTFN) as zipopen: with zipfp.open(TESTFN) as zipopen:
for line in self.line_gen: for line in self.line_gen:
linedata = zipopen.readline() linedata = zipopen.readline()
self.assertEqual(linedata, line + '\n') self.assertEqual(linedata, line)
if not isinstance(f, str):
f.close() def test_readline(self):
for f in get_files(self):
self.zip_readline_test(f, self.compression)
def zip_readlines_test(self, f, compression): def zip_readlines_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -233,9 +205,11 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -233,9 +205,11 @@ class TestsWithSourceFile(unittest.TestCase):
with zipfp.open(TESTFN) as zipopen: with zipfp.open(TESTFN) as zipopen:
ziplines = zipopen.readlines() ziplines = zipopen.readlines()
for line, zipline in zip(self.line_gen, ziplines): for line, zipline in zip(self.line_gen, ziplines):
self.assertEqual(zipline, line + '\n') self.assertEqual(zipline, line)
if not isinstance(f, str):
f.close() def test_readlines(self):
for f in get_files(self):
self.zip_readlines_test(f, self.compression)
def zip_iterlines_test(self, f, compression): def zip_iterlines_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -244,173 +218,64 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -244,173 +218,64 @@ class TestsWithSourceFile(unittest.TestCase):
with zipfile.ZipFile(f, "r") as zipfp: with zipfile.ZipFile(f, "r") as zipfp:
with zipfp.open(TESTFN) as zipopen: with zipfp.open(TESTFN) as zipopen:
for line, zipline in zip(self.line_gen, zipopen): for line, zipline in zip(self.line_gen, zipopen):
self.assertEqual(zipline, line + '\n') self.assertEqual(zipline, line)
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.zip_readline_read_test(f, zipfile.ZIP_STORED)
def test_readline_stored(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readline_test(f, zipfile.ZIP_STORED)
def test_readlines_stored(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readlines_test(f, zipfile.ZIP_STORED)
def test_iterlines_stored(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_iterlines_test(f, zipfile.ZIP_STORED)
@requires_zlib def test_iterlines(self):
def test_deflated(self): for f in get_files(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): self.zip_iterlines_test(f, self.compression)
self.zip_test(f, zipfile.ZIP_DEFLATED)
@requires_zlib
def test_open_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_open_test(f, zipfile.ZIP_DEFLATED)
@requires_zlib
def test_random_open_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_random_open_test(f, zipfile.ZIP_DEFLATED)
@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)
@requires_zlib
def test_readline_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readline_test(f, zipfile.ZIP_DEFLATED)
@requires_zlib
def test_readlines_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readlines_test(f, zipfile.ZIP_DEFLATED)
@requires_zlib
def test_iterlines_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_iterlines_test(f, zipfile.ZIP_DEFLATED)
@requires_zlib
def test_low_compression(self): def test_low_compression(self):
"""Check for cases where compressed data is larger than original.""" """Check for cases where compressed data is larger than original."""
# Create the ZIP archive # Create the ZIP archive
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_DEFLATED) as zipfp: with zipfile.ZipFile(TESTFN2, "w", self.compression) as zipfp:
zipfp.writestr("strfile", '12') zipfp.writestr("strfile", '12')
# Get an open object for strfile # Get an open object for strfile
with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_DEFLATED) as zipfp: with zipfile.ZipFile(TESTFN2, "r", self.compression) as zipfp:
with zipfp.open("strfile") as openobj: with zipfp.open("strfile") as openobj:
self.assertEqual(openobj.read(1), b'1') self.assertEqual(openobj.read(1), b'1')
self.assertEqual(openobj.read(1), b'2') self.assertEqual(openobj.read(1), b'2')
@requires_bz2 def test_writestr_compression(self):
def test_bzip2(self): zipfp = zipfile.ZipFile(TESTFN2, "w")
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): zipfp.writestr("b.txt", "hello world", compress_type=self.compression)
self.zip_test(f, zipfile.ZIP_BZIP2) info = zipfp.getinfo('b.txt')
self.assertEqual(info.compress_type, self.compression)
@requires_bz2
def test_open_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_open_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_random_open_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_random_open_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_readline_read_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readline_read_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_readline_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readline_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_readlines_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readlines_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_iterlines_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_iterlines_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_low_compression_bzip2(self):
"""Check for cases where compressed data is larger than original."""
# Create the ZIP archive
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_BZIP2) as zipfp:
zipfp.writestr("strfile", '12')
# Get an open object for strfile def test_read_return_size(self):
with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_BZIP2) as zipfp: # Issue #9837: ZipExtFile.read() shouldn't return more bytes
with zipfp.open("strfile") as openobj: # than requested.
self.assertEqual(openobj.read(1), b'1') for test_size in (1, 4095, 4096, 4097, 16384):
self.assertEqual(openobj.read(1), b'2') file_size = test_size + 1
junk = getrandbits(8 * file_size).to_bytes(file_size, 'little')
with zipfile.ZipFile(io.BytesIO(), "w", self.compression) as zipf:
zipf.writestr('foo', junk)
with zipf.open('foo', 'r') as fp:
buf = fp.read(test_size)
self.assertEqual(len(buf), test_size)
@requires_lzma def tearDown(self):
def test_lzma(self): unlink(TESTFN)
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): unlink(TESTFN2)
self.zip_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_open_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_open_test(f, zipfile.ZIP_LZMA)
@requires_lzma class StoredTestsWithSourceFile(AbstractTestsWithSourceFile,
def test_random_open_lzma(self): unittest.TestCase):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): compression = zipfile.ZIP_STORED
self.zip_random_open_test(f, zipfile.ZIP_LZMA) test_low_compression = None
@requires_lzma def zip_test_writestr_permissions(self, f, compression):
def test_readline_read_lzma(self): # Make sure that writestr creates files with mode 0600,
# Issue #7610: calls to readline() interleaved with calls to read(). # when it is passed a name rather than a ZipInfo instance.
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readline_read_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_readline_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readline_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_readlines_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_readlines_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_iterlines_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_iterlines_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_low_compression_lzma(self):
"""Check for cases where compressed data is larger than original."""
# Create the ZIP archive
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_LZMA) as zipfp:
zipfp.writestr("strfile", '12')
# Get an open object for strfile self.make_test_archive(f, compression)
with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_LZMA) as zipfp: with zipfile.ZipFile(f, "r") as zipfp:
with zipfp.open("strfile") as openobj: zinfo = zipfp.getinfo('strfile')
self.assertEqual(openobj.read(1), b'1') self.assertEqual(zinfo.external_attr, 0o600 << 16)
self.assertEqual(openobj.read(1), b'2')
def test_writestr_permissions(self):
for f in get_files(self):
self.zip_test_writestr_permissions(f, zipfile.ZIP_STORED)
def test_absolute_arcnames(self): def test_absolute_arcnames(self):
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
...@@ -470,7 +335,26 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -470,7 +335,26 @@ class TestsWithSourceFile(unittest.TestCase):
with open(TESTFN, "rb") as f: with open(TESTFN, "rb") as f:
self.assertEqual(zipfp.read(TESTFN), f.read()) self.assertEqual(zipfp.read(TESTFN), f.read())
@requires_zlib 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_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)
@requires_zlib
class DeflateTestsWithSourceFile(AbstractTestsWithSourceFile,
unittest.TestCase):
compression = zipfile.ZIP_DEFLATED
def test_per_file_compression(self): def test_per_file_compression(self):
"""Check that files within a Zip archive can have different """Check that files within a Zip archive can have different
compression options.""" compression options."""
...@@ -482,274 +366,105 @@ class TestsWithSourceFile(unittest.TestCase): ...@@ -482,274 +366,105 @@ class TestsWithSourceFile(unittest.TestCase):
self.assertEqual(sinfo.compress_type, zipfile.ZIP_STORED) self.assertEqual(sinfo.compress_type, zipfile.ZIP_STORED)
self.assertEqual(dinfo.compress_type, zipfile.ZIP_DEFLATED) self.assertEqual(dinfo.compress_type, zipfile.ZIP_DEFLATED)
def test_write_to_readonly(self): @requires_bz2
"""Check that trying to call write() on a readonly ZipFile object class Bzip2TestsWithSourceFile(AbstractTestsWithSourceFile,
raises a RuntimeError.""" unittest.TestCase):
with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: compression = zipfile.ZIP_BZIP2
zipfp.writestr("somefile.txt", "bogus")
with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: @requires_lzma
self.assertRaises(RuntimeError, zipfp.write, TESTFN) class LzmaTestsWithSourceFile(AbstractTestsWithSourceFile,
unittest.TestCase):
compression = zipfile.ZIP_LZMA
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: class AbstractTestZip64InSmallFiles:
for fpath, fdata in SMALL_TEST_DATA: # These tests test the ZIP64 functionality without using large files,
writtenfile = zipfp.extract(fpath) # see test_zipfile64 for proper tests.
# make sure it was written to the right place @classmethod
correctfile = os.path.join(os.getcwd(), fpath) def setUpClass(cls):
correctfile = os.path.normpath(correctfile) line_gen = (bytes("Test of zipfile line %d." % i, "ascii")
for i in range(0, FIXEDTEST_SIZE))
cls.data = b'\n'.join(line_gen)
self.assertEqual(writtenfile, correctfile) def setUp(self):
self._limit = zipfile.ZIP64_LIMIT
zipfile.ZIP64_LIMIT = 5
# make sure correct data is in correct file # Make a source file with some lines
with open(writtenfile, "rb") as f: with open(TESTFN, "wb") as fp:
self.assertEqual(fdata.encode(), f.read()) fp.write(self.data)
os.remove(writtenfile) 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)
# remove the test file subdirectories # Read the ZIP archive
shutil.rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) 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)
def test_extract_all(self): # Print the ZIP directory
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: fp = io.StringIO()
for fpath, fdata in SMALL_TEST_DATA: zipfp.printdir(fp)
zipfp.writestr(fpath, fdata)
with zipfile.ZipFile(TESTFN2, "r") as zipfp: directory = fp.getvalue()
zipfp.extractall() lines = directory.splitlines()
for fpath, fdata in SMALL_TEST_DATA: self.assertEqual(len(lines), 4) # Number of files + header
outfile = os.path.join(os.getcwd(), fpath)
with open(outfile, "rb") as f: self.assertIn('File Name', lines[0])
self.assertEqual(fdata.encode(), f.read()) self.assertIn('Modified', lines[0])
self.assertIn('Size', lines[0])
os.remove(outfile) fn, date, time_, size = lines[1].split()
self.assertEqual(fn, 'another.name')
# remove the test file subdirectories self.assertTrue(time.strptime(date, '%Y-%m-%d'))
shutil.rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) self.assertTrue(time.strptime(time_, '%H:%M:%S'))
self.assertEqual(size, str(len(self.data)))
def check_file(self, filename, content):
self.assertTrue(os.path.isfile(filename))
with open(filename, 'rb') as f:
self.assertEqual(f.read(), content)
def test_sanitize_windows_name(self):
san = zipfile.ZipFile._sanitize_windows_name
# Passing pathsep in allows this test to work regardless of platform.
self.assertEqual(san(r',,?,C:,foo,bar/z', ','), r'_,C_,foo,bar/z')
self.assertEqual(san(r'a\b,c<d>e|f"g?h*i', ','), r'a\b,c_d_e_f_g_h_i')
self.assertEqual(san('../../foo../../ba..r', '/'), r'foo/ba..r')
def test_extract_hackers_arcnames_common_cases(self):
common_hacknames = [
('../foo/bar', 'foo/bar'),
('foo/../bar', 'foo/bar'),
('foo/../../bar', 'foo/bar'),
('foo/bar/..', 'foo/bar'),
('./../foo/bar', 'foo/bar'),
('/foo/bar', 'foo/bar'),
('/foo/../bar', 'foo/bar'),
('/foo/../../bar', 'foo/bar'),
]
self._test_extract_hackers_arcnames(common_hacknames)
@unittest.skipIf(os.path.sep != '\\', 'Requires \\ as path separator.')
def test_extract_hackers_arcnames_windows_only(self):
"""Test combination of path fixing and windows name sanitization."""
windows_hacknames = [
(r'..\foo\bar', 'foo/bar'),
(r'..\/foo\/bar', 'foo/bar'),
(r'foo/\..\/bar', 'foo/bar'),
(r'foo\/../\bar', 'foo/bar'),
(r'C:foo/bar', 'foo/bar'),
(r'C:/foo/bar', 'foo/bar'),
(r'C://foo/bar', 'foo/bar'),
(r'C:\foo\bar', 'foo/bar'),
(r'//conky/mountpoint/foo/bar', 'foo/bar'),
(r'\\conky\mountpoint\foo\bar', 'foo/bar'),
(r'///conky/mountpoint/foo/bar', 'conky/mountpoint/foo/bar'),
(r'\\\conky\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'),
(r'//conky//mountpoint/foo/bar', 'conky/mountpoint/foo/bar'),
(r'\\conky\\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'),
(r'//?/C:/foo/bar', 'foo/bar'),
(r'\\?\C:\foo\bar', 'foo/bar'),
(r'C:/../C:/foo/bar', 'C_/foo/bar'),
(r'a:b\c<d>e|f"g?h*i', 'b/c_d_e_f_g_h_i'),
('../../foo../../ba..r', 'foo/ba..r'),
]
self._test_extract_hackers_arcnames(windows_hacknames)
@unittest.skipIf(os.path.sep != '/', r'Requires / as path separator.')
def test_extract_hackers_arcnames_posix_only(self):
posix_hacknames = [
('//foo/bar', 'foo/bar'),
('../../foo../../ba..r', 'foo../ba..r'),
(r'foo/..\bar', r'foo/..\bar'),
]
self._test_extract_hackers_arcnames(posix_hacknames)
def _test_extract_hackers_arcnames(self, hacknames):
for arcname, fixedname in hacknames:
content = b'foobar' + arcname.encode()
with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipfp:
zinfo = zipfile.ZipInfo()
# preserve backslashes
zinfo.filename = arcname
zinfo.external_attr = 0o600 << 16
zipfp.writestr(zinfo, content)
arcname = arcname.replace(os.sep, "/")
targetpath = os.path.join('target', 'subdir', 'subsub')
correctfile = os.path.join(targetpath, *fixedname.split('/'))
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
writtenfile = zipfp.extract(arcname, targetpath)
self.assertEqual(writtenfile, correctfile,
msg='extract %r: %r != %r' %
(arcname, writtenfile, correctfile))
self.check_file(correctfile, content)
shutil.rmtree('target')
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
zipfp.extractall(targetpath)
self.check_file(correctfile, content)
shutil.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)
shutil.rmtree(fixedname.split('/')[0])
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
zipfp.extractall()
self.check_file(correctfile, content)
shutil.rmtree(fixedname.split('/')[0])
os.remove(TESTFN2)
def test_writestr_compression_stored(self):
zipfp = zipfile.ZipFile(TESTFN2, "w")
zipfp.writestr("a.txt", "hello world", compress_type=zipfile.ZIP_STORED)
info = zipfp.getinfo('a.txt')
self.assertEqual(info.compress_type, zipfile.ZIP_STORED)
@requires_zlib
def test_writestr_compression_deflated(self):
zipfp = zipfile.ZipFile(TESTFN2, "w")
zipfp.writestr("b.txt", "hello world", compress_type=zipfile.ZIP_DEFLATED)
info = zipfp.getinfo('b.txt')
self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED)
@requires_bz2
def test_writestr_compression_bzip2(self):
zipfp = zipfile.ZipFile(TESTFN2, "w")
zipfp.writestr("b.txt", "hello world", compress_type=zipfile.ZIP_BZIP2)
info = zipfp.getinfo('b.txt')
self.assertEqual(info.compress_type, zipfile.ZIP_BZIP2)
@requires_lzma
def test_writestr_compression_lzma(self):
zipfp = zipfile.ZipFile(TESTFN2, "w")
zipfp.writestr("b.txt", "hello world", compress_type=zipfile.ZIP_LZMA)
info = zipfp.getinfo('b.txt')
self.assertEqual(info.compress_type, zipfile.ZIP_LZMA)
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, 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):
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)
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)
# 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_basic(self):
for f in get_files(self):
@requires_zlib self.zip_test(f, self.compression)
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()
def tearDown(self): def tearDown(self):
zipfile.ZIP64_LIMIT = self._limit
unlink(TESTFN) unlink(TESTFN)
unlink(TESTFN2) unlink(TESTFN2)
class TestZip64InSmallFiles(unittest.TestCase): class StoredTestZip64InSmallFiles(AbstractTestZip64InSmallFiles,
# These tests test the ZIP64 functionality without using large files, unittest.TestCase):
# see test_zipfile64 for proper tests. compression = zipfile.ZIP_STORED
def setUp(self):
self._limit = zipfile.ZIP64_LIMIT
zipfile.ZIP64_LIMIT = 5
line_gen = (bytes("Test of zipfile line %d." % i, "ascii")
for i in range(0, FIXEDTEST_SIZE))
self.data = b'\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): def large_file_exception_test(self, f, compression):
with zipfile.ZipFile(f, "w", compression) as zipfp: with zipfile.ZipFile(f, "w", compression) as zipfp:
...@@ -760,92 +475,12 @@ class TestZip64InSmallFiles(unittest.TestCase): ...@@ -760,92 +475,12 @@ class TestZip64InSmallFiles(unittest.TestCase):
with zipfile.ZipFile(f, "w", compression) as zipfp: with zipfile.ZipFile(f, "w", compression) as zipfp:
self.assertRaises(zipfile.LargeZipFile, self.assertRaises(zipfile.LargeZipFile,
zipfp.writestr, "another.name", self.data) zipfp.writestr, "another.name", self.data)
if not isinstance(f, str):
f.close()
def test_large_file_exception(self): def test_large_file_exception(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): for f in get_files(self):
self.large_file_exception_test(f, zipfile.ZIP_STORED) self.large_file_exception_test(f, zipfile.ZIP_STORED)
self.large_file_exception_test2(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 = 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.zip_test(f, zipfile.ZIP_STORED)
@requires_zlib
def test_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_test(f, zipfile.ZIP_DEFLATED)
@requires_bz2
def test_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_test(f, zipfile.ZIP_BZIP2)
@requires_lzma
def test_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_test(f, zipfile.ZIP_LZMA)
def test_absolute_arcnames(self): def test_absolute_arcnames(self):
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED, with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED,
allowZip64=True) as zipfp: allowZip64=True) as zipfp:
...@@ -854,13 +489,27 @@ class TestZip64InSmallFiles(unittest.TestCase): ...@@ -854,13 +489,27 @@ class TestZip64InSmallFiles(unittest.TestCase):
with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp:
self.assertEqual(zipfp.namelist(), ["absolute"]) self.assertEqual(zipfp.namelist(), ["absolute"])
def tearDown(self): @requires_zlib
zipfile.ZIP64_LIMIT = self._limit class DeflateTestZip64InSmallFiles(AbstractTestZip64InSmallFiles,
unlink(TESTFN) unittest.TestCase):
unlink(TESTFN2) compression = zipfile.ZIP_DEFLATED
@requires_bz2
class Bzip2TestZip64InSmallFiles(AbstractTestZip64InSmallFiles,
unittest.TestCase):
compression = zipfile.ZIP_BZIP2
@requires_lzma
class LzmaTestZip64InSmallFiles(AbstractTestZip64InSmallFiles,
unittest.TestCase):
compression = zipfile.ZIP_LZMA
class PyZipFileTests(unittest.TestCase): class PyZipFileTests(unittest.TestCase):
def assertCompiledIn(self, name, namelist):
if name + 'o' not in namelist:
self.assertIn(name + 'c', namelist)
def test_write_pyfile(self): def test_write_pyfile(self):
with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp:
fn = __file__ fn = __file__
...@@ -877,8 +526,7 @@ class PyZipFileTests(unittest.TestCase): ...@@ -877,8 +526,7 @@ class PyZipFileTests(unittest.TestCase):
bn = os.path.basename(fn) bn = os.path.basename(fn)
self.assertNotIn(bn, zipfp.namelist()) self.assertNotIn(bn, zipfp.namelist())
self.assertTrue(bn + 'o' in zipfp.namelist() or self.assertCompiledIn(bn, zipfp.namelist())
bn + 'c' in zipfp.namelist())
with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp:
fn = __file__ fn = __file__
...@@ -889,8 +537,7 @@ class PyZipFileTests(unittest.TestCase): ...@@ -889,8 +537,7 @@ class PyZipFileTests(unittest.TestCase):
bn = "%s/%s" % ("testpackage", os.path.basename(fn)) bn = "%s/%s" % ("testpackage", os.path.basename(fn))
self.assertNotIn(bn, zipfp.namelist()) self.assertNotIn(bn, zipfp.namelist())
self.assertTrue(bn + 'o' in zipfp.namelist() or self.assertCompiledIn(bn, zipfp.namelist())
bn + 'c' in zipfp.namelist())
def test_write_python_package(self): def test_write_python_package(self):
import email import email
...@@ -902,10 +549,8 @@ class PyZipFileTests(unittest.TestCase): ...@@ -902,10 +549,8 @@ class PyZipFileTests(unittest.TestCase):
# Check for a couple of modules at different levels of the # Check for a couple of modules at different levels of the
# hierarchy # hierarchy
names = zipfp.namelist() names = zipfp.namelist()
self.assertTrue('email/__init__.pyo' in names or self.assertCompiledIn('email/__init__.py', names)
'email/__init__.pyc' in names) self.assertCompiledIn('email/mime/text.py', names)
self.assertTrue('email/mime/text.pyo' in names or
'email/mime/text.pyc' in names)
def test_write_with_optimization(self): def test_write_with_optimization(self):
import email import email
...@@ -939,8 +584,8 @@ class PyZipFileTests(unittest.TestCase): ...@@ -939,8 +584,8 @@ class PyZipFileTests(unittest.TestCase):
zipfp.writepy(TESTFN2) zipfp.writepy(TESTFN2)
names = zipfp.namelist() names = zipfp.namelist()
self.assertTrue('mod1.pyc' in names or 'mod1.pyo' in names) self.assertCompiledIn('mod1.py', names)
self.assertTrue('mod2.pyc' in names or 'mod2.pyo' in names) self.assertCompiledIn('mod2.py', names)
self.assertNotIn('mod2.txt', names) self.assertNotIn('mod2.txt', names)
finally: finally:
...@@ -953,72 +598,242 @@ class PyZipFileTests(unittest.TestCase): ...@@ -953,72 +598,242 @@ class PyZipFileTests(unittest.TestCase):
self.assertRaises(RuntimeError, zipfp.writepy, TESTFN) self.assertRaises(RuntimeError, zipfp.writepy, TESTFN)
os.remove(TESTFN) os.remove(TESTFN)
def test_write_pyfile_bad_syntax(self): def test_write_pyfile_bad_syntax(self):
os.mkdir(TESTFN2) os.mkdir(TESTFN2)
try: try:
with open(os.path.join(TESTFN2, "mod1.py"), "w") as fp: with open(os.path.join(TESTFN2, "mod1.py"), "w") as fp:
fp.write("Bad syntax in python file\n") fp.write("Bad syntax in python file\n")
with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp:
# syntax errors are printed to stdout
with captured_stdout() as s:
zipfp.writepy(os.path.join(TESTFN2, "mod1.py"))
self.assertIn("SyntaxError", s.getvalue())
# as it will not have compiled the python file, it will
# include the .py file not .pyc or .pyo
names = zipfp.namelist()
self.assertIn('mod1.py', names)
self.assertNotIn('mod1.pyc', names)
self.assertNotIn('mod1.pyo', names)
finally:
shutil.rmtree(TESTFN2)
class ExtractTests(unittest.TestCase):
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 f:
self.assertEqual(fdata.encode(), f.read())
os.remove(writtenfile)
# remove the test file subdirectories
shutil.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 f:
self.assertEqual(fdata.encode(), f.read())
os.remove(outfile)
# remove the test file subdirectories
shutil.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)
def test_sanitize_windows_name(self):
san = zipfile.ZipFile._sanitize_windows_name
# Passing pathsep in allows this test to work regardless of platform.
self.assertEqual(san(r',,?,C:,foo,bar/z', ','), r'_,C_,foo,bar/z')
self.assertEqual(san(r'a\b,c<d>e|f"g?h*i', ','), r'a\b,c_d_e_f_g_h_i')
self.assertEqual(san('../../foo../../ba..r', '/'), r'foo/ba..r')
def test_extract_hackers_arcnames_common_cases(self):
common_hacknames = [
('../foo/bar', 'foo/bar'),
('foo/../bar', 'foo/bar'),
('foo/../../bar', 'foo/bar'),
('foo/bar/..', 'foo/bar'),
('./../foo/bar', 'foo/bar'),
('/foo/bar', 'foo/bar'),
('/foo/../bar', 'foo/bar'),
('/foo/../../bar', 'foo/bar'),
]
self._test_extract_hackers_arcnames(common_hacknames)
@unittest.skipIf(os.path.sep != '\\', 'Requires \\ as path separator.')
def test_extract_hackers_arcnames_windows_only(self):
"""Test combination of path fixing and windows name sanitization."""
windows_hacknames = [
(r'..\foo\bar', 'foo/bar'),
(r'..\/foo\/bar', 'foo/bar'),
(r'foo/\..\/bar', 'foo/bar'),
(r'foo\/../\bar', 'foo/bar'),
(r'C:foo/bar', 'foo/bar'),
(r'C:/foo/bar', 'foo/bar'),
(r'C://foo/bar', 'foo/bar'),
(r'C:\foo\bar', 'foo/bar'),
(r'//conky/mountpoint/foo/bar', 'foo/bar'),
(r'\\conky\mountpoint\foo\bar', 'foo/bar'),
(r'///conky/mountpoint/foo/bar', 'conky/mountpoint/foo/bar'),
(r'\\\conky\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'),
(r'//conky//mountpoint/foo/bar', 'conky/mountpoint/foo/bar'),
(r'\\conky\\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'),
(r'//?/C:/foo/bar', 'foo/bar'),
(r'\\?\C:\foo\bar', 'foo/bar'),
(r'C:/../C:/foo/bar', 'C_/foo/bar'),
(r'a:b\c<d>e|f"g?h*i', 'b/c_d_e_f_g_h_i'),
('../../foo../../ba..r', 'foo/ba..r'),
]
self._test_extract_hackers_arcnames(windows_hacknames)
@unittest.skipIf(os.path.sep != '/', r'Requires / as path separator.')
def test_extract_hackers_arcnames_posix_only(self):
posix_hacknames = [
('//foo/bar', 'foo/bar'),
('../../foo../../ba..r', 'foo../ba..r'),
(r'foo/..\bar', r'foo/..\bar'),
]
self._test_extract_hackers_arcnames(posix_hacknames)
def _test_extract_hackers_arcnames(self, hacknames):
for arcname, fixedname in hacknames:
content = b'foobar' + arcname.encode()
with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipfp:
zinfo = zipfile.ZipInfo()
# preserve backslashes
zinfo.filename = arcname
zinfo.external_attr = 0o600 << 16
zipfp.writestr(zinfo, content)
arcname = arcname.replace(os.sep, "/")
targetpath = os.path.join('target', 'subdir', 'subsub')
correctfile = os.path.join(targetpath, *fixedname.split('/'))
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
writtenfile = zipfp.extract(arcname, targetpath)
self.assertEqual(writtenfile, correctfile,
msg='extract %r: %r != %r' %
(arcname, writtenfile, correctfile))
self.check_file(correctfile, content)
shutil.rmtree('target')
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
zipfp.extractall(targetpath)
self.check_file(correctfile, content)
shutil.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)
shutil.rmtree(fixedname.split('/')[0])
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
zipfp.extractall()
self.check_file(correctfile, content)
shutil.rmtree(fixedname.split('/')[0])
os.remove(TESTFN2)
class OtherTests(unittest.TestCase):
def test_open_via_zip_info(self):
# Create the ZIP archive
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
zipfp.writestr("name", "foo")
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.assertIn(data, {b"foobar", b"barfoo"})
data = b""
for info in infos:
data += zipfp.read(info)
self.assertIn(data, {b"foobar", b"barfoo"})
def test_universal_readaheads(self):
f = io.BytesIO()
data = b'a\r\n' * 16 * 1024
with zipfile.ZipFile(f, 'w', zipfile.ZIP_STORED) as zipfp:
zipfp.writestr(TESTFN, data)
with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: data2 = b''
# syntax errors are printed to stdout with zipfile.ZipFile(f, 'r') as zipfp, \
with captured_stdout() as s: zipfp.open(TESTFN, 'rU') as zipopen:
zipfp.writepy(os.path.join(TESTFN2, "mod1.py")) for line in zipopen:
data2 += line
self.assertIn("SyntaxError", s.getvalue()) self.assertEqual(data, data2.replace(b'\n', b'\r\n'))
# as it will not have compiled the python file, it will def test_writestr_extended_local_header_issue1202(self):
# include the .py file not .pyc or .pyo with zipfile.ZipFile(TESTFN2, 'w') as orig_zip:
names = zipfp.namelist() for data in 'abcdefghijklmnop':
self.assertIn('mod1.py', names) zinfo = zipfile.ZipInfo(data)
self.assertNotIn('mod1.pyc', names) zinfo.flag_bits |= 0x08 # Include an extended local header.
self.assertNotIn('mod1.pyo', names) orig_zip.writestr(zinfo, data)
finally: def test_close(self):
shutil.rmtree(TESTFN2) """Check that the zipfile is closed after the 'with' block."""
with zipfile.ZipFile(TESTFN2, "w") as zipfp:
for fpath, fdata in SMALL_TEST_DATA:
zipfp.writestr(fpath, fdata)
self.assertIsNotNone(zipfp.fp, 'zipfp is not open')
self.assertIsNone(zipfp.fp, 'zipfp is not closed')
class OtherTests(unittest.TestCase): with zipfile.ZipFile(TESTFN2, "r") as zipfp:
zips_with_bad_crc = { self.assertIsNotNone(zipfp.fp, 'zipfp is not open')
zipfile.ZIP_STORED: ( self.assertIsNone(zipfp.fp, 'zipfp is not closed')
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' def test_close_on_exception(self):
b'ilehello,AworldP' """Check that the zipfile is closed if an exception is raised in the
b'K\001\002\024\003\024\0\0\0\0\0 \213\212;:' 'with' block."""
b'r\253\377\f\0\0\0\f\0\0\0\005\0\0\0\0' with zipfile.ZipFile(TESTFN2, "w") as zipfp:
b'\0\0\0\0\0\0\0\200\001\0\0\0\000afi' for fpath, fdata in SMALL_TEST_DATA:
b'lePK\005\006\0\0\0\0\001\0\001\0003\000' zipfp.writestr(fpath, fdata)
b'\0\0/\0\0\0\0\0'),
zipfile.ZIP_DEFLATED: ( try:
b'PK\x03\x04\x14\x00\x00\x00\x08\x00n}\x0c=FA' with zipfile.ZipFile(TESTFN2, "r") as zipfp2:
b'KE\x10\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' raise zipfile.BadZipFile()
b'ile\xcbH\xcd\xc9\xc9W(\xcf/\xcaI\xc9\xa0' except zipfile.BadZipFile:
b'=\x13\x00PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00n' self.assertIsNone(zipfp2.fp, 'zipfp is not closed')
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'),
zipfile.ZIP_BZIP2: (
b'PK\x03\x04\x14\x03\x00\x00\x0c\x00nu\x0c=FA'
b'KE8\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af'
b'ileBZh91AY&SY\xd4\xa8\xca'
b'\x7f\x00\x00\x0f\x11\x80@\x00\x06D\x90\x80 \x00 \xa5'
b'P\xd9!\x03\x03\x13\x13\x13\x89\xa9\xa9\xc2u5:\x9f'
b'\x8b\xb9"\x9c(HjTe?\x80PK\x01\x02\x14'
b'\x03\x14\x03\x00\x00\x0c\x00nu\x0c=FAKE8'
b'\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\x00'
b'\x00 \x80\x80\x81\x00\x00\x00\x00afilePK'
b'\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00\x00[\x00'
b'\x00\x00\x00\x00'),
zipfile.ZIP_LZMA: (
b'PK\x03\x04\x14\x03\x00\x00\x0e\x00nu\x0c=FA'
b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af'
b'ile\t\x04\x05\x00]\x00\x00\x00\x04\x004\x19I'
b'\xee\x8d\xe9\x17\x89:3`\tq!.8\x00PK'
b'\x01\x02\x14\x03\x14\x03\x00\x00\x0e\x00nu\x0c=FA'
b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00'
b'\x00\x00\x00\x00 \x80\x80\x81\x00\x00\x00\x00afil'
b'ePK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00'
b'\x00>\x00\x00\x00\x00\x00'),
}
def test_unsupported_version(self): def test_unsupported_version(self):
# File has an extract_version of 120 # File has an extract_version of 120
...@@ -1027,10 +842,19 @@ class OtherTests(unittest.TestCase): ...@@ -1027,10 +842,19 @@ class OtherTests(unittest.TestCase):
b'\x00!p\xa1@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00' b'\x00!p\xa1@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00'
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00\x00xPK\x05\x06' b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00\x00xPK\x05\x06'
b'\x00\x00\x00\x00\x01\x00\x01\x00/\x00\x00\x00\x1f\x00\x00\x00\x00\x00') b'\x00\x00\x00\x00\x01\x00\x01\x00/\x00\x00\x00\x1f\x00\x00\x00\x00\x00')
self.assertRaises(NotImplementedError, zipfile.ZipFile, self.assertRaises(NotImplementedError, zipfile.ZipFile,
io.BytesIO(data), 'r') io.BytesIO(data), 'r')
def test_unicode_filenames(self): @requires_zlib
def test_read_unicode_filenames(self):
# bug #10801
fname = findfile('zip_cp437_header.zip')
with zipfile.ZipFile(fname) as zipfp:
for name in zipfp.namelist():
zipfp.open(name).close()
def test_write_unicode_filenames(self):
with zipfile.ZipFile(TESTFN, "w") as zf: with zipfile.ZipFile(TESTFN, "w") as zf:
zf.writestr("foo.txt", "Test for unicode filename") zf.writestr("foo.txt", "Test for unicode filename")
zf.writestr("\xf6.txt", "Test for unicode filename") zf.writestr("\xf6.txt", "Test for unicode filename")
...@@ -1078,20 +902,16 @@ class OtherTests(unittest.TestCase): ...@@ -1078,20 +902,16 @@ class OtherTests(unittest.TestCase):
# - passing a filename # - passing a filename
with open(TESTFN, "w") as fp: with open(TESTFN, "w") as fp:
fp.write("this is not a legal zip file\n") fp.write("this is not a legal zip file\n")
chk = zipfile.is_zipfile(TESTFN) self.assertFalse(zipfile.is_zipfile(TESTFN))
self.assertFalse(chk)
# - passing a file object # - passing a file object
with open(TESTFN, "rb") as fp: with open(TESTFN, "rb") as fp:
chk = zipfile.is_zipfile(fp) self.assertFalse(zipfile.is_zipfile(fp))
self.assertTrue(not chk)
# - passing a file-like object # - passing a file-like object
fp = io.BytesIO() fp = io.BytesIO()
fp.write(b"this is not a legal zip file\n") fp.write(b"this is not a legal zip file\n")
chk = zipfile.is_zipfile(fp) self.assertFalse(zipfile.is_zipfile(fp))
self.assertTrue(not chk)
fp.seek(0, 0) fp.seek(0, 0)
chk = zipfile.is_zipfile(fp) self.assertFalse(zipfile.is_zipfile(fp))
self.assertTrue(not chk)
def test_damaged_zipfile(self): def test_damaged_zipfile(self):
"""Check that zipfiles with missing bytes at the end raise BadZipFile.""" """Check that zipfiles with missing bytes at the end raise BadZipFile."""
...@@ -1113,22 +933,18 @@ class OtherTests(unittest.TestCase): ...@@ -1113,22 +933,18 @@ class OtherTests(unittest.TestCase):
with zipfile.ZipFile(TESTFN, mode="w") as zipf: with zipfile.ZipFile(TESTFN, mode="w") as zipf:
zipf.writestr("foo.txt", b"O, for a Muse of Fire!") zipf.writestr("foo.txt", b"O, for a Muse of Fire!")
chk = zipfile.is_zipfile(TESTFN) self.assertTrue(zipfile.is_zipfile(TESTFN))
self.assertTrue(chk)
# - passing a file object # - passing a file object
with open(TESTFN, "rb") as fp: with open(TESTFN, "rb") as fp:
chk = zipfile.is_zipfile(fp) self.assertTrue(zipfile.is_zipfile(fp))
self.assertTrue(chk)
fp.seek(0, 0) fp.seek(0, 0)
zip_contents = fp.read() zip_contents = fp.read()
# - passing a file-like object # - passing a file-like object
fp = io.BytesIO() fp = io.BytesIO()
fp.write(zip_contents) fp.write(zip_contents)
chk = zipfile.is_zipfile(fp) self.assertTrue(zipfile.is_zipfile(fp))
self.assertTrue(chk)
fp.seek(0, 0) fp.seek(0, 0)
chk = zipfile.is_zipfile(fp) self.assertTrue(zipfile.is_zipfile(fp))
self.assertTrue(chk)
def test_non_existent_file_raises_IOError(self): def test_non_existent_file_raises_IOError(self):
# make sure we don't raise an AttributeError when a partially-constructed # make sure we don't raise an AttributeError when a partially-constructed
...@@ -1309,93 +1125,6 @@ class OtherTests(unittest.TestCase): ...@@ -1309,93 +1125,6 @@ class OtherTests(unittest.TestCase):
with zipfile.ZipFile(TESTFN, "r") as zipf: with zipfile.ZipFile(TESTFN, "r") as zipf:
self.assertEqual(zipf.comment, b"this is a comment") 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)
@requires_zlib
def test_testzip_with_bad_crc_deflated(self):
self.check_testzip_with_bad_crc(zipfile.ZIP_DEFLATED)
@requires_bz2
def test_testzip_with_bad_crc_bzip2(self):
self.check_testzip_with_bad_crc(zipfile.ZIP_BZIP2)
@requires_lzma
def test_testzip_with_bad_crc_lzma(self):
self.check_testzip_with_bad_crc(zipfile.ZIP_LZMA)
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)
@requires_zlib
def test_read_with_bad_crc_deflated(self):
self.check_read_with_bad_crc(zipfile.ZIP_DEFLATED)
@requires_bz2
def test_read_with_bad_crc_bzip2(self):
self.check_read_with_bad_crc(zipfile.ZIP_BZIP2)
@requires_lzma
def test_read_with_bad_crc_lzma(self):
self.check_read_with_bad_crc(zipfile.ZIP_LZMA)
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 = b''.join(struct.pack('B', randint(0, 255))
for x in range(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)
@requires_zlib
def test_read_return_size_deflated(self):
self.check_read_return_size(zipfile.ZIP_DEFLATED)
@requires_bz2
def test_read_return_size_bzip2(self):
self.check_read_return_size(zipfile.ZIP_BZIP2)
@requires_lzma
def test_read_return_size_lzma(self):
self.check_read_return_size(zipfile.ZIP_LZMA)
def test_empty_zipfile(self): def test_empty_zipfile(self):
# Check that creating a file in 'w' or 'a' mode and closing without # 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 # adding any files to the archives creates a valid empty ZIP file
...@@ -1430,6 +1159,93 @@ class OtherTests(unittest.TestCase): ...@@ -1430,6 +1159,93 @@ class OtherTests(unittest.TestCase):
unlink(TESTFN2) unlink(TESTFN2)
class AbstractBadCrcTests:
def test_testzip_with_bad_crc(self):
"""Tests that files with bad CRCs return their name from testzip."""
zipdata = self.zip_with_bad_crc
with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf:
# testzip returns the name of the first corrupt file, or None
self.assertEqual('afile', zipf.testzip())
def test_read_with_bad_crc(self):
"""Tests that files with bad CRCs raise a BadZipFile exception when read."""
zipdata = self.zip_with_bad_crc
# Using ZipFile.read()
with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf:
self.assertRaises(zipfile.BadZipFile, zipf.read, 'afile')
# Using ZipExtFile.read()
with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf:
with zipf.open('afile', 'r') as corrupt_file:
self.assertRaises(zipfile.BadZipFile, corrupt_file.read)
# Same with small reads (in order to exercise the buffering logic)
with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf:
with zipf.open('afile', 'r') as corrupt_file:
corrupt_file.MIN_READ_SIZE = 2
with self.assertRaises(zipfile.BadZipFile):
while corrupt_file.read(2):
pass
class StoredBadCrcTests(AbstractBadCrcTests, unittest.TestCase):
compression = zipfile.ZIP_STORED
zip_with_bad_crc = (
b'PK\003\004\024\0\0\0\0\0 \213\212;:r'
b'\253\377\f\0\0\0\f\0\0\0\005\0\0\000af'
b'ilehello,AworldP'
b'K\001\002\024\003\024\0\0\0\0\0 \213\212;:'
b'r\253\377\f\0\0\0\f\0\0\0\005\0\0\0\0'
b'\0\0\0\0\0\0\0\200\001\0\0\0\000afi'
b'lePK\005\006\0\0\0\0\001\0\001\0003\000'
b'\0\0/\0\0\0\0\0')
@requires_zlib
class DeflateBadCrcTests(AbstractBadCrcTests, unittest.TestCase):
compression = zipfile.ZIP_DEFLATED
zip_with_bad_crc = (
b'PK\x03\x04\x14\x00\x00\x00\x08\x00n}\x0c=FA'
b'KE\x10\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af'
b'ile\xcbH\xcd\xc9\xc9W(\xcf/\xcaI\xc9\xa0'
b'=\x13\x00PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00n'
b'}\x0c=FAKE\x10\x00\x00\x00n\x00\x00\x00\x05'
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00'
b'\x00afilePK\x05\x06\x00\x00\x00\x00\x01\x00'
b'\x01\x003\x00\x00\x003\x00\x00\x00\x00\x00')
@requires_bz2
class Bzip2BadCrcTests(AbstractBadCrcTests, unittest.TestCase):
compression = zipfile.ZIP_BZIP2
zip_with_bad_crc = (
b'PK\x03\x04\x14\x03\x00\x00\x0c\x00nu\x0c=FA'
b'KE8\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af'
b'ileBZh91AY&SY\xd4\xa8\xca'
b'\x7f\x00\x00\x0f\x11\x80@\x00\x06D\x90\x80 \x00 \xa5'
b'P\xd9!\x03\x03\x13\x13\x13\x89\xa9\xa9\xc2u5:\x9f'
b'\x8b\xb9"\x9c(HjTe?\x80PK\x01\x02\x14'
b'\x03\x14\x03\x00\x00\x0c\x00nu\x0c=FAKE8'
b'\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\x00'
b'\x00 \x80\x80\x81\x00\x00\x00\x00afilePK'
b'\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00\x00[\x00'
b'\x00\x00\x00\x00')
@requires_lzma
class LzmaBadCrcTests(AbstractBadCrcTests, unittest.TestCase):
compression = zipfile.ZIP_LZMA
zip_with_bad_crc = (
b'PK\x03\x04\x14\x03\x00\x00\x0e\x00nu\x0c=FA'
b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af'
b'ile\t\x04\x05\x00]\x00\x00\x00\x04\x004\x19I'
b'\xee\x8d\xe9\x17\x89:3`\tq!.8\x00PK'
b'\x01\x02\x14\x03\x14\x03\x00\x00\x0e\x00nu\x0c=FA'
b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00'
b'\x00\x00\x00\x00 \x80\x80\x81\x00\x00\x00\x00afil'
b'ePK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00'
b'\x00>\x00\x00\x00\x00\x00')
class DecryptionTests(unittest.TestCase): class DecryptionTests(unittest.TestCase):
"""Check that ZIP decryption works. Since the library does not """Check that ZIP decryption works. Since the library does not
support encryption at the moment, we use a pre-generated encrypted support encryption at the moment, we use a pre-generated encrypted
...@@ -1495,13 +1311,14 @@ class DecryptionTests(unittest.TestCase): ...@@ -1495,13 +1311,14 @@ class DecryptionTests(unittest.TestCase):
self.assertRaises(TypeError, self.zip.open, "test.txt", pwd="python") self.assertRaises(TypeError, self.zip.open, "test.txt", pwd="python")
self.assertRaises(TypeError, self.zip.extract, "test.txt", pwd="python") self.assertRaises(TypeError, self.zip.extract, "test.txt", pwd="python")
class AbstractTestsWithRandomBinaryFiles:
class TestsWithRandomBinaryFiles(unittest.TestCase): @classmethod
def setUp(self): def setUpClass(cls):
datacount = randint(16, 64)*1024 + randint(1, 1024) datacount = randint(16, 64)*1024 + randint(1, 1024)
self.data = b''.join(struct.pack('<f', random()*randint(-1000, 1000)) cls.data = b''.join(struct.pack('<f', random()*randint(-1000, 1000))
for i in range(datacount)) for i in range(datacount))
def setUp(self):
# Make a source file with some lines # Make a source file with some lines
with open(TESTFN, "wb") as fp: with open(TESTFN, "wb") as fp:
fp.write(self.data) fp.write(self.data)
...@@ -1525,27 +1342,10 @@ class TestsWithRandomBinaryFiles(unittest.TestCase): ...@@ -1525,27 +1342,10 @@ class TestsWithRandomBinaryFiles(unittest.TestCase):
self.assertEqual(len(testdata), len(self.data)) self.assertEqual(len(testdata), len(self.data))
self.assertEqual(testdata, self.data) self.assertEqual(testdata, self.data)
self.assertEqual(zipfp.read("another.name"), 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)
@requires_zlib def test_read(self):
def test_deflated(self): for f in get_files(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): self.zip_test(f, self.compression)
self.zip_test(f, zipfile.ZIP_DEFLATED)
@requires_bz2
def test_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_test(f, zipfile.ZIP_BZIP2)
@requires_lzma
def test_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_test(f, zipfile.ZIP_LZMA)
def zip_open_test(self, f, compression): def zip_open_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -1575,27 +1375,10 @@ class TestsWithRandomBinaryFiles(unittest.TestCase): ...@@ -1575,27 +1375,10 @@ class TestsWithRandomBinaryFiles(unittest.TestCase):
testdata2 = b''.join(zipdata2) testdata2 = b''.join(zipdata2)
self.assertEqual(len(testdata2), len(self.data)) self.assertEqual(len(testdata2), len(self.data))
self.assertEqual(testdata2, 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)
@requires_zlib
def test_open_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_open_test(f, zipfile.ZIP_DEFLATED)
@requires_bz2
def test_open_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_open_test(f, zipfile.ZIP_BZIP2)
@requires_lzma def test_open(self):
def test_open_lzma(self): for f in get_files(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): self.zip_open_test(f, self.compression)
self.zip_open_test(f, zipfile.ZIP_LZMA)
def zip_random_open_test(self, f, compression): def zip_random_open_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -1613,27 +1396,30 @@ class TestsWithRandomBinaryFiles(unittest.TestCase): ...@@ -1613,27 +1396,30 @@ class TestsWithRandomBinaryFiles(unittest.TestCase):
testdata = b''.join(zipdata1) testdata = b''.join(zipdata1)
self.assertEqual(len(testdata), len(self.data)) self.assertEqual(len(testdata), len(self.data))
self.assertEqual(testdata, self.data) self.assertEqual(testdata, self.data)
if not isinstance(f, str):
f.close()
def test_random_open_stored(self): def test_random_open(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): for f in get_files(self):
self.zip_random_open_test(f, zipfile.ZIP_STORED) self.zip_random_open_test(f, self.compression)
@requires_zlib
def test_random_open_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.zip_random_open_test(f, zipfile.ZIP_DEFLATED)
@requires_bz2 class StoredTestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles,
def test_random_open_bzip2(self): unittest.TestCase):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): compression = zipfile.ZIP_STORED
self.zip_random_open_test(f, zipfile.ZIP_BZIP2)
@requires_zlib
class DeflateTestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles,
unittest.TestCase):
compression = zipfile.ZIP_DEFLATED
@requires_bz2
class Bzip2TestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles,
unittest.TestCase):
compression = zipfile.ZIP_BZIP2
@requires_lzma @requires_lzma
def test_random_open_lzma(self): class LzmaTestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles,
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): unittest.TestCase):
self.zip_random_open_test(f, zipfile.ZIP_LZMA) compression = zipfile.ZIP_LZMA
@requires_zlib @requires_zlib
...@@ -1711,21 +1497,22 @@ class TestWithDirectory(unittest.TestCase): ...@@ -1711,21 +1497,22 @@ class TestWithDirectory(unittest.TestCase):
unlink(TESTFN) unlink(TESTFN)
class UniversalNewlineTests(unittest.TestCase): class AbstractUniversalNewlineTests:
def setUp(self): @classmethod
self.line_gen = [bytes("Test of zipfile line %d." % i, "ascii") def setUpClass(cls):
cls.line_gen = [bytes("Test of zipfile line %d." % i, "ascii")
for i in range(FIXEDTEST_SIZE)] for i in range(FIXEDTEST_SIZE)]
self.seps = ('\r', '\r\n', '\n') cls.seps = (b'\r', b'\r\n', b'\n')
self.arcdata, self.arcfiles = {}, {} cls.arcdata = {}
for n, s in enumerate(cls.seps):
cls.arcdata[s] = s.join(cls.line_gen) + s
def setUp(self):
self.arcfiles = {}
for n, s in enumerate(self.seps): for n, s in enumerate(self.seps):
b = s.encode("ascii")
self.arcdata[s] = b.join(self.line_gen) + b
self.arcfiles[s] = '%s-%d' % (TESTFN, n) self.arcfiles[s] = '%s-%d' % (TESTFN, n)
f = open(self.arcfiles[s], "wb") with open(self.arcfiles[s], "wb") as f:
try:
f.write(self.arcdata[s]) f.write(self.arcdata[s])
finally:
f.close()
def make_test_archive(self, f, compression): def make_test_archive(self, f, compression):
# Create the ZIP archive # Create the ZIP archive
...@@ -1742,8 +1529,10 @@ class UniversalNewlineTests(unittest.TestCase): ...@@ -1742,8 +1529,10 @@ class UniversalNewlineTests(unittest.TestCase):
with zipfp.open(fn, "rU") as fp: with zipfp.open(fn, "rU") as fp:
zipdata = fp.read() zipdata = fp.read()
self.assertEqual(self.arcdata[sep], zipdata) self.assertEqual(self.arcdata[sep], zipdata)
if not isinstance(f, str):
f.close() def test_read(self):
for f in get_files(self):
self.read_test(f, self.compression)
def readline_read_test(self, f, compression): def readline_read_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -1764,10 +1553,11 @@ class UniversalNewlineTests(unittest.TestCase): ...@@ -1764,10 +1553,11 @@ class UniversalNewlineTests(unittest.TestCase):
break break
data += read data += read
self.assertEqual(data, self.arcdata['\n']) self.assertEqual(data, self.arcdata[b'\n'])
if not isinstance(f, str): def test_readline_read(self):
f.close() for f in get_files(self):
self.readline_read_test(f, self.compression)
def readline_test(self, f, compression): def readline_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -1779,8 +1569,10 @@ class UniversalNewlineTests(unittest.TestCase): ...@@ -1779,8 +1569,10 @@ class UniversalNewlineTests(unittest.TestCase):
for line in self.line_gen: for line in self.line_gen:
linedata = zipopen.readline() linedata = zipopen.readline()
self.assertEqual(linedata, line + b'\n') self.assertEqual(linedata, line + b'\n')
if not isinstance(f, str):
f.close() def test_readline(self):
for f in get_files(self):
self.readline_test(f, self.compression)
def readlines_test(self, f, compression): def readlines_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -1792,8 +1584,10 @@ class UniversalNewlineTests(unittest.TestCase): ...@@ -1792,8 +1584,10 @@ class UniversalNewlineTests(unittest.TestCase):
ziplines = fp.readlines() ziplines = fp.readlines()
for line, zipline in zip(self.line_gen, ziplines): for line, zipline in zip(self.line_gen, ziplines):
self.assertEqual(zipline, line + b'\n') self.assertEqual(zipline, line + b'\n')
if not isinstance(f, str):
f.close() def test_readlines(self):
for f in get_files(self):
self.readlines_test(f, self.compression)
def iterlines_test(self, f, compression): def iterlines_test(self, f, compression):
self.make_test_archive(f, compression) self.make_test_archive(f, compression)
...@@ -1804,105 +1598,10 @@ class UniversalNewlineTests(unittest.TestCase): ...@@ -1804,105 +1598,10 @@ class UniversalNewlineTests(unittest.TestCase):
with zipfp.open(fn, "rU") as fp: with zipfp.open(fn, "rU") as fp:
for line, zipline in zip(self.line_gen, fp): for line, zipline in zip(self.line_gen, fp):
self.assertEqual(zipline, line + b'\n') 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 test_readline_read_stored(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_STORED)
def test_readline_stored(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readline_test(f, zipfile.ZIP_STORED)
def test_readlines_stored(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readlines_test(f, zipfile.ZIP_STORED)
def test_iterlines_stored(self): def test_iterlines(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()): for f in get_files(self):
self.iterlines_test(f, zipfile.ZIP_STORED) self.iterlines_test(f, self.compression)
@requires_zlib
def test_read_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.read_test(f, zipfile.ZIP_DEFLATED)
@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)
@requires_zlib
def test_readline_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readline_test(f, zipfile.ZIP_DEFLATED)
@requires_zlib
def test_readlines_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readlines_test(f, zipfile.ZIP_DEFLATED)
@requires_zlib
def test_iterlines_deflated(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.iterlines_test(f, zipfile.ZIP_DEFLATED)
@requires_bz2
def test_read_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.read_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_readline_read_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readline_read_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_readline_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readline_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_readlines_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readlines_test(f, zipfile.ZIP_BZIP2)
@requires_bz2
def test_iterlines_bzip2(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.iterlines_test(f, zipfile.ZIP_BZIP2)
@requires_lzma
def test_read_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.read_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_readline_read_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readline_read_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_readline_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readline_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_readlines_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.readlines_test(f, zipfile.ZIP_LZMA)
@requires_lzma
def test_iterlines_lzma(self):
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
self.iterlines_test(f, zipfile.ZIP_LZMA)
def tearDown(self): def tearDown(self):
for sep, fn in self.arcfiles.items(): for sep, fn in self.arcfiles.items():
...@@ -1911,11 +1610,25 @@ class UniversalNewlineTests(unittest.TestCase): ...@@ -1911,11 +1610,25 @@ class UniversalNewlineTests(unittest.TestCase):
unlink(TESTFN2) unlink(TESTFN2)
def test_main(): class StoredUniversalNewlineTests(AbstractUniversalNewlineTests,
run_unittest(TestsWithSourceFile, TestZip64InSmallFiles, OtherTests, unittest.TestCase):
PyZipFileTests, DecryptionTests, TestsWithMultipleOpens, compression = zipfile.ZIP_STORED
TestWithDirectory, UniversalNewlineTests,
TestsWithRandomBinaryFiles) @requires_zlib
class DeflateUniversalNewlineTests(AbstractUniversalNewlineTests,
unittest.TestCase):
compression = zipfile.ZIP_DEFLATED
@requires_bz2
class Bzip2UniversalNewlineTests(AbstractUniversalNewlineTests,
unittest.TestCase):
compression = zipfile.ZIP_BZIP2
@requires_lzma
class LzmaUniversalNewlineTests(AbstractUniversalNewlineTests,
unittest.TestCase):
compression = zipfile.ZIP_LZMA
if __name__ == "__main__": if __name__ == "__main__":
test_main() unittest.main()
Misc/NEWS
View file @ fa6bc299
...@@ -208,6 +208,10 @@ IDLE ...@@ -208,6 +208,10 @@ IDLE
Tests Tests
----- -----
- Issue #17944: test_zipfile now discoverable and uses subclassing to
generate tests for different compression types. Fixed a bug with skipping
some tests due to use of exhausted iterators.
- Issue #18266: test_largefile now works with unittest test discovery and - Issue #18266: test_largefile now works with unittest test discovery and
supports running only selected tests. Patch by Zachary Ware. supports running only selected tests. Patch by Zachary Ware.
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7