1852 lines
72 KiB
Python
1852 lines
72 KiB
Python
# We can test part of the module without zlib.
|
|
try:
|
|
import zlib
|
|
except ImportError:
|
|
zlib = None
|
|
|
|
import os
|
|
import io
|
|
import sys
|
|
import time
|
|
import struct
|
|
import zipfile
|
|
import unittest
|
|
|
|
from StringIO import StringIO
|
|
from tempfile import TemporaryFile
|
|
from random import randint, random, getrandbits
|
|
from unittest import skipUnless
|
|
|
|
from test import script_helper
|
|
from test.test_support import TESTFN, TESTFN_UNICODE, TESTFN_ENCODING, \
|
|
run_unittest, findfile, unlink, rmtree, \
|
|
check_warnings, captured_stdout
|
|
try:
|
|
TESTFN_UNICODE.encode(TESTFN_ENCODING)
|
|
except (UnicodeError, TypeError):
|
|
# Either the file system encoding is None, or the file name
|
|
# cannot be encoded in the file system encoding.
|
|
TESTFN_UNICODE = None
|
|
|
|
TESTFN2 = TESTFN + "2"
|
|
TESTFNDIR = TESTFN + "d"
|
|
FIXEDTEST_SIZE = 1000
|
|
|
|
SMALL_TEST_DATA = [('_ziptest1', '1q2w3e4r5t'),
|
|
('ziptest2dir/_ziptest2', 'qawsedrftg'),
|
|
('ziptest2dir/ziptest3dir/_ziptest3', 'azsxdcfvgb'),
|
|
('ziptest2dir/ziptest3dir/ziptest4dir/_ziptest3', '6y7u8i9o0p')]
|
|
|
|
def getrandbytes(size):
|
|
return bytes(bytearray.fromhex('%0*x' % (2 * size, getrandbits(8 * size))))
|
|
|
|
class TestsWithSourceFile(unittest.TestCase):
|
|
def setUp(self):
|
|
self.line_gen = ["Zipfile test line %d. random float: %f" % (i, random())
|
|
for i in xrange(FIXEDTEST_SIZE)]
|
|
self.data = '\n'.join(self.line_gen) + '\n'
|
|
|
|
# Make a source file with some lines
|
|
with open(TESTFN, "wb") as fp:
|
|
fp.write(self.data)
|
|
|
|
def make_test_archive(self, f, compression):
|
|
# Create the ZIP archive
|
|
with zipfile.ZipFile(f, "w", compression) as zipfp:
|
|
zipfp.write(TESTFN, "another.name")
|
|
zipfp.write(TESTFN, TESTFN)
|
|
zipfp.writestr("strfile", self.data)
|
|
|
|
def zip_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r", compression) as zipfp:
|
|
self.assertEqual(zipfp.read(TESTFN), self.data)
|
|
self.assertEqual(zipfp.read("another.name"), self.data)
|
|
self.assertEqual(zipfp.read("strfile"), self.data)
|
|
|
|
# Print the ZIP directory
|
|
fp = StringIO()
|
|
stdout = sys.stdout
|
|
try:
|
|
sys.stdout = fp
|
|
zipfp.printdir()
|
|
finally:
|
|
sys.stdout = stdout
|
|
|
|
directory = fp.getvalue()
|
|
lines = directory.splitlines()
|
|
self.assertEqual(len(lines), 4) # Number of files + header
|
|
|
|
self.assertIn('File Name', lines[0])
|
|
self.assertIn('Modified', lines[0])
|
|
self.assertIn('Size', lines[0])
|
|
|
|
fn, date, time_, size = lines[1].split()
|
|
self.assertEqual(fn, 'another.name')
|
|
self.assertTrue(time.strptime(date, '%Y-%m-%d'))
|
|
self.assertTrue(time.strptime(time_, '%H:%M:%S'))
|
|
self.assertEqual(size, str(len(self.data)))
|
|
|
|
# Check the namelist
|
|
names = zipfp.namelist()
|
|
self.assertEqual(len(names), 3)
|
|
self.assertIn(TESTFN, names)
|
|
self.assertIn("another.name", names)
|
|
self.assertIn("strfile", names)
|
|
|
|
# Check infolist
|
|
infos = zipfp.infolist()
|
|
names = [i.filename for i in infos]
|
|
self.assertEqual(len(names), 3)
|
|
self.assertIn(TESTFN, names)
|
|
self.assertIn("another.name", names)
|
|
self.assertIn("strfile", names)
|
|
for i in infos:
|
|
self.assertEqual(i.file_size, len(self.data))
|
|
|
|
# check getinfo
|
|
for nm in (TESTFN, "another.name", "strfile"):
|
|
info = zipfp.getinfo(nm)
|
|
self.assertEqual(info.filename, nm)
|
|
self.assertEqual(info.file_size, len(self.data))
|
|
|
|
# Check that testzip doesn't raise an exception
|
|
zipfp.testzip()
|
|
|
|
def test_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_test(f, zipfile.ZIP_STORED)
|
|
|
|
def zip_open_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r", compression) as zipfp:
|
|
zipdata1 = []
|
|
with zipfp.open(TESTFN) as zipopen1:
|
|
while True:
|
|
read_data = zipopen1.read(256)
|
|
if not read_data:
|
|
break
|
|
zipdata1.append(read_data)
|
|
|
|
zipdata2 = []
|
|
with zipfp.open("another.name") as zipopen2:
|
|
while True:
|
|
read_data = zipopen2.read(256)
|
|
if not read_data:
|
|
break
|
|
zipdata2.append(read_data)
|
|
|
|
self.assertEqual(''.join(zipdata1), self.data)
|
|
self.assertEqual(''.join(zipdata2), self.data)
|
|
|
|
def test_open_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_open_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_open_via_zip_info(self):
|
|
# Create the ZIP archive
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.writestr("name", "foo")
|
|
with check_warnings(('', UserWarning)):
|
|
zipfp.writestr("name", "bar")
|
|
self.assertEqual(zipfp.namelist(), ["name"] * 2)
|
|
|
|
with zipfile.ZipFile(TESTFN2, "r") as zipfp:
|
|
infos = zipfp.infolist()
|
|
data = ""
|
|
for info in infos:
|
|
with zipfp.open(info) as f:
|
|
data += f.read()
|
|
self.assertTrue(data == "foobar" or data == "barfoo")
|
|
data = ""
|
|
for info in infos:
|
|
data += zipfp.read(info)
|
|
self.assertTrue(data == "foobar" or data == "barfoo")
|
|
|
|
def zip_random_open_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r", compression) as zipfp:
|
|
zipdata1 = []
|
|
with zipfp.open(TESTFN) as zipopen1:
|
|
while True:
|
|
read_data = zipopen1.read(randint(1, 1024))
|
|
if not read_data:
|
|
break
|
|
zipdata1.append(read_data)
|
|
|
|
self.assertEqual(''.join(zipdata1), self.data)
|
|
|
|
def test_random_open_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_random_open_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_universal_readaheads(self):
|
|
f = StringIO()
|
|
|
|
data = 'a\r\n' * 16 * 1024
|
|
with zipfile.ZipFile(f, 'w', zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.writestr(TESTFN, data)
|
|
|
|
data2 = ''
|
|
with zipfile.ZipFile(f, 'r') as zipfp:
|
|
with zipfp.open(TESTFN, 'rU') as zipopen:
|
|
for line in zipopen:
|
|
data2 += line
|
|
|
|
self.assertEqual(data, data2.replace('\n', '\r\n'))
|
|
|
|
def zip_readline_read_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
with zipfp.open(TESTFN) as zipopen:
|
|
data = ''
|
|
while True:
|
|
read = zipopen.readline()
|
|
if not read:
|
|
break
|
|
data += read
|
|
|
|
read = zipopen.read(100)
|
|
if not read:
|
|
break
|
|
data += read
|
|
|
|
self.assertEqual(data, self.data)
|
|
|
|
def zip_readline_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
with zipfp.open(TESTFN) as zipopen:
|
|
for line in self.line_gen:
|
|
linedata = zipopen.readline()
|
|
self.assertEqual(linedata, line + '\n')
|
|
|
|
def zip_readlines_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
with zipfp.open(TESTFN) as zo:
|
|
ziplines = zo.readlines()
|
|
for line, zipline in zip(self.line_gen, ziplines):
|
|
self.assertEqual(zipline, line + '\n')
|
|
|
|
def zip_iterlines_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
for line, zipline in zip(self.line_gen, zipfp.open(TESTFN)):
|
|
self.assertEqual(zipline, line + '\n')
|
|
|
|
def test_readline_read_stored(self):
|
|
# Issue #7610: calls to readline() interleaved with calls to read().
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_readline_read_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_readline_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_readline_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_readlines_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_readlines_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_iterlines_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_iterlines_test(f, zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_open_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_open_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_random_open_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_random_open_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_readline_read_deflated(self):
|
|
# Issue #7610: calls to readline() interleaved with calls to read().
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_readline_read_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_readline_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_readline_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_readlines_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_readlines_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_iterlines_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_iterlines_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_low_compression(self):
|
|
"""Check for cases where compressed data is larger than original."""
|
|
# Create the ZIP archive
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_DEFLATED) as zipfp:
|
|
zipfp.writestr("strfile", '12')
|
|
|
|
# Get an open object for strfile
|
|
with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_DEFLATED) as zipfp:
|
|
with zipfp.open("strfile") as openobj:
|
|
self.assertEqual(openobj.read(1), '1')
|
|
self.assertEqual(openobj.read(1), '2')
|
|
|
|
def test_absolute_arcnames(self):
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.write(TESTFN, "/absolute")
|
|
|
|
with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp:
|
|
self.assertEqual(zipfp.namelist(), ["absolute"])
|
|
|
|
def test_append_to_zip_file(self):
|
|
"""Test appending to an existing zipfile."""
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.write(TESTFN, TESTFN)
|
|
|
|
with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.writestr("strfile", self.data)
|
|
self.assertEqual(zipfp.namelist(), [TESTFN, "strfile"])
|
|
|
|
def test_append_to_non_zip_file(self):
|
|
"""Test appending to an existing file that is not a zipfile."""
|
|
# NOTE: this test fails if len(d) < 22 because of the first
|
|
# line "fpin.seek(-22, 2)" in _EndRecData
|
|
data = 'I am not a ZipFile!'*10
|
|
with open(TESTFN2, 'wb') as f:
|
|
f.write(data)
|
|
|
|
with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.write(TESTFN, TESTFN)
|
|
|
|
with open(TESTFN2, 'rb') as f:
|
|
f.seek(len(data))
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
self.assertEqual(zipfp.namelist(), [TESTFN])
|
|
self.assertEqual(zipfp.read(TESTFN), self.data)
|
|
with open(TESTFN2, 'rb') as f:
|
|
self.assertEqual(f.read(len(data)), data)
|
|
zipfiledata = f.read()
|
|
with io.BytesIO(zipfiledata) as bio, zipfile.ZipFile(bio) as zipfp:
|
|
self.assertEqual(zipfp.namelist(), [TESTFN])
|
|
self.assertEqual(zipfp.read(TESTFN), self.data)
|
|
|
|
def test_read_concatenated_zip_file(self):
|
|
with io.BytesIO() as bio:
|
|
with zipfile.ZipFile(bio, 'w', zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.write(TESTFN, TESTFN)
|
|
zipfiledata = bio.getvalue()
|
|
data = b'I am not a ZipFile!'*10
|
|
with open(TESTFN2, 'wb') as f:
|
|
f.write(data)
|
|
f.write(zipfiledata)
|
|
|
|
with zipfile.ZipFile(TESTFN2) as zipfp:
|
|
self.assertEqual(zipfp.namelist(), [TESTFN])
|
|
self.assertEqual(zipfp.read(TESTFN), self.data)
|
|
|
|
def test_append_to_concatenated_zip_file(self):
|
|
with io.BytesIO() as bio:
|
|
with zipfile.ZipFile(bio, 'w', zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.write(TESTFN, TESTFN)
|
|
zipfiledata = bio.getvalue()
|
|
data = b'I am not a ZipFile!'*1000000
|
|
with open(TESTFN2, 'wb') as f:
|
|
f.write(data)
|
|
f.write(zipfiledata)
|
|
|
|
with zipfile.ZipFile(TESTFN2, 'a') as zipfp:
|
|
self.assertEqual(zipfp.namelist(), [TESTFN])
|
|
zipfp.writestr('strfile', self.data)
|
|
|
|
with open(TESTFN2, 'rb') as f:
|
|
self.assertEqual(f.read(len(data)), data)
|
|
zipfiledata = f.read()
|
|
with io.BytesIO(zipfiledata) as bio, zipfile.ZipFile(bio) as zipfp:
|
|
self.assertEqual(zipfp.namelist(), [TESTFN, 'strfile'])
|
|
self.assertEqual(zipfp.read(TESTFN), self.data)
|
|
self.assertEqual(zipfp.read('strfile'), self.data)
|
|
|
|
def test_ignores_newline_at_end(self):
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.write(TESTFN, TESTFN)
|
|
with open(TESTFN2, 'a') as f:
|
|
f.write("\r\n\00\00\00")
|
|
with zipfile.ZipFile(TESTFN2, "r") as zipfp:
|
|
self.assertIsInstance(zipfp, zipfile.ZipFile)
|
|
|
|
def test_ignores_stuff_appended_past_comments(self):
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
|
|
zipfp.comment = b"this is a comment"
|
|
zipfp.write(TESTFN, TESTFN)
|
|
with open(TESTFN2, 'a') as f:
|
|
f.write("abcdef\r\n")
|
|
with zipfile.ZipFile(TESTFN2, "r") as zipfp:
|
|
self.assertIsInstance(zipfp, zipfile.ZipFile)
|
|
self.assertEqual(zipfp.comment, b"this is a comment")
|
|
|
|
def test_write_default_name(self):
|
|
"""Check that calling ZipFile.write without arcname specified
|
|
produces the expected result."""
|
|
with zipfile.ZipFile(TESTFN2, "w") as zipfp:
|
|
zipfp.write(TESTFN)
|
|
with open(TESTFN,'r') as fid:
|
|
self.assertEqual(zipfp.read(TESTFN), fid.read())
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_per_file_compression(self):
|
|
"""Check that files within a Zip archive can have different
|
|
compression options."""
|
|
with zipfile.ZipFile(TESTFN2, "w") as zipfp:
|
|
zipfp.write(TESTFN, 'storeme', zipfile.ZIP_STORED)
|
|
zipfp.write(TESTFN, 'deflateme', zipfile.ZIP_DEFLATED)
|
|
sinfo = zipfp.getinfo('storeme')
|
|
dinfo = zipfp.getinfo('deflateme')
|
|
self.assertEqual(sinfo.compress_type, zipfile.ZIP_STORED)
|
|
self.assertEqual(dinfo.compress_type, zipfile.ZIP_DEFLATED)
|
|
|
|
def test_write_to_readonly(self):
|
|
"""Check that trying to call write() on a readonly ZipFile object
|
|
raises a RuntimeError."""
|
|
with zipfile.ZipFile(TESTFN2, mode="w") as zipfp:
|
|
zipfp.writestr("somefile.txt", "bogus")
|
|
|
|
with zipfile.ZipFile(TESTFN2, mode="r") as zipfp:
|
|
self.assertRaises(RuntimeError, zipfp.write, TESTFN)
|
|
|
|
def test_extract(self):
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
|
|
for fpath, fdata in SMALL_TEST_DATA:
|
|
zipfp.writestr(fpath, fdata)
|
|
|
|
with zipfile.ZipFile(TESTFN2, "r") as zipfp:
|
|
for fpath, fdata in SMALL_TEST_DATA:
|
|
writtenfile = zipfp.extract(fpath)
|
|
|
|
# make sure it was written to the right place
|
|
correctfile = os.path.join(os.getcwd(), fpath)
|
|
correctfile = os.path.normpath(correctfile)
|
|
|
|
self.assertEqual(writtenfile, correctfile)
|
|
|
|
# make sure correct data is in correct file
|
|
with open(writtenfile, "rb") as fid:
|
|
self.assertEqual(fdata, fid.read())
|
|
os.remove(writtenfile)
|
|
|
|
# remove the test file subdirectories
|
|
rmtree(os.path.join(os.getcwd(), 'ziptest2dir'))
|
|
|
|
def test_extract_all(self):
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
|
|
for fpath, fdata in SMALL_TEST_DATA:
|
|
zipfp.writestr(fpath, fdata)
|
|
|
|
with zipfile.ZipFile(TESTFN2, "r") as zipfp:
|
|
zipfp.extractall()
|
|
for fpath, fdata in SMALL_TEST_DATA:
|
|
outfile = os.path.join(os.getcwd(), fpath)
|
|
|
|
with open(outfile, "rb") as fid:
|
|
self.assertEqual(fdata, fid.read())
|
|
os.remove(outfile)
|
|
|
|
# remove the test file subdirectories
|
|
rmtree(os.path.join(os.getcwd(), 'ziptest2dir'))
|
|
|
|
def check_file(self, filename, content):
|
|
self.assertTrue(os.path.isfile(filename))
|
|
with open(filename, 'rb') as f:
|
|
self.assertEqual(f.read(), content)
|
|
|
|
@skipUnless(TESTFN_UNICODE, "No Unicode filesystem semantics on this platform.")
|
|
def test_extract_unicode_filenames(self):
|
|
fnames = [u'foo.txt', os.path.basename(TESTFN_UNICODE)]
|
|
content = 'Test for unicode filename'
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp:
|
|
for fname in fnames:
|
|
zipfp.writestr(fname, content)
|
|
|
|
with zipfile.ZipFile(TESTFN2, "r") as zipfp:
|
|
for fname in fnames:
|
|
writtenfile = zipfp.extract(fname)
|
|
|
|
# make sure it was written to the right place
|
|
correctfile = os.path.join(os.getcwd(), fname)
|
|
correctfile = os.path.normpath(correctfile)
|
|
self.assertEqual(writtenfile, correctfile)
|
|
|
|
self.check_file(writtenfile, content)
|
|
os.remove(writtenfile)
|
|
|
|
def test_extract_hackers_arcnames(self):
|
|
hacknames = [
|
|
('../foo/bar', 'foo/bar'),
|
|
('foo/../bar', 'foo/bar'),
|
|
('foo/../../bar', 'foo/bar'),
|
|
('foo/bar/..', 'foo/bar'),
|
|
('./../foo/bar', 'foo/bar'),
|
|
('/foo/bar', 'foo/bar'),
|
|
('/foo/../bar', 'foo/bar'),
|
|
('/foo/../../bar', 'foo/bar'),
|
|
]
|
|
if os.path.sep == '\\':
|
|
hacknames.extend([
|
|
(r'..\foo\bar', 'foo/bar'),
|
|
(r'..\/foo\/bar', 'foo/bar'),
|
|
(r'foo/\..\/bar', 'foo/bar'),
|
|
(r'foo\/../\bar', 'foo/bar'),
|
|
(r'C:foo/bar', 'foo/bar'),
|
|
(r'C:/foo/bar', 'foo/bar'),
|
|
(r'C://foo/bar', 'foo/bar'),
|
|
(r'C:\foo\bar', 'foo/bar'),
|
|
(r'//conky/mountpoint/foo/bar', 'foo/bar'),
|
|
(r'\\conky\mountpoint\foo\bar', 'foo/bar'),
|
|
(r'///conky/mountpoint/foo/bar', 'conky/mountpoint/foo/bar'),
|
|
(r'\\\conky\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'),
|
|
(r'//conky//mountpoint/foo/bar', 'conky/mountpoint/foo/bar'),
|
|
(r'\\conky\\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'),
|
|
(r'//?/C:/foo/bar', 'foo/bar'),
|
|
(r'\\?\C:\foo\bar', 'foo/bar'),
|
|
(r'C:/../C:/foo/bar', 'C_/foo/bar'),
|
|
(r'a:b\c<d>e|f"g?h*i', 'b/c_d_e_f_g_h_i'),
|
|
('../../foo../../ba..r', 'foo/ba..r'),
|
|
])
|
|
else: # Unix
|
|
hacknames.extend([
|
|
('//foo/bar', 'foo/bar'),
|
|
('../../foo../../ba..r', 'foo../ba..r'),
|
|
(r'foo/..\bar', r'foo/..\bar'),
|
|
])
|
|
|
|
for arcname, fixedname in hacknames:
|
|
content = b'foobar' + arcname.encode()
|
|
with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipfp:
|
|
zinfo = zipfile.ZipInfo()
|
|
# preserve backslashes
|
|
zinfo.filename = arcname
|
|
zinfo.external_attr = 0o600 << 16
|
|
zipfp.writestr(zinfo, content)
|
|
|
|
arcname = arcname.replace(os.sep, "/")
|
|
targetpath = os.path.join('target', 'subdir', 'subsub')
|
|
correctfile = os.path.join(targetpath, *fixedname.split('/'))
|
|
|
|
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
|
|
writtenfile = zipfp.extract(arcname, targetpath)
|
|
self.assertEqual(writtenfile, correctfile,
|
|
msg="extract %r" % arcname)
|
|
self.check_file(correctfile, content)
|
|
rmtree('target')
|
|
|
|
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
|
|
zipfp.extractall(targetpath)
|
|
self.check_file(correctfile, content)
|
|
rmtree('target')
|
|
|
|
correctfile = os.path.join(os.getcwd(), *fixedname.split('/'))
|
|
|
|
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
|
|
writtenfile = zipfp.extract(arcname)
|
|
self.assertEqual(writtenfile, correctfile,
|
|
msg="extract %r" % arcname)
|
|
self.check_file(correctfile, content)
|
|
rmtree(fixedname.split('/')[0])
|
|
|
|
with zipfile.ZipFile(TESTFN2, 'r') as zipfp:
|
|
zipfp.extractall()
|
|
self.check_file(correctfile, content)
|
|
rmtree(fixedname.split('/')[0])
|
|
|
|
os.remove(TESTFN2)
|
|
|
|
def test_writestr_compression(self):
|
|
zipfp = zipfile.ZipFile(TESTFN2, "w")
|
|
zipfp.writestr("a.txt", "hello world", compress_type=zipfile.ZIP_STORED)
|
|
if zlib:
|
|
zipfp.writestr("b.txt", "hello world", compress_type=zipfile.ZIP_DEFLATED)
|
|
|
|
info = zipfp.getinfo('a.txt')
|
|
self.assertEqual(info.compress_type, zipfile.ZIP_STORED)
|
|
|
|
if zlib:
|
|
info = zipfp.getinfo('b.txt')
|
|
self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED)
|
|
|
|
|
|
def zip_test_writestr_permissions(self, f, compression):
|
|
# Make sure that writestr creates files with mode 0600,
|
|
# when it is passed a name rather than a ZipInfo instance.
|
|
|
|
self.make_test_archive(f, compression)
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
zinfo = zipfp.getinfo('strfile')
|
|
self.assertEqual(zinfo.external_attr, 0600 << 16)
|
|
|
|
def test_writestr_permissions(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_test_writestr_permissions(f, zipfile.ZIP_STORED)
|
|
|
|
def test_close(self):
|
|
"""Check that the zipfile is closed after the 'with' block."""
|
|
with zipfile.ZipFile(TESTFN2, "w") as zipfp:
|
|
for fpath, fdata in SMALL_TEST_DATA:
|
|
zipfp.writestr(fpath, fdata)
|
|
self.assertTrue(zipfp.fp is not None, 'zipfp is not open')
|
|
self.assertTrue(zipfp.fp is None, 'zipfp is not closed')
|
|
|
|
with zipfile.ZipFile(TESTFN2, "r") as zipfp:
|
|
self.assertTrue(zipfp.fp is not None, 'zipfp is not open')
|
|
self.assertTrue(zipfp.fp is None, 'zipfp is not closed')
|
|
|
|
def test_close_on_exception(self):
|
|
"""Check that the zipfile is closed if an exception is raised in the
|
|
'with' block."""
|
|
with zipfile.ZipFile(TESTFN2, "w") as zipfp:
|
|
for fpath, fdata in SMALL_TEST_DATA:
|
|
zipfp.writestr(fpath, fdata)
|
|
|
|
try:
|
|
with zipfile.ZipFile(TESTFN2, "r") as zipfp2:
|
|
raise zipfile.BadZipfile()
|
|
except zipfile.BadZipfile:
|
|
self.assertTrue(zipfp2.fp is None, 'zipfp is not closed')
|
|
|
|
def test_add_file_before_1980(self):
|
|
# Set atime and mtime to 1970-01-01
|
|
os.utime(TESTFN, (0, 0))
|
|
with zipfile.ZipFile(TESTFN2, "w") as zipfp:
|
|
self.assertRaises(ValueError, zipfp.write, TESTFN)
|
|
|
|
def tearDown(self):
|
|
unlink(TESTFN)
|
|
unlink(TESTFN2)
|
|
|
|
|
|
class TestZip64InSmallFiles(unittest.TestCase):
|
|
# These tests test the ZIP64 functionality without using large files,
|
|
# see test_zipfile64 for proper tests.
|
|
|
|
def setUp(self):
|
|
self._limit = zipfile.ZIP64_LIMIT
|
|
self._filecount_limit = zipfile.ZIP_FILECOUNT_LIMIT
|
|
zipfile.ZIP64_LIMIT = 1000
|
|
zipfile.ZIP_FILECOUNT_LIMIT = 9
|
|
|
|
line_gen = ("Test of zipfile line %d." % i
|
|
for i in range(0, FIXEDTEST_SIZE))
|
|
self.data = '\n'.join(line_gen)
|
|
|
|
# Make a source file with some lines
|
|
with open(TESTFN, "wb") as fp:
|
|
fp.write(self.data)
|
|
|
|
def large_file_exception_test(self, f, compression):
|
|
with zipfile.ZipFile(f, "w", compression) as zipfp:
|
|
self.assertRaises(zipfile.LargeZipFile,
|
|
zipfp.write, TESTFN, "another.name")
|
|
|
|
def large_file_exception_test2(self, f, compression):
|
|
with zipfile.ZipFile(f, "w", compression) as zipfp:
|
|
self.assertRaises(zipfile.LargeZipFile,
|
|
zipfp.writestr, "another.name", self.data)
|
|
|
|
def test_large_file_exception(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.large_file_exception_test(f, zipfile.ZIP_STORED)
|
|
self.large_file_exception_test2(f, zipfile.ZIP_STORED)
|
|
|
|
def zip_test(self, f, compression):
|
|
# Create the ZIP archive
|
|
with zipfile.ZipFile(f, "w", compression, allowZip64=True) as zipfp:
|
|
zipfp.write(TESTFN, "another.name")
|
|
zipfp.write(TESTFN, TESTFN)
|
|
zipfp.writestr("strfile", self.data)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r", compression) as zipfp:
|
|
self.assertEqual(zipfp.read(TESTFN), self.data)
|
|
self.assertEqual(zipfp.read("another.name"), self.data)
|
|
self.assertEqual(zipfp.read("strfile"), self.data)
|
|
|
|
# Print the ZIP directory
|
|
fp = StringIO()
|
|
stdout = sys.stdout
|
|
try:
|
|
sys.stdout = fp
|
|
zipfp.printdir()
|
|
finally:
|
|
sys.stdout = stdout
|
|
|
|
directory = fp.getvalue()
|
|
lines = directory.splitlines()
|
|
self.assertEqual(len(lines), 4) # Number of files + header
|
|
|
|
self.assertIn('File Name', lines[0])
|
|
self.assertIn('Modified', lines[0])
|
|
self.assertIn('Size', lines[0])
|
|
|
|
fn, date, time_, size = lines[1].split()
|
|
self.assertEqual(fn, 'another.name')
|
|
self.assertTrue(time.strptime(date, '%Y-%m-%d'))
|
|
self.assertTrue(time.strptime(time_, '%H:%M:%S'))
|
|
self.assertEqual(size, str(len(self.data)))
|
|
|
|
# Check the namelist
|
|
names = zipfp.namelist()
|
|
self.assertEqual(len(names), 3)
|
|
self.assertIn(TESTFN, names)
|
|
self.assertIn("another.name", names)
|
|
self.assertIn("strfile", names)
|
|
|
|
# Check infolist
|
|
infos = zipfp.infolist()
|
|
names = [i.filename for i in infos]
|
|
self.assertEqual(len(names), 3)
|
|
self.assertIn(TESTFN, names)
|
|
self.assertIn("another.name", names)
|
|
self.assertIn("strfile", names)
|
|
for i in infos:
|
|
self.assertEqual(i.file_size, len(self.data))
|
|
|
|
# check getinfo
|
|
for nm in (TESTFN, "another.name", "strfile"):
|
|
info = zipfp.getinfo(nm)
|
|
self.assertEqual(info.filename, nm)
|
|
self.assertEqual(info.file_size, len(self.data))
|
|
|
|
# Check that testzip doesn't raise an exception
|
|
zipfp.testzip()
|
|
|
|
def test_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_test(f, zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
def test_absolute_arcnames(self):
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED,
|
|
allowZip64=True) as zipfp:
|
|
zipfp.write(TESTFN, "/absolute")
|
|
|
|
with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp:
|
|
self.assertEqual(zipfp.namelist(), ["absolute"])
|
|
|
|
def test_too_many_files(self):
|
|
# This test checks that more than 64k files can be added to an archive,
|
|
# and that the resulting archive can be read properly by ZipFile
|
|
zipf = zipfile.ZipFile(TESTFN, mode="w", allowZip64=True)
|
|
zipf.debug = 100
|
|
numfiles = 15
|
|
for i in range(numfiles):
|
|
zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57))
|
|
self.assertEqual(len(zipf.namelist()), numfiles)
|
|
zipf.close()
|
|
|
|
zipf2 = zipfile.ZipFile(TESTFN, mode="r")
|
|
self.assertEqual(len(zipf2.namelist()), numfiles)
|
|
for i in range(numfiles):
|
|
content = zipf2.read("foo%08d" % i)
|
|
self.assertEqual(content, "%d" % (i**3 % 57))
|
|
zipf2.close()
|
|
|
|
def test_too_many_files_append(self):
|
|
zipf = zipfile.ZipFile(TESTFN, mode="w", allowZip64=False)
|
|
zipf.debug = 100
|
|
numfiles = 9
|
|
for i in range(numfiles):
|
|
zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57))
|
|
self.assertEqual(len(zipf.namelist()), numfiles)
|
|
with self.assertRaises(zipfile.LargeZipFile):
|
|
zipf.writestr("foo%08d" % numfiles, b'')
|
|
self.assertEqual(len(zipf.namelist()), numfiles)
|
|
zipf.close()
|
|
|
|
zipf = zipfile.ZipFile(TESTFN, mode="a", allowZip64=False)
|
|
zipf.debug = 100
|
|
self.assertEqual(len(zipf.namelist()), numfiles)
|
|
with self.assertRaises(zipfile.LargeZipFile):
|
|
zipf.writestr("foo%08d" % numfiles, b'')
|
|
self.assertEqual(len(zipf.namelist()), numfiles)
|
|
zipf.close()
|
|
|
|
zipf = zipfile.ZipFile(TESTFN, mode="a", allowZip64=True)
|
|
zipf.debug = 100
|
|
self.assertEqual(len(zipf.namelist()), numfiles)
|
|
numfiles2 = 15
|
|
for i in range(numfiles, numfiles2):
|
|
zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57))
|
|
self.assertEqual(len(zipf.namelist()), numfiles2)
|
|
zipf.close()
|
|
|
|
zipf2 = zipfile.ZipFile(TESTFN, mode="r")
|
|
self.assertEqual(len(zipf2.namelist()), numfiles2)
|
|
for i in range(numfiles2):
|
|
content = zipf2.read("foo%08d" % i)
|
|
self.assertEqual(content, "%d" % (i**3 % 57))
|
|
zipf2.close()
|
|
|
|
def tearDown(self):
|
|
zipfile.ZIP64_LIMIT = self._limit
|
|
zipfile.ZIP_FILECOUNT_LIMIT = self._filecount_limit
|
|
unlink(TESTFN)
|
|
unlink(TESTFN2)
|
|
|
|
|
|
class PyZipFileTests(unittest.TestCase):
|
|
def requiresWriteAccess(self, path):
|
|
if not os.access(path, os.W_OK):
|
|
self.skipTest('requires write access to the installed location')
|
|
filename = os.path.join(path, 'test_zipfile.try')
|
|
try:
|
|
fd = os.open(filename, os.O_WRONLY | os.O_CREAT)
|
|
os.close(fd)
|
|
except Exception:
|
|
self.skipTest('requires write access to the installed location')
|
|
unlink(filename)
|
|
|
|
def test_write_pyfile(self):
|
|
self.requiresWriteAccess(os.path.dirname(__file__))
|
|
with zipfile.PyZipFile(TemporaryFile(), "w") as zipfp:
|
|
fn = __file__
|
|
if fn.endswith('.pyc') or fn.endswith('.pyo'):
|
|
fn = fn[:-1]
|
|
|
|
zipfp.writepy(fn)
|
|
|
|
bn = os.path.basename(fn)
|
|
self.assertNotIn(bn, zipfp.namelist())
|
|
self.assertTrue(bn + 'o' in zipfp.namelist() or
|
|
bn + 'c' in zipfp.namelist())
|
|
|
|
with zipfile.PyZipFile(TemporaryFile(), "w") as zipfp:
|
|
fn = __file__
|
|
if fn.endswith(('.pyc', '.pyo')):
|
|
fn = fn[:-1]
|
|
|
|
zipfp.writepy(fn, "testpackage")
|
|
|
|
bn = "%s/%s" % ("testpackage", os.path.basename(fn))
|
|
self.assertNotIn(bn, zipfp.namelist())
|
|
self.assertTrue(bn + 'o' in zipfp.namelist() or
|
|
bn + 'c' in zipfp.namelist())
|
|
|
|
def test_write_python_package(self):
|
|
import email
|
|
packagedir = os.path.dirname(email.__file__)
|
|
self.requiresWriteAccess(packagedir)
|
|
|
|
with zipfile.PyZipFile(TemporaryFile(), "w") as zipfp:
|
|
zipfp.writepy(packagedir)
|
|
|
|
# Check for a couple of modules at different levels of the
|
|
# hierarchy
|
|
names = zipfp.namelist()
|
|
self.assertTrue('email/__init__.pyo' in names or
|
|
'email/__init__.pyc' in names)
|
|
self.assertTrue('email/mime/text.pyo' in names or
|
|
'email/mime/text.pyc' in names)
|
|
|
|
def test_write_python_directory(self):
|
|
os.mkdir(TESTFN2)
|
|
try:
|
|
with open(os.path.join(TESTFN2, "mod1.py"), "w") as fp:
|
|
fp.write("print(42)\n")
|
|
|
|
with open(os.path.join(TESTFN2, "mod2.py"), "w") as fp:
|
|
fp.write("print(42 * 42)\n")
|
|
|
|
with open(os.path.join(TESTFN2, "mod2.txt"), "w") as fp:
|
|
fp.write("bla bla bla\n")
|
|
|
|
zipfp = zipfile.PyZipFile(TemporaryFile(), "w")
|
|
zipfp.writepy(TESTFN2)
|
|
|
|
names = zipfp.namelist()
|
|
self.assertTrue('mod1.pyc' in names or 'mod1.pyo' in names)
|
|
self.assertTrue('mod2.pyc' in names or 'mod2.pyo' in names)
|
|
self.assertNotIn('mod2.txt', names)
|
|
|
|
finally:
|
|
rmtree(TESTFN2)
|
|
|
|
def test_write_non_pyfile(self):
|
|
with zipfile.PyZipFile(TemporaryFile(), "w") as zipfp:
|
|
with open(TESTFN, 'w') as fid:
|
|
fid.write('most definitely not a python file')
|
|
self.assertRaises(RuntimeError, zipfp.writepy, TESTFN)
|
|
os.remove(TESTFN)
|
|
|
|
|
|
class OtherTests(unittest.TestCase):
|
|
zips_with_bad_crc = {
|
|
zipfile.ZIP_STORED: (
|
|
b'PK\003\004\024\0\0\0\0\0 \213\212;:r'
|
|
b'\253\377\f\0\0\0\f\0\0\0\005\0\0\000af'
|
|
b'ilehello,AworldP'
|
|
b'K\001\002\024\003\024\0\0\0\0\0 \213\212;:'
|
|
b'r\253\377\f\0\0\0\f\0\0\0\005\0\0\0\0'
|
|
b'\0\0\0\0\0\0\0\200\001\0\0\0\000afi'
|
|
b'lePK\005\006\0\0\0\0\001\0\001\0003\000'
|
|
b'\0\0/\0\0\0\0\0'),
|
|
zipfile.ZIP_DEFLATED: (
|
|
b'PK\x03\x04\x14\x00\x00\x00\x08\x00n}\x0c=FA'
|
|
b'KE\x10\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af'
|
|
b'ile\xcbH\xcd\xc9\xc9W(\xcf/\xcaI\xc9\xa0'
|
|
b'=\x13\x00PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00n'
|
|
b'}\x0c=FAKE\x10\x00\x00\x00n\x00\x00\x00\x05'
|
|
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00'
|
|
b'\x00afilePK\x05\x06\x00\x00\x00\x00\x01\x00'
|
|
b'\x01\x003\x00\x00\x003\x00\x00\x00\x00\x00'),
|
|
}
|
|
|
|
def test_unicode_filenames(self):
|
|
with zipfile.ZipFile(TESTFN, "w") as zf:
|
|
zf.writestr(u"foo.txt", "Test for unicode filename")
|
|
zf.writestr(u"\xf6.txt", "Test for unicode filename")
|
|
self.assertIsInstance(zf.infolist()[0].filename, unicode)
|
|
|
|
with zipfile.ZipFile(TESTFN, "r") as zf:
|
|
self.assertEqual(zf.filelist[0].filename, "foo.txt")
|
|
self.assertEqual(zf.filelist[1].filename, u"\xf6.txt")
|
|
|
|
def test_create_non_existent_file_for_append(self):
|
|
if os.path.exists(TESTFN):
|
|
os.unlink(TESTFN)
|
|
|
|
filename = 'testfile.txt'
|
|
content = 'hello, world. this is some content.'
|
|
|
|
try:
|
|
with zipfile.ZipFile(TESTFN, 'a') as zf:
|
|
zf.writestr(filename, content)
|
|
except IOError:
|
|
self.fail('Could not append data to a non-existent zip file.')
|
|
|
|
self.assertTrue(os.path.exists(TESTFN))
|
|
|
|
with zipfile.ZipFile(TESTFN, 'r') as zf:
|
|
self.assertEqual(zf.read(filename), content)
|
|
|
|
def test_close_erroneous_file(self):
|
|
# This test checks that the ZipFile constructor closes the file object
|
|
# it opens if there's an error in the file. If it doesn't, the
|
|
# traceback holds a reference to the ZipFile object and, indirectly,
|
|
# the file object.
|
|
# On Windows, this causes the os.unlink() call to fail because the
|
|
# underlying file is still open. This is SF bug #412214.
|
|
#
|
|
with open(TESTFN, "w") as fp:
|
|
fp.write("this is not a legal zip file\n")
|
|
try:
|
|
zf = zipfile.ZipFile(TESTFN)
|
|
except zipfile.BadZipfile:
|
|
pass
|
|
|
|
def test_is_zip_erroneous_file(self):
|
|
"""Check that is_zipfile() correctly identifies non-zip files."""
|
|
# - passing a filename
|
|
with open(TESTFN, "w") as fp:
|
|
fp.write("this is not a legal zip file\n")
|
|
chk = zipfile.is_zipfile(TESTFN)
|
|
self.assertFalse(chk)
|
|
# - passing a file object
|
|
with open(TESTFN, "rb") as fp:
|
|
chk = zipfile.is_zipfile(fp)
|
|
self.assertTrue(not chk)
|
|
# - passing a file-like object
|
|
fp = StringIO()
|
|
fp.write("this is not a legal zip file\n")
|
|
chk = zipfile.is_zipfile(fp)
|
|
self.assertTrue(not chk)
|
|
fp.seek(0, 0)
|
|
chk = zipfile.is_zipfile(fp)
|
|
self.assertTrue(not chk)
|
|
|
|
def test_damaged_zipfile(self):
|
|
"""Check that zipfiles with missing bytes at the end raise BadZipFile."""
|
|
# - Create a valid zip file
|
|
fp = io.BytesIO()
|
|
with zipfile.ZipFile(fp, mode="w") as zipf:
|
|
zipf.writestr("foo.txt", b"O, for a Muse of Fire!")
|
|
zipfiledata = fp.getvalue()
|
|
|
|
# - Now create copies of it missing the last N bytes and make sure
|
|
# a BadZipFile exception is raised when we try to open it
|
|
for N in range(len(zipfiledata)):
|
|
fp = io.BytesIO(zipfiledata[:N])
|
|
self.assertRaises(zipfile.BadZipfile, zipfile.ZipFile, fp)
|
|
|
|
def test_is_zip_valid_file(self):
|
|
"""Check that is_zipfile() correctly identifies zip files."""
|
|
# - passing a filename
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
chk = zipfile.is_zipfile(TESTFN)
|
|
self.assertTrue(chk)
|
|
# - passing a file object
|
|
with open(TESTFN, "rb") as fp:
|
|
chk = zipfile.is_zipfile(fp)
|
|
self.assertTrue(chk)
|
|
fp.seek(0, 0)
|
|
zip_contents = fp.read()
|
|
# - passing a file-like object
|
|
fp = StringIO()
|
|
fp.write(zip_contents)
|
|
chk = zipfile.is_zipfile(fp)
|
|
self.assertTrue(chk)
|
|
fp.seek(0, 0)
|
|
chk = zipfile.is_zipfile(fp)
|
|
self.assertTrue(chk)
|
|
|
|
def test_non_existent_file_raises_IOError(self):
|
|
# make sure we don't raise an AttributeError when a partially-constructed
|
|
# ZipFile instance is finalized; this tests for regression on SF tracker
|
|
# bug #403871.
|
|
|
|
# The bug we're testing for caused an AttributeError to be raised
|
|
# when a ZipFile instance was created for a file that did not
|
|
# exist; the .fp member was not initialized but was needed by the
|
|
# __del__() method. Since the AttributeError is in the __del__(),
|
|
# it is ignored, but the user should be sufficiently annoyed by
|
|
# the message on the output that regression will be noticed
|
|
# quickly.
|
|
self.assertRaises(IOError, zipfile.ZipFile, TESTFN)
|
|
|
|
def test_empty_file_raises_BadZipFile(self):
|
|
with open(TESTFN, 'w') as f:
|
|
pass
|
|
self.assertRaises(zipfile.BadZipfile, zipfile.ZipFile, TESTFN)
|
|
|
|
with open(TESTFN, 'w') as fp:
|
|
fp.write("short file")
|
|
self.assertRaises(zipfile.BadZipfile, zipfile.ZipFile, TESTFN)
|
|
|
|
def test_closed_zip_raises_RuntimeError(self):
|
|
"""Verify that testzip() doesn't swallow inappropriate exceptions."""
|
|
data = StringIO()
|
|
with zipfile.ZipFile(data, mode="w") as zipf:
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
|
|
# This is correct; calling .read on a closed ZipFile should raise
|
|
# a RuntimeError, and so should calling .testzip. An earlier
|
|
# version of .testzip would swallow this exception (and any other)
|
|
# and report that the first file in the archive was corrupt.
|
|
self.assertRaises(RuntimeError, zipf.read, "foo.txt")
|
|
self.assertRaises(RuntimeError, zipf.open, "foo.txt")
|
|
self.assertRaises(RuntimeError, zipf.testzip)
|
|
self.assertRaises(RuntimeError, zipf.writestr, "bogus.txt", "bogus")
|
|
with open(TESTFN, 'w') as fid:
|
|
fid.write('zipfile test data')
|
|
self.assertRaises(RuntimeError, zipf.write, TESTFN)
|
|
|
|
def test_bad_constructor_mode(self):
|
|
"""Check that bad modes passed to ZipFile constructor are caught."""
|
|
self.assertRaises(RuntimeError, zipfile.ZipFile, TESTFN, "q")
|
|
|
|
def test_bad_open_mode(self):
|
|
"""Check that bad modes passed to ZipFile.open are caught."""
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
|
|
with zipfile.ZipFile(TESTFN, mode="r") as zipf:
|
|
# read the data to make sure the file is there
|
|
zipf.read("foo.txt")
|
|
self.assertRaises(RuntimeError, zipf.open, "foo.txt", "q")
|
|
|
|
def test_read0(self):
|
|
"""Check that calling read(0) on a ZipExtFile object returns an empty
|
|
string and doesn't advance file pointer."""
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
# read the data to make sure the file is there
|
|
with zipf.open("foo.txt") as f:
|
|
for i in xrange(FIXEDTEST_SIZE):
|
|
self.assertEqual(f.read(0), '')
|
|
|
|
self.assertEqual(f.read(), "O, for a Muse of Fire!")
|
|
|
|
def test_open_non_existent_item(self):
|
|
"""Check that attempting to call open() for an item that doesn't
|
|
exist in the archive raises a RuntimeError."""
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
self.assertRaises(KeyError, zipf.open, "foo.txt", "r")
|
|
|
|
def test_bad_compression_mode(self):
|
|
"""Check that bad compression methods passed to ZipFile.open are
|
|
caught."""
|
|
self.assertRaises(RuntimeError, zipfile.ZipFile, TESTFN, "w", -1)
|
|
|
|
def test_unsupported_compression(self):
|
|
# data is declared as shrunk, but actually deflated
|
|
data = (b'PK\x03\x04.\x00\x00\x00\x01\x00\xe4C\xa1@\x00\x00\x00'
|
|
b'\x00\x02\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00x\x03\x00PK\x01'
|
|
b'\x02.\x03.\x00\x00\x00\x01\x00\xe4C\xa1@\x00\x00\x00\x00\x02\x00\x00'
|
|
b'\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
|
|
b'\x80\x01\x00\x00\x00\x00xPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x00'
|
|
b'/\x00\x00\x00!\x00\x00\x00\x00\x00')
|
|
with zipfile.ZipFile(io.BytesIO(data), 'r') as zipf:
|
|
self.assertRaises(NotImplementedError, zipf.open, 'x')
|
|
|
|
def test_null_byte_in_filename(self):
|
|
"""Check that a filename containing a null byte is properly
|
|
terminated."""
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
zipf.writestr("foo.txt\x00qqq", "O, for a Muse of Fire!")
|
|
self.assertEqual(zipf.namelist(), ['foo.txt'])
|
|
|
|
def test_struct_sizes(self):
|
|
"""Check that ZIP internal structure sizes are calculated correctly."""
|
|
self.assertEqual(zipfile.sizeEndCentDir, 22)
|
|
self.assertEqual(zipfile.sizeCentralDir, 46)
|
|
self.assertEqual(zipfile.sizeEndCentDir64, 56)
|
|
self.assertEqual(zipfile.sizeEndCentDir64Locator, 20)
|
|
|
|
def test_comments(self):
|
|
"""Check that comments on the archive are handled properly."""
|
|
|
|
# check default comment is empty
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
self.assertEqual(zipf.comment, '')
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
|
|
with zipfile.ZipFile(TESTFN, mode="r") as zipf:
|
|
self.assertEqual(zipf.comment, '')
|
|
|
|
# check a simple short comment
|
|
comment = 'Bravely taking to his feet, he beat a very brave retreat.'
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
zipf.comment = comment
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
with zipfile.ZipFile(TESTFN, mode="r") as zipf:
|
|
self.assertEqual(zipf.comment, comment)
|
|
|
|
# check a comment of max length
|
|
comment2 = ''.join(['%d' % (i**3 % 10) for i in xrange((1 << 16)-1)])
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
zipf.comment = comment2
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
|
|
with zipfile.ZipFile(TESTFN, mode="r") as zipf:
|
|
self.assertEqual(zipf.comment, comment2)
|
|
|
|
# check a comment that is too long is truncated
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
with check_warnings(('', UserWarning)):
|
|
zipf.comment = comment2 + 'oops'
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
with zipfile.ZipFile(TESTFN, mode="r") as zipf:
|
|
self.assertEqual(zipf.comment, comment2)
|
|
|
|
def test_change_comment_in_empty_archive(self):
|
|
with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf:
|
|
self.assertFalse(zipf.filelist)
|
|
zipf.comment = b"this is a comment"
|
|
with zipfile.ZipFile(TESTFN, "r") as zipf:
|
|
self.assertEqual(zipf.comment, b"this is a comment")
|
|
|
|
def test_change_comment_in_nonempty_archive(self):
|
|
with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf:
|
|
zipf.writestr("foo.txt", "O, for a Muse of Fire!")
|
|
with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf:
|
|
self.assertTrue(zipf.filelist)
|
|
zipf.comment = b"this is a comment"
|
|
with zipfile.ZipFile(TESTFN, "r") as zipf:
|
|
self.assertEqual(zipf.comment, b"this is a comment")
|
|
|
|
def check_testzip_with_bad_crc(self, compression):
|
|
"""Tests that files with bad CRCs return their name from testzip."""
|
|
zipdata = self.zips_with_bad_crc[compression]
|
|
|
|
with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf:
|
|
# testzip returns the name of the first corrupt file, or None
|
|
self.assertEqual('afile', zipf.testzip())
|
|
|
|
def test_testzip_with_bad_crc_stored(self):
|
|
self.check_testzip_with_bad_crc(zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_testzip_with_bad_crc_deflated(self):
|
|
self.check_testzip_with_bad_crc(zipfile.ZIP_DEFLATED)
|
|
|
|
def check_read_with_bad_crc(self, compression):
|
|
"""Tests that files with bad CRCs raise a BadZipfile exception when read."""
|
|
zipdata = self.zips_with_bad_crc[compression]
|
|
|
|
# Using ZipFile.read()
|
|
with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf:
|
|
self.assertRaises(zipfile.BadZipfile, zipf.read, 'afile')
|
|
|
|
# Using ZipExtFile.read()
|
|
with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf:
|
|
with zipf.open('afile', 'r') as corrupt_file:
|
|
self.assertRaises(zipfile.BadZipfile, corrupt_file.read)
|
|
|
|
# Same with small reads (in order to exercise the buffering logic)
|
|
with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf:
|
|
with zipf.open('afile', 'r') as corrupt_file:
|
|
corrupt_file.MIN_READ_SIZE = 2
|
|
with self.assertRaises(zipfile.BadZipfile):
|
|
while corrupt_file.read(2):
|
|
pass
|
|
|
|
def test_read_with_bad_crc_stored(self):
|
|
self.check_read_with_bad_crc(zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_read_with_bad_crc_deflated(self):
|
|
self.check_read_with_bad_crc(zipfile.ZIP_DEFLATED)
|
|
|
|
def check_read_return_size(self, compression):
|
|
# Issue #9837: ZipExtFile.read() shouldn't return more bytes
|
|
# than requested.
|
|
for test_size in (1, 4095, 4096, 4097, 16384):
|
|
file_size = test_size + 1
|
|
junk = getrandbytes(file_size)
|
|
with zipfile.ZipFile(io.BytesIO(), "w", compression) as zipf:
|
|
zipf.writestr('foo', junk)
|
|
with zipf.open('foo', 'r') as fp:
|
|
buf = fp.read(test_size)
|
|
self.assertEqual(len(buf), test_size)
|
|
|
|
def test_read_return_size_stored(self):
|
|
self.check_read_return_size(zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_read_return_size_deflated(self):
|
|
self.check_read_return_size(zipfile.ZIP_DEFLATED)
|
|
|
|
def test_empty_zipfile(self):
|
|
# Check that creating a file in 'w' or 'a' mode and closing without
|
|
# adding any files to the archives creates a valid empty ZIP file
|
|
with zipfile.ZipFile(TESTFN, mode="w") as zipf:
|
|
pass
|
|
try:
|
|
zipf = zipfile.ZipFile(TESTFN, mode="r")
|
|
zipf.close()
|
|
except zipfile.BadZipfile:
|
|
self.fail("Unable to create empty ZIP file in 'w' mode")
|
|
|
|
with zipfile.ZipFile(TESTFN, mode="a") as zipf:
|
|
pass
|
|
try:
|
|
zipf = zipfile.ZipFile(TESTFN, mode="r")
|
|
zipf.close()
|
|
except:
|
|
self.fail("Unable to create empty ZIP file in 'a' mode")
|
|
|
|
def test_open_empty_file(self):
|
|
# Issue 1710703: Check that opening a file with less than 22 bytes
|
|
# raises a BadZipfile exception (rather than the previously unhelpful
|
|
# IOError)
|
|
with open(TESTFN, 'w') as f:
|
|
pass
|
|
self.assertRaises(zipfile.BadZipfile, zipfile.ZipFile, TESTFN, 'r')
|
|
|
|
def test_create_zipinfo_before_1980(self):
|
|
self.assertRaises(ValueError,
|
|
zipfile.ZipInfo, 'seventies', (1979, 1, 1, 0, 0, 0))
|
|
|
|
def test_zipfile_with_short_extra_field(self):
|
|
"""If an extra field in the header is less than 4 bytes, skip it."""
|
|
zipdata = (
|
|
b'PK\x03\x04\x14\x00\x00\x00\x00\x00\x93\x9b\xad@\x8b\x9e'
|
|
b'\xd9\xd3\x01\x00\x00\x00\x01\x00\x00\x00\x03\x00\x03\x00ab'
|
|
b'c\x00\x00\x00APK\x01\x02\x14\x03\x14\x00\x00\x00\x00'
|
|
b'\x00\x93\x9b\xad@\x8b\x9e\xd9\xd3\x01\x00\x00\x00\x01\x00\x00'
|
|
b'\x00\x03\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa4\x81\x00'
|
|
b'\x00\x00\x00abc\x00\x00PK\x05\x06\x00\x00\x00\x00'
|
|
b'\x01\x00\x01\x003\x00\x00\x00%\x00\x00\x00\x00\x00'
|
|
)
|
|
with zipfile.ZipFile(io.BytesIO(zipdata), 'r') as zipf:
|
|
# testzip returns the name of the first corrupt file, or None
|
|
self.assertIsNone(zipf.testzip())
|
|
|
|
def tearDown(self):
|
|
unlink(TESTFN)
|
|
unlink(TESTFN2)
|
|
|
|
|
|
class DecryptionTests(unittest.TestCase):
|
|
"""Check that ZIP decryption works. Since the library does not
|
|
support encryption at the moment, we use a pre-generated encrypted
|
|
ZIP file."""
|
|
|
|
data = (
|
|
'PK\x03\x04\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00\x1a\x00'
|
|
'\x00\x00\x08\x00\x00\x00test.txt\xfa\x10\xa0gly|\xfa-\xc5\xc0=\xf9y'
|
|
'\x18\xe0\xa8r\xb3Z}Lg\xbc\xae\xf9|\x9b\x19\xe4\x8b\xba\xbb)\x8c\xb0\xdbl'
|
|
'PK\x01\x02\x14\x00\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00'
|
|
'\x1a\x00\x00\x00\x08\x00\x00\x00\x00\x00\x00\x00\x01\x00 \x00\xb6\x81'
|
|
'\x00\x00\x00\x00test.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x006\x00'
|
|
'\x00\x00L\x00\x00\x00\x00\x00' )
|
|
data2 = (
|
|
'PK\x03\x04\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02'
|
|
'\x00\x00\x04\x00\x15\x00zeroUT\t\x00\x03\xd6\x8b\x92G\xda\x8b\x92GUx\x04'
|
|
'\x00\xe8\x03\xe8\x03\xc7<M\xb5a\xceX\xa3Y&\x8b{oE\xd7\x9d\x8c\x98\x02\xc0'
|
|
'PK\x07\x08xu\xaa\xb2\x14\x00\x00\x00\x00\x02\x00\x00PK\x01\x02\x17\x03'
|
|
'\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02\x00\x00'
|
|
'\x04\x00\r\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa4\x81\x00\x00\x00\x00ze'
|
|
'roUT\x05\x00\x03\xd6\x8b\x92GUx\x00\x00PK\x05\x06\x00\x00\x00\x00\x01'
|
|
'\x00\x01\x00?\x00\x00\x00[\x00\x00\x00\x00\x00' )
|
|
|
|
plain = 'zipfile.py encryption test'
|
|
plain2 = '\x00'*512
|
|
|
|
def setUp(self):
|
|
with open(TESTFN, "wb") as fp:
|
|
fp.write(self.data)
|
|
self.zip = zipfile.ZipFile(TESTFN, "r")
|
|
with open(TESTFN2, "wb") as fp:
|
|
fp.write(self.data2)
|
|
self.zip2 = zipfile.ZipFile(TESTFN2, "r")
|
|
|
|
def tearDown(self):
|
|
self.zip.close()
|
|
os.unlink(TESTFN)
|
|
self.zip2.close()
|
|
os.unlink(TESTFN2)
|
|
|
|
def test_no_password(self):
|
|
# Reading the encrypted file without password
|
|
# must generate a RunTime exception
|
|
self.assertRaises(RuntimeError, self.zip.read, "test.txt")
|
|
self.assertRaises(RuntimeError, self.zip2.read, "zero")
|
|
|
|
def test_bad_password(self):
|
|
self.zip.setpassword("perl")
|
|
self.assertRaises(RuntimeError, self.zip.read, "test.txt")
|
|
self.zip2.setpassword("perl")
|
|
self.assertRaises(RuntimeError, self.zip2.read, "zero")
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_good_password(self):
|
|
self.zip.setpassword("python")
|
|
self.assertEqual(self.zip.read("test.txt"), self.plain)
|
|
self.zip2.setpassword("12345")
|
|
self.assertEqual(self.zip2.read("zero"), self.plain2)
|
|
|
|
|
|
class TestsWithRandomBinaryFiles(unittest.TestCase):
|
|
def setUp(self):
|
|
datacount = randint(16, 64)*1024 + randint(1, 1024)
|
|
self.data = ''.join(struct.pack('<f', random()*randint(-1000, 1000))
|
|
for i in xrange(datacount))
|
|
|
|
# Make a source file with some lines
|
|
with open(TESTFN, "wb") as fp:
|
|
fp.write(self.data)
|
|
|
|
def tearDown(self):
|
|
unlink(TESTFN)
|
|
unlink(TESTFN2)
|
|
|
|
def make_test_archive(self, f, compression):
|
|
# Create the ZIP archive
|
|
with zipfile.ZipFile(f, "w", compression) as zipfp:
|
|
zipfp.write(TESTFN, "another.name")
|
|
zipfp.write(TESTFN, TESTFN)
|
|
|
|
def zip_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r", compression) as zipfp:
|
|
testdata = zipfp.read(TESTFN)
|
|
self.assertEqual(len(testdata), len(self.data))
|
|
self.assertEqual(testdata, self.data)
|
|
self.assertEqual(zipfp.read("another.name"), self.data)
|
|
|
|
def test_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_test(f, zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
|
|
self.zip_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
def zip_open_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r", compression) as zipfp:
|
|
zipdata1 = []
|
|
with zipfp.open(TESTFN) as zipopen1:
|
|
while True:
|
|
read_data = zipopen1.read(256)
|
|
if not read_data:
|
|
break
|
|
zipdata1.append(read_data)
|
|
|
|
zipdata2 = []
|
|
with zipfp.open("another.name") as zipopen2:
|
|
while True:
|
|
read_data = zipopen2.read(256)
|
|
if not read_data:
|
|
break
|
|
zipdata2.append(read_data)
|
|
|
|
testdata1 = ''.join(zipdata1)
|
|
self.assertEqual(len(testdata1), len(self.data))
|
|
self.assertEqual(testdata1, self.data)
|
|
|
|
testdata2 = ''.join(zipdata2)
|
|
self.assertEqual(len(testdata2), len(self.data))
|
|
self.assertEqual(testdata2, self.data)
|
|
|
|
def test_open_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_open_test(f, zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_open_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
|
|
self.zip_open_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
def zip_random_open_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r", compression) as zipfp:
|
|
zipdata1 = []
|
|
with zipfp.open(TESTFN) as zipopen1:
|
|
while True:
|
|
read_data = zipopen1.read(randint(1, 1024))
|
|
if not read_data:
|
|
break
|
|
zipdata1.append(read_data)
|
|
|
|
testdata = ''.join(zipdata1)
|
|
self.assertEqual(len(testdata), len(self.data))
|
|
self.assertEqual(testdata, self.data)
|
|
|
|
def test_random_open_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.zip_random_open_test(f, zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_random_open_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), io.BytesIO()):
|
|
self.zip_random_open_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
class TestsWithMultipleOpens(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.data1 = b'111' + getrandbytes(10000)
|
|
cls.data2 = b'222' + getrandbytes(10000)
|
|
|
|
def make_test_archive(self, f):
|
|
# Create the ZIP archive
|
|
with zipfile.ZipFile(f, "w", zipfile.ZIP_DEFLATED) as zipfp:
|
|
zipfp.writestr('ones', self.data1)
|
|
zipfp.writestr('twos', self.data2)
|
|
|
|
def test_same_file(self):
|
|
# Verify that (when the ZipFile is in control of creating file objects)
|
|
# multiple open() calls can be made without interfering with each other.
|
|
self.make_test_archive(TESTFN2)
|
|
with zipfile.ZipFile(TESTFN2, mode="r") as zipf:
|
|
with zipf.open('ones') as zopen1, zipf.open('ones') as zopen2:
|
|
data1 = zopen1.read(500)
|
|
data2 = zopen2.read(500)
|
|
data1 += zopen1.read()
|
|
data2 += zopen2.read()
|
|
self.assertEqual(data1, data2)
|
|
self.assertEqual(data1, self.data1)
|
|
|
|
def test_different_file(self):
|
|
# Verify that (when the ZipFile is in control of creating file objects)
|
|
# multiple open() calls can be made without interfering with each other.
|
|
self.make_test_archive(TESTFN2)
|
|
with zipfile.ZipFile(TESTFN2, mode="r") as zipf:
|
|
with zipf.open('ones') as zopen1, zipf.open('twos') as zopen2:
|
|
data1 = zopen1.read(500)
|
|
data2 = zopen2.read(500)
|
|
data1 += zopen1.read()
|
|
data2 += zopen2.read()
|
|
self.assertEqual(data1, self.data1)
|
|
self.assertEqual(data2, self.data2)
|
|
|
|
def test_interleaved(self):
|
|
# Verify that (when the ZipFile is in control of creating file objects)
|
|
# multiple open() calls can be made without interfering with each other.
|
|
self.make_test_archive(TESTFN2)
|
|
with zipfile.ZipFile(TESTFN2, mode="r") as zipf:
|
|
with zipf.open('ones') as zopen1:
|
|
data1 = zopen1.read(500)
|
|
with zipf.open('twos') as zopen2:
|
|
data2 = zopen2.read(500)
|
|
data1 += zopen1.read()
|
|
data2 += zopen2.read()
|
|
self.assertEqual(data1, self.data1)
|
|
self.assertEqual(data2, self.data2)
|
|
|
|
def test_read_after_close(self):
|
|
self.make_test_archive(TESTFN2)
|
|
zopen1 = zopen2 = None
|
|
try:
|
|
with zipfile.ZipFile(TESTFN2, 'r') as zipf:
|
|
zopen1 = zipf.open('ones')
|
|
zopen2 = zipf.open('twos')
|
|
data1 = zopen1.read(500)
|
|
data2 = zopen2.read(500)
|
|
data1 += zopen1.read()
|
|
data2 += zopen2.read()
|
|
finally:
|
|
if zopen1:
|
|
zopen1.close()
|
|
if zopen2:
|
|
zopen2.close()
|
|
self.assertEqual(data1, self.data1)
|
|
self.assertEqual(data2, self.data2)
|
|
|
|
def test_read_after_write(self):
|
|
with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_DEFLATED) as zipf:
|
|
zipf.writestr('ones', self.data1)
|
|
zipf.writestr('twos', self.data2)
|
|
with zipf.open('ones') as zopen1:
|
|
data1 = zopen1.read(500)
|
|
self.assertEqual(data1, self.data1[:500])
|
|
with zipfile.ZipFile(TESTFN2, 'r') as zipf:
|
|
data1 = zipf.read('ones')
|
|
data2 = zipf.read('twos')
|
|
self.assertEqual(data1, self.data1)
|
|
self.assertEqual(data2, self.data2)
|
|
|
|
def test_write_after_read(self):
|
|
with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_DEFLATED) as zipf:
|
|
zipf.writestr('ones', self.data1)
|
|
with zipf.open('ones') as zopen1:
|
|
zopen1.read(500)
|
|
zipf.writestr('twos', self.data2)
|
|
with zipfile.ZipFile(TESTFN2, 'r') as zipf:
|
|
data1 = zipf.read('ones')
|
|
data2 = zipf.read('twos')
|
|
self.assertEqual(data1, self.data1)
|
|
self.assertEqual(data2, self.data2)
|
|
|
|
def test_many_opens(self):
|
|
# Verify that read() and open() promptly close the file descriptor,
|
|
# and don't rely on the garbage collector to free resources.
|
|
self.make_test_archive(TESTFN2)
|
|
with zipfile.ZipFile(TESTFN2, mode="r") as zipf:
|
|
for x in range(100):
|
|
zipf.read('ones')
|
|
with zipf.open('ones') as zopen1:
|
|
pass
|
|
with open(os.devnull) as f:
|
|
self.assertLess(f.fileno(), 100)
|
|
|
|
def tearDown(self):
|
|
unlink(TESTFN2)
|
|
|
|
|
|
class TestWithDirectory(unittest.TestCase):
|
|
def setUp(self):
|
|
os.mkdir(TESTFN2)
|
|
|
|
def test_extract_dir(self):
|
|
with zipfile.ZipFile(findfile("zipdir.zip")) as zipf:
|
|
zipf.extractall(TESTFN2)
|
|
self.assertTrue(os.path.isdir(os.path.join(TESTFN2, "a")))
|
|
self.assertTrue(os.path.isdir(os.path.join(TESTFN2, "a", "b")))
|
|
self.assertTrue(os.path.exists(os.path.join(TESTFN2, "a", "b", "c")))
|
|
|
|
def test_bug_6050(self):
|
|
# Extraction should succeed if directories already exist
|
|
os.mkdir(os.path.join(TESTFN2, "a"))
|
|
self.test_extract_dir()
|
|
|
|
def test_write_dir(self):
|
|
dirpath = os.path.join(TESTFN2, "x")
|
|
os.mkdir(dirpath)
|
|
mode = os.stat(dirpath).st_mode & 0xFFFF
|
|
with zipfile.ZipFile(TESTFN, "w") as zipf:
|
|
zipf.write(dirpath)
|
|
zinfo = zipf.filelist[0]
|
|
self.assertTrue(zinfo.filename.endswith("/x/"))
|
|
self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10)
|
|
zipf.write(dirpath, "y")
|
|
zinfo = zipf.filelist[1]
|
|
self.assertTrue(zinfo.filename, "y/")
|
|
self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10)
|
|
with zipfile.ZipFile(TESTFN, "r") as zipf:
|
|
zinfo = zipf.filelist[0]
|
|
self.assertTrue(zinfo.filename.endswith("/x/"))
|
|
self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10)
|
|
zinfo = zipf.filelist[1]
|
|
self.assertTrue(zinfo.filename, "y/")
|
|
self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10)
|
|
target = os.path.join(TESTFN2, "target")
|
|
os.mkdir(target)
|
|
zipf.extractall(target)
|
|
self.assertTrue(os.path.isdir(os.path.join(target, "y")))
|
|
self.assertEqual(len(os.listdir(target)), 2)
|
|
|
|
def test_writestr_dir(self):
|
|
os.mkdir(os.path.join(TESTFN2, "x"))
|
|
with zipfile.ZipFile(TESTFN, "w") as zipf:
|
|
zipf.writestr("x/", b'')
|
|
zinfo = zipf.filelist[0]
|
|
self.assertEqual(zinfo.filename, "x/")
|
|
self.assertEqual(zinfo.external_attr, (0o40775 << 16) | 0x10)
|
|
with zipfile.ZipFile(TESTFN, "r") as zipf:
|
|
zinfo = zipf.filelist[0]
|
|
self.assertTrue(zinfo.filename.endswith("x/"))
|
|
self.assertEqual(zinfo.external_attr, (0o40775 << 16) | 0x10)
|
|
target = os.path.join(TESTFN2, "target")
|
|
os.mkdir(target)
|
|
zipf.extractall(target)
|
|
self.assertTrue(os.path.isdir(os.path.join(target, "x")))
|
|
self.assertEqual(os.listdir(target), ["x"])
|
|
|
|
def tearDown(self):
|
|
rmtree(TESTFN2)
|
|
if os.path.exists(TESTFN):
|
|
unlink(TESTFN)
|
|
|
|
|
|
class UniversalNewlineTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.line_gen = ["Test of zipfile line %d." % i
|
|
for i in xrange(FIXEDTEST_SIZE)]
|
|
self.seps = ('\r', '\r\n', '\n')
|
|
self.arcdata, self.arcfiles = {}, {}
|
|
for n, s in enumerate(self.seps):
|
|
self.arcdata[s] = s.join(self.line_gen) + s
|
|
self.arcfiles[s] = '%s-%d' % (TESTFN, n)
|
|
with open(self.arcfiles[s], "wb") as fid:
|
|
fid.write(self.arcdata[s])
|
|
|
|
def make_test_archive(self, f, compression):
|
|
# Create the ZIP archive
|
|
with zipfile.ZipFile(f, "w", compression) as zipfp:
|
|
for fn in self.arcfiles.values():
|
|
zipfp.write(fn, fn)
|
|
|
|
def read_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
for sep, fn in self.arcfiles.items():
|
|
with zipfp.open(fn, "rU") as fp:
|
|
zipdata = fp.read()
|
|
self.assertEqual(self.arcdata[sep], zipdata)
|
|
|
|
def readline_read_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
zipfp = zipfile.ZipFile(f, "r")
|
|
for sep, fn in self.arcfiles.items():
|
|
with zipfp.open(fn, "rU") as zipopen:
|
|
data = ''
|
|
while True:
|
|
read = zipopen.readline()
|
|
if not read:
|
|
break
|
|
data += read
|
|
|
|
read = zipopen.read(5)
|
|
if not read:
|
|
break
|
|
data += read
|
|
|
|
self.assertEqual(data, self.arcdata['\n'])
|
|
|
|
zipfp.close()
|
|
|
|
def readline_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
for sep, fn in self.arcfiles.items():
|
|
with zipfp.open(fn, "rU") as zipopen:
|
|
for line in self.line_gen:
|
|
linedata = zipopen.readline()
|
|
self.assertEqual(linedata, line + '\n')
|
|
|
|
def readlines_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
for sep, fn in self.arcfiles.items():
|
|
with zipfp.open(fn, "rU") as fp:
|
|
ziplines = fp.readlines()
|
|
for line, zipline in zip(self.line_gen, ziplines):
|
|
self.assertEqual(zipline, line + '\n')
|
|
|
|
def iterlines_test(self, f, compression):
|
|
self.make_test_archive(f, compression)
|
|
|
|
# Read the ZIP archive
|
|
with zipfile.ZipFile(f, "r") as zipfp:
|
|
for sep, fn in self.arcfiles.items():
|
|
with zipfp.open(fn, "rU") as fid:
|
|
for line, zipline in zip(self.line_gen, fid):
|
|
self.assertEqual(zipline, line + '\n')
|
|
|
|
def test_read_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.read_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_readline_read_stored(self):
|
|
# Issue #7610: calls to readline() interleaved with calls to read().
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.readline_read_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_readline_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.readline_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_readlines_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.readlines_test(f, zipfile.ZIP_STORED)
|
|
|
|
def test_iterlines_stored(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.iterlines_test(f, zipfile.ZIP_STORED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_read_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.read_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_readline_read_deflated(self):
|
|
# Issue #7610: calls to readline() interleaved with calls to read().
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.readline_read_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_readline_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.readline_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_readlines_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.readlines_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_iterlines_deflated(self):
|
|
for f in (TESTFN2, TemporaryFile(), StringIO()):
|
|
self.iterlines_test(f, zipfile.ZIP_DEFLATED)
|
|
|
|
def tearDown(self):
|
|
for sep, fn in self.arcfiles.items():
|
|
os.remove(fn)
|
|
unlink(TESTFN)
|
|
unlink(TESTFN2)
|
|
|
|
|
|
class CommandLineTest(unittest.TestCase):
|
|
|
|
def zipfilecmd(self, *args, **kwargs):
|
|
rc, out, err = script_helper.assert_python_ok('-m', 'zipfile', *args,
|
|
**kwargs)
|
|
return out.replace(os.linesep.encode(), b'\n')
|
|
|
|
def zipfilecmd_failure(self, *args):
|
|
return script_helper.assert_python_failure('-m', 'zipfile', *args)
|
|
|
|
def test_test_command(self):
|
|
zip_name = findfile('zipdir.zip')
|
|
out = self.zipfilecmd('-t', zip_name)
|
|
self.assertEqual(out.rstrip(), b'Done testing')
|
|
zip_name = findfile('testtar.tar')
|
|
rc, out, err = self.zipfilecmd_failure('-t', zip_name)
|
|
self.assertEqual(out, b'')
|
|
|
|
def test_list_command(self):
|
|
zip_name = findfile('zipdir.zip')
|
|
with captured_stdout() as t, zipfile.ZipFile(zip_name, 'r') as tf:
|
|
tf.printdir()
|
|
expected = t.getvalue().encode('ascii', 'backslashreplace')
|
|
out = self.zipfilecmd('-l', zip_name,
|
|
PYTHONIOENCODING='ascii:backslashreplace')
|
|
self.assertEqual(out, expected)
|
|
|
|
@skipUnless(zlib, "requires zlib")
|
|
def test_create_command(self):
|
|
self.addCleanup(unlink, TESTFN)
|
|
with open(TESTFN, 'w') as f:
|
|
f.write('test 1')
|
|
os.mkdir(TESTFNDIR)
|
|
self.addCleanup(rmtree, TESTFNDIR)
|
|
with open(os.path.join(TESTFNDIR, 'file.txt'), 'w') as f:
|
|
f.write('test 2')
|
|
files = [TESTFN, TESTFNDIR]
|
|
namelist = [TESTFN, TESTFNDIR + '/', TESTFNDIR + '/file.txt']
|
|
try:
|
|
out = self.zipfilecmd('-c', TESTFN2, *files)
|
|
self.assertEqual(out, b'')
|
|
with zipfile.ZipFile(TESTFN2) as zf:
|
|
self.assertEqual(zf.namelist(), namelist)
|
|
self.assertEqual(zf.read(namelist[0]), b'test 1')
|
|
self.assertEqual(zf.read(namelist[2]), b'test 2')
|
|
finally:
|
|
unlink(TESTFN2)
|
|
|
|
def test_extract_command(self):
|
|
zip_name = findfile('zipdir.zip')
|
|
extdir = TESTFNDIR
|
|
os.mkdir(extdir)
|
|
try:
|
|
out = self.zipfilecmd('-e', zip_name, extdir)
|
|
self.assertEqual(out, b'')
|
|
with zipfile.ZipFile(zip_name) as zf:
|
|
for zi in zf.infolist():
|
|
path = os.path.join(extdir,
|
|
zi.filename.replace('/', os.sep))
|
|
if zi.filename.endswith('/'):
|
|
self.assertTrue(os.path.isdir(path))
|
|
else:
|
|
self.assertTrue(os.path.isfile(path))
|
|
with open(path, 'rb') as f:
|
|
self.assertEqual(f.read(), zf.read(zi))
|
|
finally:
|
|
rmtree(extdir)
|
|
|
|
def test_main():
|
|
run_unittest(TestsWithSourceFile, TestZip64InSmallFiles, OtherTests,
|
|
PyZipFileTests, DecryptionTests, TestsWithMultipleOpens,
|
|
TestWithDirectory, UniversalNewlineTests,
|
|
TestsWithRandomBinaryFiles, CommandLineTest)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_main()
|