1 | n/a | import contextlib |
---|
2 | n/a | import io |
---|
3 | n/a | import os |
---|
4 | n/a | import importlib.util |
---|
5 | n/a | import posixpath |
---|
6 | n/a | import time |
---|
7 | n/a | import struct |
---|
8 | n/a | import zipfile |
---|
9 | n/a | import unittest |
---|
10 | n/a | |
---|
11 | n/a | |
---|
12 | n/a | from tempfile import TemporaryFile |
---|
13 | n/a | from random import randint, random, getrandbits |
---|
14 | n/a | |
---|
15 | n/a | from test.support import script_helper |
---|
16 | n/a | from test.support import (TESTFN, findfile, unlink, rmtree, temp_dir, |
---|
17 | n/a | requires_zlib, requires_bz2, requires_lzma, |
---|
18 | n/a | captured_stdout, check_warnings) |
---|
19 | n/a | |
---|
20 | n/a | TESTFN2 = TESTFN + "2" |
---|
21 | n/a | TESTFNDIR = TESTFN + "d" |
---|
22 | n/a | FIXEDTEST_SIZE = 1000 |
---|
23 | n/a | DATAFILES_DIR = 'zipfile_datafiles' |
---|
24 | n/a | |
---|
25 | n/a | SMALL_TEST_DATA = [('_ziptest1', '1q2w3e4r5t'), |
---|
26 | n/a | ('ziptest2dir/_ziptest2', 'qawsedrftg'), |
---|
27 | n/a | ('ziptest2dir/ziptest3dir/_ziptest3', 'azsxdcfvgb'), |
---|
28 | n/a | ('ziptest2dir/ziptest3dir/ziptest4dir/_ziptest3', '6y7u8i9o0p')] |
---|
29 | n/a | |
---|
30 | n/a | def getrandbytes(size): |
---|
31 | n/a | return getrandbits(8 * size).to_bytes(size, 'little') |
---|
32 | n/a | |
---|
33 | n/a | def get_files(test): |
---|
34 | n/a | yield TESTFN2 |
---|
35 | n/a | with TemporaryFile() as f: |
---|
36 | n/a | yield f |
---|
37 | n/a | test.assertFalse(f.closed) |
---|
38 | n/a | with io.BytesIO() as f: |
---|
39 | n/a | yield f |
---|
40 | n/a | test.assertFalse(f.closed) |
---|
41 | n/a | |
---|
42 | n/a | class AbstractTestsWithSourceFile: |
---|
43 | n/a | @classmethod |
---|
44 | n/a | def setUpClass(cls): |
---|
45 | n/a | cls.line_gen = [bytes("Zipfile test line %d. random float: %f\n" % |
---|
46 | n/a | (i, random()), "ascii") |
---|
47 | n/a | for i in range(FIXEDTEST_SIZE)] |
---|
48 | n/a | cls.data = b''.join(cls.line_gen) |
---|
49 | n/a | |
---|
50 | n/a | def setUp(self): |
---|
51 | n/a | # Make a source file with some lines |
---|
52 | n/a | with open(TESTFN, "wb") as fp: |
---|
53 | n/a | fp.write(self.data) |
---|
54 | n/a | |
---|
55 | n/a | def make_test_archive(self, f, compression): |
---|
56 | n/a | # Create the ZIP archive |
---|
57 | n/a | with zipfile.ZipFile(f, "w", compression) as zipfp: |
---|
58 | n/a | zipfp.write(TESTFN, "another.name") |
---|
59 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
60 | n/a | zipfp.writestr("strfile", self.data) |
---|
61 | n/a | with zipfp.open('written-open-w', mode='w') as f: |
---|
62 | n/a | for line in self.line_gen: |
---|
63 | n/a | f.write(line) |
---|
64 | n/a | |
---|
65 | n/a | def zip_test(self, f, compression): |
---|
66 | n/a | self.make_test_archive(f, compression) |
---|
67 | n/a | |
---|
68 | n/a | # Read the ZIP archive |
---|
69 | n/a | with zipfile.ZipFile(f, "r", compression) as zipfp: |
---|
70 | n/a | self.assertEqual(zipfp.read(TESTFN), self.data) |
---|
71 | n/a | self.assertEqual(zipfp.read("another.name"), self.data) |
---|
72 | n/a | self.assertEqual(zipfp.read("strfile"), self.data) |
---|
73 | n/a | |
---|
74 | n/a | # Print the ZIP directory |
---|
75 | n/a | fp = io.StringIO() |
---|
76 | n/a | zipfp.printdir(file=fp) |
---|
77 | n/a | directory = fp.getvalue() |
---|
78 | n/a | lines = directory.splitlines() |
---|
79 | n/a | self.assertEqual(len(lines), 5) # Number of files + header |
---|
80 | n/a | |
---|
81 | n/a | self.assertIn('File Name', lines[0]) |
---|
82 | n/a | self.assertIn('Modified', lines[0]) |
---|
83 | n/a | self.assertIn('Size', lines[0]) |
---|
84 | n/a | |
---|
85 | n/a | fn, date, time_, size = lines[1].split() |
---|
86 | n/a | self.assertEqual(fn, 'another.name') |
---|
87 | n/a | self.assertTrue(time.strptime(date, '%Y-%m-%d')) |
---|
88 | n/a | self.assertTrue(time.strptime(time_, '%H:%M:%S')) |
---|
89 | n/a | self.assertEqual(size, str(len(self.data))) |
---|
90 | n/a | |
---|
91 | n/a | # Check the namelist |
---|
92 | n/a | names = zipfp.namelist() |
---|
93 | n/a | self.assertEqual(len(names), 4) |
---|
94 | n/a | self.assertIn(TESTFN, names) |
---|
95 | n/a | self.assertIn("another.name", names) |
---|
96 | n/a | self.assertIn("strfile", names) |
---|
97 | n/a | self.assertIn("written-open-w", names) |
---|
98 | n/a | |
---|
99 | n/a | # Check infolist |
---|
100 | n/a | infos = zipfp.infolist() |
---|
101 | n/a | names = [i.filename for i in infos] |
---|
102 | n/a | self.assertEqual(len(names), 4) |
---|
103 | n/a | self.assertIn(TESTFN, names) |
---|
104 | n/a | self.assertIn("another.name", names) |
---|
105 | n/a | self.assertIn("strfile", names) |
---|
106 | n/a | self.assertIn("written-open-w", names) |
---|
107 | n/a | for i in infos: |
---|
108 | n/a | self.assertEqual(i.file_size, len(self.data)) |
---|
109 | n/a | |
---|
110 | n/a | # check getinfo |
---|
111 | n/a | for nm in (TESTFN, "another.name", "strfile", "written-open-w"): |
---|
112 | n/a | info = zipfp.getinfo(nm) |
---|
113 | n/a | self.assertEqual(info.filename, nm) |
---|
114 | n/a | self.assertEqual(info.file_size, len(self.data)) |
---|
115 | n/a | |
---|
116 | n/a | # Check that testzip doesn't raise an exception |
---|
117 | n/a | zipfp.testzip() |
---|
118 | n/a | |
---|
119 | n/a | def test_basic(self): |
---|
120 | n/a | for f in get_files(self): |
---|
121 | n/a | self.zip_test(f, self.compression) |
---|
122 | n/a | |
---|
123 | n/a | def zip_open_test(self, f, compression): |
---|
124 | n/a | self.make_test_archive(f, compression) |
---|
125 | n/a | |
---|
126 | n/a | # Read the ZIP archive |
---|
127 | n/a | with zipfile.ZipFile(f, "r", compression) as zipfp: |
---|
128 | n/a | zipdata1 = [] |
---|
129 | n/a | with zipfp.open(TESTFN) as zipopen1: |
---|
130 | n/a | while True: |
---|
131 | n/a | read_data = zipopen1.read(256) |
---|
132 | n/a | if not read_data: |
---|
133 | n/a | break |
---|
134 | n/a | zipdata1.append(read_data) |
---|
135 | n/a | |
---|
136 | n/a | zipdata2 = [] |
---|
137 | n/a | with zipfp.open("another.name") as zipopen2: |
---|
138 | n/a | while True: |
---|
139 | n/a | read_data = zipopen2.read(256) |
---|
140 | n/a | if not read_data: |
---|
141 | n/a | break |
---|
142 | n/a | zipdata2.append(read_data) |
---|
143 | n/a | |
---|
144 | n/a | self.assertEqual(b''.join(zipdata1), self.data) |
---|
145 | n/a | self.assertEqual(b''.join(zipdata2), self.data) |
---|
146 | n/a | |
---|
147 | n/a | def test_open(self): |
---|
148 | n/a | for f in get_files(self): |
---|
149 | n/a | self.zip_open_test(f, self.compression) |
---|
150 | n/a | |
---|
151 | n/a | def zip_random_open_test(self, f, compression): |
---|
152 | n/a | self.make_test_archive(f, compression) |
---|
153 | n/a | |
---|
154 | n/a | # Read the ZIP archive |
---|
155 | n/a | with zipfile.ZipFile(f, "r", compression) as zipfp: |
---|
156 | n/a | zipdata1 = [] |
---|
157 | n/a | with zipfp.open(TESTFN) as zipopen1: |
---|
158 | n/a | while True: |
---|
159 | n/a | read_data = zipopen1.read(randint(1, 1024)) |
---|
160 | n/a | if not read_data: |
---|
161 | n/a | break |
---|
162 | n/a | zipdata1.append(read_data) |
---|
163 | n/a | |
---|
164 | n/a | self.assertEqual(b''.join(zipdata1), self.data) |
---|
165 | n/a | |
---|
166 | n/a | def test_random_open(self): |
---|
167 | n/a | for f in get_files(self): |
---|
168 | n/a | self.zip_random_open_test(f, self.compression) |
---|
169 | n/a | |
---|
170 | n/a | def zip_read1_test(self, f, compression): |
---|
171 | n/a | self.make_test_archive(f, compression) |
---|
172 | n/a | |
---|
173 | n/a | # Read the ZIP archive |
---|
174 | n/a | with zipfile.ZipFile(f, "r") as zipfp, \ |
---|
175 | n/a | zipfp.open(TESTFN) as zipopen: |
---|
176 | n/a | zipdata = [] |
---|
177 | n/a | while True: |
---|
178 | n/a | read_data = zipopen.read1(-1) |
---|
179 | n/a | if not read_data: |
---|
180 | n/a | break |
---|
181 | n/a | zipdata.append(read_data) |
---|
182 | n/a | |
---|
183 | n/a | self.assertEqual(b''.join(zipdata), self.data) |
---|
184 | n/a | |
---|
185 | n/a | def test_read1(self): |
---|
186 | n/a | for f in get_files(self): |
---|
187 | n/a | self.zip_read1_test(f, self.compression) |
---|
188 | n/a | |
---|
189 | n/a | def zip_read1_10_test(self, f, compression): |
---|
190 | n/a | self.make_test_archive(f, compression) |
---|
191 | n/a | |
---|
192 | n/a | # Read the ZIP archive |
---|
193 | n/a | with zipfile.ZipFile(f, "r") as zipfp, \ |
---|
194 | n/a | zipfp.open(TESTFN) as zipopen: |
---|
195 | n/a | zipdata = [] |
---|
196 | n/a | while True: |
---|
197 | n/a | read_data = zipopen.read1(10) |
---|
198 | n/a | self.assertLessEqual(len(read_data), 10) |
---|
199 | n/a | if not read_data: |
---|
200 | n/a | break |
---|
201 | n/a | zipdata.append(read_data) |
---|
202 | n/a | |
---|
203 | n/a | self.assertEqual(b''.join(zipdata), self.data) |
---|
204 | n/a | |
---|
205 | n/a | def test_read1_10(self): |
---|
206 | n/a | for f in get_files(self): |
---|
207 | n/a | self.zip_read1_10_test(f, self.compression) |
---|
208 | n/a | |
---|
209 | n/a | def zip_readline_read_test(self, f, compression): |
---|
210 | n/a | self.make_test_archive(f, compression) |
---|
211 | n/a | |
---|
212 | n/a | # Read the ZIP archive |
---|
213 | n/a | with zipfile.ZipFile(f, "r") as zipfp, \ |
---|
214 | n/a | zipfp.open(TESTFN) as zipopen: |
---|
215 | n/a | data = b'' |
---|
216 | n/a | while True: |
---|
217 | n/a | read = zipopen.readline() |
---|
218 | n/a | if not read: |
---|
219 | n/a | break |
---|
220 | n/a | data += read |
---|
221 | n/a | |
---|
222 | n/a | read = zipopen.read(100) |
---|
223 | n/a | if not read: |
---|
224 | n/a | break |
---|
225 | n/a | data += read |
---|
226 | n/a | |
---|
227 | n/a | self.assertEqual(data, self.data) |
---|
228 | n/a | |
---|
229 | n/a | def test_readline_read(self): |
---|
230 | n/a | # Issue #7610: calls to readline() interleaved with calls to read(). |
---|
231 | n/a | for f in get_files(self): |
---|
232 | n/a | self.zip_readline_read_test(f, self.compression) |
---|
233 | n/a | |
---|
234 | n/a | def zip_readline_test(self, f, compression): |
---|
235 | n/a | self.make_test_archive(f, compression) |
---|
236 | n/a | |
---|
237 | n/a | # Read the ZIP archive |
---|
238 | n/a | with zipfile.ZipFile(f, "r") as zipfp: |
---|
239 | n/a | with zipfp.open(TESTFN) as zipopen: |
---|
240 | n/a | for line in self.line_gen: |
---|
241 | n/a | linedata = zipopen.readline() |
---|
242 | n/a | self.assertEqual(linedata, line) |
---|
243 | n/a | |
---|
244 | n/a | def test_readline(self): |
---|
245 | n/a | for f in get_files(self): |
---|
246 | n/a | self.zip_readline_test(f, self.compression) |
---|
247 | n/a | |
---|
248 | n/a | def zip_readlines_test(self, f, compression): |
---|
249 | n/a | self.make_test_archive(f, compression) |
---|
250 | n/a | |
---|
251 | n/a | # Read the ZIP archive |
---|
252 | n/a | with zipfile.ZipFile(f, "r") as zipfp: |
---|
253 | n/a | with zipfp.open(TESTFN) as zipopen: |
---|
254 | n/a | ziplines = zipopen.readlines() |
---|
255 | n/a | for line, zipline in zip(self.line_gen, ziplines): |
---|
256 | n/a | self.assertEqual(zipline, line) |
---|
257 | n/a | |
---|
258 | n/a | def test_readlines(self): |
---|
259 | n/a | for f in get_files(self): |
---|
260 | n/a | self.zip_readlines_test(f, self.compression) |
---|
261 | n/a | |
---|
262 | n/a | def zip_iterlines_test(self, f, compression): |
---|
263 | n/a | self.make_test_archive(f, compression) |
---|
264 | n/a | |
---|
265 | n/a | # Read the ZIP archive |
---|
266 | n/a | with zipfile.ZipFile(f, "r") as zipfp: |
---|
267 | n/a | with zipfp.open(TESTFN) as zipopen: |
---|
268 | n/a | for line, zipline in zip(self.line_gen, zipopen): |
---|
269 | n/a | self.assertEqual(zipline, line) |
---|
270 | n/a | |
---|
271 | n/a | def test_iterlines(self): |
---|
272 | n/a | for f in get_files(self): |
---|
273 | n/a | self.zip_iterlines_test(f, self.compression) |
---|
274 | n/a | |
---|
275 | n/a | def test_low_compression(self): |
---|
276 | n/a | """Check for cases where compressed data is larger than original.""" |
---|
277 | n/a | # Create the ZIP archive |
---|
278 | n/a | with zipfile.ZipFile(TESTFN2, "w", self.compression) as zipfp: |
---|
279 | n/a | zipfp.writestr("strfile", '12') |
---|
280 | n/a | |
---|
281 | n/a | # Get an open object for strfile |
---|
282 | n/a | with zipfile.ZipFile(TESTFN2, "r", self.compression) as zipfp: |
---|
283 | n/a | with zipfp.open("strfile") as openobj: |
---|
284 | n/a | self.assertEqual(openobj.read(1), b'1') |
---|
285 | n/a | self.assertEqual(openobj.read(1), b'2') |
---|
286 | n/a | |
---|
287 | n/a | def test_writestr_compression(self): |
---|
288 | n/a | zipfp = zipfile.ZipFile(TESTFN2, "w") |
---|
289 | n/a | zipfp.writestr("b.txt", "hello world", compress_type=self.compression) |
---|
290 | n/a | info = zipfp.getinfo('b.txt') |
---|
291 | n/a | self.assertEqual(info.compress_type, self.compression) |
---|
292 | n/a | |
---|
293 | n/a | def test_read_return_size(self): |
---|
294 | n/a | # Issue #9837: ZipExtFile.read() shouldn't return more bytes |
---|
295 | n/a | # than requested. |
---|
296 | n/a | for test_size in (1, 4095, 4096, 4097, 16384): |
---|
297 | n/a | file_size = test_size + 1 |
---|
298 | n/a | junk = getrandbytes(file_size) |
---|
299 | n/a | with zipfile.ZipFile(io.BytesIO(), "w", self.compression) as zipf: |
---|
300 | n/a | zipf.writestr('foo', junk) |
---|
301 | n/a | with zipf.open('foo', 'r') as fp: |
---|
302 | n/a | buf = fp.read(test_size) |
---|
303 | n/a | self.assertEqual(len(buf), test_size) |
---|
304 | n/a | |
---|
305 | n/a | def test_truncated_zipfile(self): |
---|
306 | n/a | fp = io.BytesIO() |
---|
307 | n/a | with zipfile.ZipFile(fp, mode='w') as zipf: |
---|
308 | n/a | zipf.writestr('strfile', self.data, compress_type=self.compression) |
---|
309 | n/a | end_offset = fp.tell() |
---|
310 | n/a | zipfiledata = fp.getvalue() |
---|
311 | n/a | |
---|
312 | n/a | fp = io.BytesIO(zipfiledata) |
---|
313 | n/a | with zipfile.ZipFile(fp) as zipf: |
---|
314 | n/a | with zipf.open('strfile') as zipopen: |
---|
315 | n/a | fp.truncate(end_offset - 20) |
---|
316 | n/a | with self.assertRaises(EOFError): |
---|
317 | n/a | zipopen.read() |
---|
318 | n/a | |
---|
319 | n/a | fp = io.BytesIO(zipfiledata) |
---|
320 | n/a | with zipfile.ZipFile(fp) as zipf: |
---|
321 | n/a | with zipf.open('strfile') as zipopen: |
---|
322 | n/a | fp.truncate(end_offset - 20) |
---|
323 | n/a | with self.assertRaises(EOFError): |
---|
324 | n/a | while zipopen.read(100): |
---|
325 | n/a | pass |
---|
326 | n/a | |
---|
327 | n/a | fp = io.BytesIO(zipfiledata) |
---|
328 | n/a | with zipfile.ZipFile(fp) as zipf: |
---|
329 | n/a | with zipf.open('strfile') as zipopen: |
---|
330 | n/a | fp.truncate(end_offset - 20) |
---|
331 | n/a | with self.assertRaises(EOFError): |
---|
332 | n/a | while zipopen.read1(100): |
---|
333 | n/a | pass |
---|
334 | n/a | |
---|
335 | n/a | def test_repr(self): |
---|
336 | n/a | fname = 'file.name' |
---|
337 | n/a | for f in get_files(self): |
---|
338 | n/a | with zipfile.ZipFile(f, 'w', self.compression) as zipfp: |
---|
339 | n/a | zipfp.write(TESTFN, fname) |
---|
340 | n/a | r = repr(zipfp) |
---|
341 | n/a | self.assertIn("mode='w'", r) |
---|
342 | n/a | |
---|
343 | n/a | with zipfile.ZipFile(f, 'r') as zipfp: |
---|
344 | n/a | r = repr(zipfp) |
---|
345 | n/a | if isinstance(f, str): |
---|
346 | n/a | self.assertIn('filename=%r' % f, r) |
---|
347 | n/a | else: |
---|
348 | n/a | self.assertIn('file=%r' % f, r) |
---|
349 | n/a | self.assertIn("mode='r'", r) |
---|
350 | n/a | r = repr(zipfp.getinfo(fname)) |
---|
351 | n/a | self.assertIn('filename=%r' % fname, r) |
---|
352 | n/a | self.assertIn('filemode=', r) |
---|
353 | n/a | self.assertIn('file_size=', r) |
---|
354 | n/a | if self.compression != zipfile.ZIP_STORED: |
---|
355 | n/a | self.assertIn('compress_type=', r) |
---|
356 | n/a | self.assertIn('compress_size=', r) |
---|
357 | n/a | with zipfp.open(fname) as zipopen: |
---|
358 | n/a | r = repr(zipopen) |
---|
359 | n/a | self.assertIn('name=%r' % fname, r) |
---|
360 | n/a | self.assertIn("mode='r'", r) |
---|
361 | n/a | if self.compression != zipfile.ZIP_STORED: |
---|
362 | n/a | self.assertIn('compress_type=', r) |
---|
363 | n/a | self.assertIn('[closed]', repr(zipopen)) |
---|
364 | n/a | self.assertIn('[closed]', repr(zipfp)) |
---|
365 | n/a | |
---|
366 | n/a | def tearDown(self): |
---|
367 | n/a | unlink(TESTFN) |
---|
368 | n/a | unlink(TESTFN2) |
---|
369 | n/a | |
---|
370 | n/a | |
---|
371 | n/a | class StoredTestsWithSourceFile(AbstractTestsWithSourceFile, |
---|
372 | n/a | unittest.TestCase): |
---|
373 | n/a | compression = zipfile.ZIP_STORED |
---|
374 | n/a | test_low_compression = None |
---|
375 | n/a | |
---|
376 | n/a | def zip_test_writestr_permissions(self, f, compression): |
---|
377 | n/a | # Make sure that writestr and open(... mode='w') create files with |
---|
378 | n/a | # mode 0600, when they are passed a name rather than a ZipInfo |
---|
379 | n/a | # instance. |
---|
380 | n/a | |
---|
381 | n/a | self.make_test_archive(f, compression) |
---|
382 | n/a | with zipfile.ZipFile(f, "r") as zipfp: |
---|
383 | n/a | zinfo = zipfp.getinfo('strfile') |
---|
384 | n/a | self.assertEqual(zinfo.external_attr, 0o600 << 16) |
---|
385 | n/a | |
---|
386 | n/a | zinfo2 = zipfp.getinfo('written-open-w') |
---|
387 | n/a | self.assertEqual(zinfo2.external_attr, 0o600 << 16) |
---|
388 | n/a | |
---|
389 | n/a | def test_writestr_permissions(self): |
---|
390 | n/a | for f in get_files(self): |
---|
391 | n/a | self.zip_test_writestr_permissions(f, zipfile.ZIP_STORED) |
---|
392 | n/a | |
---|
393 | n/a | def test_absolute_arcnames(self): |
---|
394 | n/a | with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: |
---|
395 | n/a | zipfp.write(TESTFN, "/absolute") |
---|
396 | n/a | |
---|
397 | n/a | with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: |
---|
398 | n/a | self.assertEqual(zipfp.namelist(), ["absolute"]) |
---|
399 | n/a | |
---|
400 | n/a | def test_append_to_zip_file(self): |
---|
401 | n/a | """Test appending to an existing zipfile.""" |
---|
402 | n/a | with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: |
---|
403 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
404 | n/a | |
---|
405 | n/a | with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp: |
---|
406 | n/a | zipfp.writestr("strfile", self.data) |
---|
407 | n/a | self.assertEqual(zipfp.namelist(), [TESTFN, "strfile"]) |
---|
408 | n/a | |
---|
409 | n/a | def test_append_to_non_zip_file(self): |
---|
410 | n/a | """Test appending to an existing file that is not a zipfile.""" |
---|
411 | n/a | # NOTE: this test fails if len(d) < 22 because of the first |
---|
412 | n/a | # line "fpin.seek(-22, 2)" in _EndRecData |
---|
413 | n/a | data = b'I am not a ZipFile!'*10 |
---|
414 | n/a | with open(TESTFN2, 'wb') as f: |
---|
415 | n/a | f.write(data) |
---|
416 | n/a | |
---|
417 | n/a | with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp: |
---|
418 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
419 | n/a | |
---|
420 | n/a | with open(TESTFN2, 'rb') as f: |
---|
421 | n/a | f.seek(len(data)) |
---|
422 | n/a | with zipfile.ZipFile(f, "r") as zipfp: |
---|
423 | n/a | self.assertEqual(zipfp.namelist(), [TESTFN]) |
---|
424 | n/a | self.assertEqual(zipfp.read(TESTFN), self.data) |
---|
425 | n/a | with open(TESTFN2, 'rb') as f: |
---|
426 | n/a | self.assertEqual(f.read(len(data)), data) |
---|
427 | n/a | zipfiledata = f.read() |
---|
428 | n/a | with io.BytesIO(zipfiledata) as bio, zipfile.ZipFile(bio) as zipfp: |
---|
429 | n/a | self.assertEqual(zipfp.namelist(), [TESTFN]) |
---|
430 | n/a | self.assertEqual(zipfp.read(TESTFN), self.data) |
---|
431 | n/a | |
---|
432 | n/a | def test_read_concatenated_zip_file(self): |
---|
433 | n/a | with io.BytesIO() as bio: |
---|
434 | n/a | with zipfile.ZipFile(bio, 'w', zipfile.ZIP_STORED) as zipfp: |
---|
435 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
436 | n/a | zipfiledata = bio.getvalue() |
---|
437 | n/a | data = b'I am not a ZipFile!'*10 |
---|
438 | n/a | with open(TESTFN2, 'wb') as f: |
---|
439 | n/a | f.write(data) |
---|
440 | n/a | f.write(zipfiledata) |
---|
441 | n/a | |
---|
442 | n/a | with zipfile.ZipFile(TESTFN2) as zipfp: |
---|
443 | n/a | self.assertEqual(zipfp.namelist(), [TESTFN]) |
---|
444 | n/a | self.assertEqual(zipfp.read(TESTFN), self.data) |
---|
445 | n/a | |
---|
446 | n/a | def test_append_to_concatenated_zip_file(self): |
---|
447 | n/a | with io.BytesIO() as bio: |
---|
448 | n/a | with zipfile.ZipFile(bio, 'w', zipfile.ZIP_STORED) as zipfp: |
---|
449 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
450 | n/a | zipfiledata = bio.getvalue() |
---|
451 | n/a | data = b'I am not a ZipFile!'*1000000 |
---|
452 | n/a | with open(TESTFN2, 'wb') as f: |
---|
453 | n/a | f.write(data) |
---|
454 | n/a | f.write(zipfiledata) |
---|
455 | n/a | |
---|
456 | n/a | with zipfile.ZipFile(TESTFN2, 'a') as zipfp: |
---|
457 | n/a | self.assertEqual(zipfp.namelist(), [TESTFN]) |
---|
458 | n/a | zipfp.writestr('strfile', self.data) |
---|
459 | n/a | |
---|
460 | n/a | with open(TESTFN2, 'rb') as f: |
---|
461 | n/a | self.assertEqual(f.read(len(data)), data) |
---|
462 | n/a | zipfiledata = f.read() |
---|
463 | n/a | with io.BytesIO(zipfiledata) as bio, zipfile.ZipFile(bio) as zipfp: |
---|
464 | n/a | self.assertEqual(zipfp.namelist(), [TESTFN, 'strfile']) |
---|
465 | n/a | self.assertEqual(zipfp.read(TESTFN), self.data) |
---|
466 | n/a | self.assertEqual(zipfp.read('strfile'), self.data) |
---|
467 | n/a | |
---|
468 | n/a | def test_ignores_newline_at_end(self): |
---|
469 | n/a | with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: |
---|
470 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
471 | n/a | with open(TESTFN2, 'a') as f: |
---|
472 | n/a | f.write("\r\n\00\00\00") |
---|
473 | n/a | with zipfile.ZipFile(TESTFN2, "r") as zipfp: |
---|
474 | n/a | self.assertIsInstance(zipfp, zipfile.ZipFile) |
---|
475 | n/a | |
---|
476 | n/a | def test_ignores_stuff_appended_past_comments(self): |
---|
477 | n/a | with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: |
---|
478 | n/a | zipfp.comment = b"this is a comment" |
---|
479 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
480 | n/a | with open(TESTFN2, 'a') as f: |
---|
481 | n/a | f.write("abcdef\r\n") |
---|
482 | n/a | with zipfile.ZipFile(TESTFN2, "r") as zipfp: |
---|
483 | n/a | self.assertIsInstance(zipfp, zipfile.ZipFile) |
---|
484 | n/a | self.assertEqual(zipfp.comment, b"this is a comment") |
---|
485 | n/a | |
---|
486 | n/a | def test_write_default_name(self): |
---|
487 | n/a | """Check that calling ZipFile.write without arcname specified |
---|
488 | n/a | produces the expected result.""" |
---|
489 | n/a | with zipfile.ZipFile(TESTFN2, "w") as zipfp: |
---|
490 | n/a | zipfp.write(TESTFN) |
---|
491 | n/a | with open(TESTFN, "rb") as f: |
---|
492 | n/a | self.assertEqual(zipfp.read(TESTFN), f.read()) |
---|
493 | n/a | |
---|
494 | n/a | def test_write_to_readonly(self): |
---|
495 | n/a | """Check that trying to call write() on a readonly ZipFile object |
---|
496 | n/a | raises a ValueError.""" |
---|
497 | n/a | with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: |
---|
498 | n/a | zipfp.writestr("somefile.txt", "bogus") |
---|
499 | n/a | |
---|
500 | n/a | with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: |
---|
501 | n/a | self.assertRaises(ValueError, zipfp.write, TESTFN) |
---|
502 | n/a | |
---|
503 | n/a | with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: |
---|
504 | n/a | with self.assertRaises(ValueError): |
---|
505 | n/a | zipfp.open(TESTFN, mode='w') |
---|
506 | n/a | |
---|
507 | n/a | def test_add_file_before_1980(self): |
---|
508 | n/a | # Set atime and mtime to 1970-01-01 |
---|
509 | n/a | os.utime(TESTFN, (0, 0)) |
---|
510 | n/a | with zipfile.ZipFile(TESTFN2, "w") as zipfp: |
---|
511 | n/a | self.assertRaises(ValueError, zipfp.write, TESTFN) |
---|
512 | n/a | |
---|
513 | n/a | |
---|
514 | n/a | @requires_zlib |
---|
515 | n/a | class DeflateTestsWithSourceFile(AbstractTestsWithSourceFile, |
---|
516 | n/a | unittest.TestCase): |
---|
517 | n/a | compression = zipfile.ZIP_DEFLATED |
---|
518 | n/a | |
---|
519 | n/a | def test_per_file_compression(self): |
---|
520 | n/a | """Check that files within a Zip archive can have different |
---|
521 | n/a | compression options.""" |
---|
522 | n/a | with zipfile.ZipFile(TESTFN2, "w") as zipfp: |
---|
523 | n/a | zipfp.write(TESTFN, 'storeme', zipfile.ZIP_STORED) |
---|
524 | n/a | zipfp.write(TESTFN, 'deflateme', zipfile.ZIP_DEFLATED) |
---|
525 | n/a | sinfo = zipfp.getinfo('storeme') |
---|
526 | n/a | dinfo = zipfp.getinfo('deflateme') |
---|
527 | n/a | self.assertEqual(sinfo.compress_type, zipfile.ZIP_STORED) |
---|
528 | n/a | self.assertEqual(dinfo.compress_type, zipfile.ZIP_DEFLATED) |
---|
529 | n/a | |
---|
530 | n/a | @requires_bz2 |
---|
531 | n/a | class Bzip2TestsWithSourceFile(AbstractTestsWithSourceFile, |
---|
532 | n/a | unittest.TestCase): |
---|
533 | n/a | compression = zipfile.ZIP_BZIP2 |
---|
534 | n/a | |
---|
535 | n/a | @requires_lzma |
---|
536 | n/a | class LzmaTestsWithSourceFile(AbstractTestsWithSourceFile, |
---|
537 | n/a | unittest.TestCase): |
---|
538 | n/a | compression = zipfile.ZIP_LZMA |
---|
539 | n/a | |
---|
540 | n/a | |
---|
541 | n/a | class AbstractTestZip64InSmallFiles: |
---|
542 | n/a | # These tests test the ZIP64 functionality without using large files, |
---|
543 | n/a | # see test_zipfile64 for proper tests. |
---|
544 | n/a | |
---|
545 | n/a | @classmethod |
---|
546 | n/a | def setUpClass(cls): |
---|
547 | n/a | line_gen = (bytes("Test of zipfile line %d." % i, "ascii") |
---|
548 | n/a | for i in range(0, FIXEDTEST_SIZE)) |
---|
549 | n/a | cls.data = b'\n'.join(line_gen) |
---|
550 | n/a | |
---|
551 | n/a | def setUp(self): |
---|
552 | n/a | self._limit = zipfile.ZIP64_LIMIT |
---|
553 | n/a | self._filecount_limit = zipfile.ZIP_FILECOUNT_LIMIT |
---|
554 | n/a | zipfile.ZIP64_LIMIT = 1000 |
---|
555 | n/a | zipfile.ZIP_FILECOUNT_LIMIT = 9 |
---|
556 | n/a | |
---|
557 | n/a | # Make a source file with some lines |
---|
558 | n/a | with open(TESTFN, "wb") as fp: |
---|
559 | n/a | fp.write(self.data) |
---|
560 | n/a | |
---|
561 | n/a | def zip_test(self, f, compression): |
---|
562 | n/a | # Create the ZIP archive |
---|
563 | n/a | with zipfile.ZipFile(f, "w", compression, allowZip64=True) as zipfp: |
---|
564 | n/a | zipfp.write(TESTFN, "another.name") |
---|
565 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
566 | n/a | zipfp.writestr("strfile", self.data) |
---|
567 | n/a | |
---|
568 | n/a | # Read the ZIP archive |
---|
569 | n/a | with zipfile.ZipFile(f, "r", compression) as zipfp: |
---|
570 | n/a | self.assertEqual(zipfp.read(TESTFN), self.data) |
---|
571 | n/a | self.assertEqual(zipfp.read("another.name"), self.data) |
---|
572 | n/a | self.assertEqual(zipfp.read("strfile"), self.data) |
---|
573 | n/a | |
---|
574 | n/a | # Print the ZIP directory |
---|
575 | n/a | fp = io.StringIO() |
---|
576 | n/a | zipfp.printdir(fp) |
---|
577 | n/a | |
---|
578 | n/a | directory = fp.getvalue() |
---|
579 | n/a | lines = directory.splitlines() |
---|
580 | n/a | self.assertEqual(len(lines), 4) # Number of files + header |
---|
581 | n/a | |
---|
582 | n/a | self.assertIn('File Name', lines[0]) |
---|
583 | n/a | self.assertIn('Modified', lines[0]) |
---|
584 | n/a | self.assertIn('Size', lines[0]) |
---|
585 | n/a | |
---|
586 | n/a | fn, date, time_, size = lines[1].split() |
---|
587 | n/a | self.assertEqual(fn, 'another.name') |
---|
588 | n/a | self.assertTrue(time.strptime(date, '%Y-%m-%d')) |
---|
589 | n/a | self.assertTrue(time.strptime(time_, '%H:%M:%S')) |
---|
590 | n/a | self.assertEqual(size, str(len(self.data))) |
---|
591 | n/a | |
---|
592 | n/a | # Check the namelist |
---|
593 | n/a | names = zipfp.namelist() |
---|
594 | n/a | self.assertEqual(len(names), 3) |
---|
595 | n/a | self.assertIn(TESTFN, names) |
---|
596 | n/a | self.assertIn("another.name", names) |
---|
597 | n/a | self.assertIn("strfile", names) |
---|
598 | n/a | |
---|
599 | n/a | # Check infolist |
---|
600 | n/a | infos = zipfp.infolist() |
---|
601 | n/a | names = [i.filename for i in infos] |
---|
602 | n/a | self.assertEqual(len(names), 3) |
---|
603 | n/a | self.assertIn(TESTFN, names) |
---|
604 | n/a | self.assertIn("another.name", names) |
---|
605 | n/a | self.assertIn("strfile", names) |
---|
606 | n/a | for i in infos: |
---|
607 | n/a | self.assertEqual(i.file_size, len(self.data)) |
---|
608 | n/a | |
---|
609 | n/a | # check getinfo |
---|
610 | n/a | for nm in (TESTFN, "another.name", "strfile"): |
---|
611 | n/a | info = zipfp.getinfo(nm) |
---|
612 | n/a | self.assertEqual(info.filename, nm) |
---|
613 | n/a | self.assertEqual(info.file_size, len(self.data)) |
---|
614 | n/a | |
---|
615 | n/a | # Check that testzip doesn't raise an exception |
---|
616 | n/a | zipfp.testzip() |
---|
617 | n/a | |
---|
618 | n/a | def test_basic(self): |
---|
619 | n/a | for f in get_files(self): |
---|
620 | n/a | self.zip_test(f, self.compression) |
---|
621 | n/a | |
---|
622 | n/a | def test_too_many_files(self): |
---|
623 | n/a | # This test checks that more than 64k files can be added to an archive, |
---|
624 | n/a | # and that the resulting archive can be read properly by ZipFile |
---|
625 | n/a | zipf = zipfile.ZipFile(TESTFN, "w", self.compression, |
---|
626 | n/a | allowZip64=True) |
---|
627 | n/a | zipf.debug = 100 |
---|
628 | n/a | numfiles = 15 |
---|
629 | n/a | for i in range(numfiles): |
---|
630 | n/a | zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) |
---|
631 | n/a | self.assertEqual(len(zipf.namelist()), numfiles) |
---|
632 | n/a | zipf.close() |
---|
633 | n/a | |
---|
634 | n/a | zipf2 = zipfile.ZipFile(TESTFN, "r", self.compression) |
---|
635 | n/a | self.assertEqual(len(zipf2.namelist()), numfiles) |
---|
636 | n/a | for i in range(numfiles): |
---|
637 | n/a | content = zipf2.read("foo%08d" % i).decode('ascii') |
---|
638 | n/a | self.assertEqual(content, "%d" % (i**3 % 57)) |
---|
639 | n/a | zipf2.close() |
---|
640 | n/a | |
---|
641 | n/a | def test_too_many_files_append(self): |
---|
642 | n/a | zipf = zipfile.ZipFile(TESTFN, "w", self.compression, |
---|
643 | n/a | allowZip64=False) |
---|
644 | n/a | zipf.debug = 100 |
---|
645 | n/a | numfiles = 9 |
---|
646 | n/a | for i in range(numfiles): |
---|
647 | n/a | zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) |
---|
648 | n/a | self.assertEqual(len(zipf.namelist()), numfiles) |
---|
649 | n/a | with self.assertRaises(zipfile.LargeZipFile): |
---|
650 | n/a | zipf.writestr("foo%08d" % numfiles, b'') |
---|
651 | n/a | self.assertEqual(len(zipf.namelist()), numfiles) |
---|
652 | n/a | zipf.close() |
---|
653 | n/a | |
---|
654 | n/a | zipf = zipfile.ZipFile(TESTFN, "a", self.compression, |
---|
655 | n/a | allowZip64=False) |
---|
656 | n/a | zipf.debug = 100 |
---|
657 | n/a | self.assertEqual(len(zipf.namelist()), numfiles) |
---|
658 | n/a | with self.assertRaises(zipfile.LargeZipFile): |
---|
659 | n/a | zipf.writestr("foo%08d" % numfiles, b'') |
---|
660 | n/a | self.assertEqual(len(zipf.namelist()), numfiles) |
---|
661 | n/a | zipf.close() |
---|
662 | n/a | |
---|
663 | n/a | zipf = zipfile.ZipFile(TESTFN, "a", self.compression, |
---|
664 | n/a | allowZip64=True) |
---|
665 | n/a | zipf.debug = 100 |
---|
666 | n/a | self.assertEqual(len(zipf.namelist()), numfiles) |
---|
667 | n/a | numfiles2 = 15 |
---|
668 | n/a | for i in range(numfiles, numfiles2): |
---|
669 | n/a | zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) |
---|
670 | n/a | self.assertEqual(len(zipf.namelist()), numfiles2) |
---|
671 | n/a | zipf.close() |
---|
672 | n/a | |
---|
673 | n/a | zipf2 = zipfile.ZipFile(TESTFN, "r", self.compression) |
---|
674 | n/a | self.assertEqual(len(zipf2.namelist()), numfiles2) |
---|
675 | n/a | for i in range(numfiles2): |
---|
676 | n/a | content = zipf2.read("foo%08d" % i).decode('ascii') |
---|
677 | n/a | self.assertEqual(content, "%d" % (i**3 % 57)) |
---|
678 | n/a | zipf2.close() |
---|
679 | n/a | |
---|
680 | n/a | def tearDown(self): |
---|
681 | n/a | zipfile.ZIP64_LIMIT = self._limit |
---|
682 | n/a | zipfile.ZIP_FILECOUNT_LIMIT = self._filecount_limit |
---|
683 | n/a | unlink(TESTFN) |
---|
684 | n/a | unlink(TESTFN2) |
---|
685 | n/a | |
---|
686 | n/a | |
---|
687 | n/a | class StoredTestZip64InSmallFiles(AbstractTestZip64InSmallFiles, |
---|
688 | n/a | unittest.TestCase): |
---|
689 | n/a | compression = zipfile.ZIP_STORED |
---|
690 | n/a | |
---|
691 | n/a | def large_file_exception_test(self, f, compression): |
---|
692 | n/a | with zipfile.ZipFile(f, "w", compression, allowZip64=False) as zipfp: |
---|
693 | n/a | self.assertRaises(zipfile.LargeZipFile, |
---|
694 | n/a | zipfp.write, TESTFN, "another.name") |
---|
695 | n/a | |
---|
696 | n/a | def large_file_exception_test2(self, f, compression): |
---|
697 | n/a | with zipfile.ZipFile(f, "w", compression, allowZip64=False) as zipfp: |
---|
698 | n/a | self.assertRaises(zipfile.LargeZipFile, |
---|
699 | n/a | zipfp.writestr, "another.name", self.data) |
---|
700 | n/a | |
---|
701 | n/a | def test_large_file_exception(self): |
---|
702 | n/a | for f in get_files(self): |
---|
703 | n/a | self.large_file_exception_test(f, zipfile.ZIP_STORED) |
---|
704 | n/a | self.large_file_exception_test2(f, zipfile.ZIP_STORED) |
---|
705 | n/a | |
---|
706 | n/a | def test_absolute_arcnames(self): |
---|
707 | n/a | with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED, |
---|
708 | n/a | allowZip64=True) as zipfp: |
---|
709 | n/a | zipfp.write(TESTFN, "/absolute") |
---|
710 | n/a | |
---|
711 | n/a | with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: |
---|
712 | n/a | self.assertEqual(zipfp.namelist(), ["absolute"]) |
---|
713 | n/a | |
---|
714 | n/a | @requires_zlib |
---|
715 | n/a | class DeflateTestZip64InSmallFiles(AbstractTestZip64InSmallFiles, |
---|
716 | n/a | unittest.TestCase): |
---|
717 | n/a | compression = zipfile.ZIP_DEFLATED |
---|
718 | n/a | |
---|
719 | n/a | @requires_bz2 |
---|
720 | n/a | class Bzip2TestZip64InSmallFiles(AbstractTestZip64InSmallFiles, |
---|
721 | n/a | unittest.TestCase): |
---|
722 | n/a | compression = zipfile.ZIP_BZIP2 |
---|
723 | n/a | |
---|
724 | n/a | @requires_lzma |
---|
725 | n/a | class LzmaTestZip64InSmallFiles(AbstractTestZip64InSmallFiles, |
---|
726 | n/a | unittest.TestCase): |
---|
727 | n/a | compression = zipfile.ZIP_LZMA |
---|
728 | n/a | |
---|
729 | n/a | |
---|
730 | n/a | class PyZipFileTests(unittest.TestCase): |
---|
731 | n/a | def assertCompiledIn(self, name, namelist): |
---|
732 | n/a | if name + 'o' not in namelist: |
---|
733 | n/a | self.assertIn(name + 'c', namelist) |
---|
734 | n/a | |
---|
735 | n/a | def requiresWriteAccess(self, path): |
---|
736 | n/a | # effective_ids unavailable on windows |
---|
737 | n/a | if not os.access(path, os.W_OK, |
---|
738 | n/a | effective_ids=os.access in os.supports_effective_ids): |
---|
739 | n/a | self.skipTest('requires write access to the installed location') |
---|
740 | n/a | filename = os.path.join(path, 'test_zipfile.try') |
---|
741 | n/a | try: |
---|
742 | n/a | fd = os.open(filename, os.O_WRONLY | os.O_CREAT) |
---|
743 | n/a | os.close(fd) |
---|
744 | n/a | except Exception: |
---|
745 | n/a | self.skipTest('requires write access to the installed location') |
---|
746 | n/a | unlink(filename) |
---|
747 | n/a | |
---|
748 | n/a | def test_write_pyfile(self): |
---|
749 | n/a | self.requiresWriteAccess(os.path.dirname(__file__)) |
---|
750 | n/a | with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: |
---|
751 | n/a | fn = __file__ |
---|
752 | n/a | if fn.endswith('.pyc'): |
---|
753 | n/a | path_split = fn.split(os.sep) |
---|
754 | n/a | if os.altsep is not None: |
---|
755 | n/a | path_split.extend(fn.split(os.altsep)) |
---|
756 | n/a | if '__pycache__' in path_split: |
---|
757 | n/a | fn = importlib.util.source_from_cache(fn) |
---|
758 | n/a | else: |
---|
759 | n/a | fn = fn[:-1] |
---|
760 | n/a | |
---|
761 | n/a | zipfp.writepy(fn) |
---|
762 | n/a | |
---|
763 | n/a | bn = os.path.basename(fn) |
---|
764 | n/a | self.assertNotIn(bn, zipfp.namelist()) |
---|
765 | n/a | self.assertCompiledIn(bn, zipfp.namelist()) |
---|
766 | n/a | |
---|
767 | n/a | with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: |
---|
768 | n/a | fn = __file__ |
---|
769 | n/a | if fn.endswith('.pyc'): |
---|
770 | n/a | fn = fn[:-1] |
---|
771 | n/a | |
---|
772 | n/a | zipfp.writepy(fn, "testpackage") |
---|
773 | n/a | |
---|
774 | n/a | bn = "%s/%s" % ("testpackage", os.path.basename(fn)) |
---|
775 | n/a | self.assertNotIn(bn, zipfp.namelist()) |
---|
776 | n/a | self.assertCompiledIn(bn, zipfp.namelist()) |
---|
777 | n/a | |
---|
778 | n/a | def test_write_python_package(self): |
---|
779 | n/a | import email |
---|
780 | n/a | packagedir = os.path.dirname(email.__file__) |
---|
781 | n/a | self.requiresWriteAccess(packagedir) |
---|
782 | n/a | |
---|
783 | n/a | with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: |
---|
784 | n/a | zipfp.writepy(packagedir) |
---|
785 | n/a | |
---|
786 | n/a | # Check for a couple of modules at different levels of the |
---|
787 | n/a | # hierarchy |
---|
788 | n/a | names = zipfp.namelist() |
---|
789 | n/a | self.assertCompiledIn('email/__init__.py', names) |
---|
790 | n/a | self.assertCompiledIn('email/mime/text.py', names) |
---|
791 | n/a | |
---|
792 | n/a | def test_write_filtered_python_package(self): |
---|
793 | n/a | import test |
---|
794 | n/a | packagedir = os.path.dirname(test.__file__) |
---|
795 | n/a | self.requiresWriteAccess(packagedir) |
---|
796 | n/a | |
---|
797 | n/a | with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: |
---|
798 | n/a | |
---|
799 | n/a | # first make sure that the test folder gives error messages |
---|
800 | n/a | # (on the badsyntax_... files) |
---|
801 | n/a | with captured_stdout() as reportSIO: |
---|
802 | n/a | zipfp.writepy(packagedir) |
---|
803 | n/a | reportStr = reportSIO.getvalue() |
---|
804 | n/a | self.assertTrue('SyntaxError' in reportStr) |
---|
805 | n/a | |
---|
806 | n/a | # then check that the filter works on the whole package |
---|
807 | n/a | with captured_stdout() as reportSIO: |
---|
808 | n/a | zipfp.writepy(packagedir, filterfunc=lambda whatever: False) |
---|
809 | n/a | reportStr = reportSIO.getvalue() |
---|
810 | n/a | self.assertTrue('SyntaxError' not in reportStr) |
---|
811 | n/a | |
---|
812 | n/a | # then check that the filter works on individual files |
---|
813 | n/a | def filter(path): |
---|
814 | n/a | return not os.path.basename(path).startswith("bad") |
---|
815 | n/a | with captured_stdout() as reportSIO, self.assertWarns(UserWarning): |
---|
816 | n/a | zipfp.writepy(packagedir, filterfunc=filter) |
---|
817 | n/a | reportStr = reportSIO.getvalue() |
---|
818 | n/a | if reportStr: |
---|
819 | n/a | print(reportStr) |
---|
820 | n/a | self.assertTrue('SyntaxError' not in reportStr) |
---|
821 | n/a | |
---|
822 | n/a | def test_write_with_optimization(self): |
---|
823 | n/a | import email |
---|
824 | n/a | packagedir = os.path.dirname(email.__file__) |
---|
825 | n/a | self.requiresWriteAccess(packagedir) |
---|
826 | n/a | optlevel = 1 if __debug__ else 0 |
---|
827 | n/a | ext = '.pyc' |
---|
828 | n/a | |
---|
829 | n/a | with TemporaryFile() as t, \ |
---|
830 | n/a | zipfile.PyZipFile(t, "w", optimize=optlevel) as zipfp: |
---|
831 | n/a | zipfp.writepy(packagedir) |
---|
832 | n/a | |
---|
833 | n/a | names = zipfp.namelist() |
---|
834 | n/a | self.assertIn('email/__init__' + ext, names) |
---|
835 | n/a | self.assertIn('email/mime/text' + ext, names) |
---|
836 | n/a | |
---|
837 | n/a | def test_write_python_directory(self): |
---|
838 | n/a | os.mkdir(TESTFN2) |
---|
839 | n/a | try: |
---|
840 | n/a | with open(os.path.join(TESTFN2, "mod1.py"), "w") as fp: |
---|
841 | n/a | fp.write("print(42)\n") |
---|
842 | n/a | |
---|
843 | n/a | with open(os.path.join(TESTFN2, "mod2.py"), "w") as fp: |
---|
844 | n/a | fp.write("print(42 * 42)\n") |
---|
845 | n/a | |
---|
846 | n/a | with open(os.path.join(TESTFN2, "mod2.txt"), "w") as fp: |
---|
847 | n/a | fp.write("bla bla bla\n") |
---|
848 | n/a | |
---|
849 | n/a | with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: |
---|
850 | n/a | zipfp.writepy(TESTFN2) |
---|
851 | n/a | |
---|
852 | n/a | names = zipfp.namelist() |
---|
853 | n/a | self.assertCompiledIn('mod1.py', names) |
---|
854 | n/a | self.assertCompiledIn('mod2.py', names) |
---|
855 | n/a | self.assertNotIn('mod2.txt', names) |
---|
856 | n/a | |
---|
857 | n/a | finally: |
---|
858 | n/a | rmtree(TESTFN2) |
---|
859 | n/a | |
---|
860 | n/a | def test_write_python_directory_filtered(self): |
---|
861 | n/a | os.mkdir(TESTFN2) |
---|
862 | n/a | try: |
---|
863 | n/a | with open(os.path.join(TESTFN2, "mod1.py"), "w") as fp: |
---|
864 | n/a | fp.write("print(42)\n") |
---|
865 | n/a | |
---|
866 | n/a | with open(os.path.join(TESTFN2, "mod2.py"), "w") as fp: |
---|
867 | n/a | fp.write("print(42 * 42)\n") |
---|
868 | n/a | |
---|
869 | n/a | with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: |
---|
870 | n/a | zipfp.writepy(TESTFN2, filterfunc=lambda fn: |
---|
871 | n/a | not fn.endswith('mod2.py')) |
---|
872 | n/a | |
---|
873 | n/a | names = zipfp.namelist() |
---|
874 | n/a | self.assertCompiledIn('mod1.py', names) |
---|
875 | n/a | self.assertNotIn('mod2.py', names) |
---|
876 | n/a | |
---|
877 | n/a | finally: |
---|
878 | n/a | rmtree(TESTFN2) |
---|
879 | n/a | |
---|
880 | n/a | def test_write_non_pyfile(self): |
---|
881 | n/a | with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: |
---|
882 | n/a | with open(TESTFN, 'w') as f: |
---|
883 | n/a | f.write('most definitely not a python file') |
---|
884 | n/a | self.assertRaises(RuntimeError, zipfp.writepy, TESTFN) |
---|
885 | n/a | unlink(TESTFN) |
---|
886 | n/a | |
---|
887 | n/a | def test_write_pyfile_bad_syntax(self): |
---|
888 | n/a | os.mkdir(TESTFN2) |
---|
889 | n/a | try: |
---|
890 | n/a | with open(os.path.join(TESTFN2, "mod1.py"), "w") as fp: |
---|
891 | n/a | fp.write("Bad syntax in python file\n") |
---|
892 | n/a | |
---|
893 | n/a | with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: |
---|
894 | n/a | # syntax errors are printed to stdout |
---|
895 | n/a | with captured_stdout() as s: |
---|
896 | n/a | zipfp.writepy(os.path.join(TESTFN2, "mod1.py")) |
---|
897 | n/a | |
---|
898 | n/a | self.assertIn("SyntaxError", s.getvalue()) |
---|
899 | n/a | |
---|
900 | n/a | # as it will not have compiled the python file, it will |
---|
901 | n/a | # include the .py file not .pyc |
---|
902 | n/a | names = zipfp.namelist() |
---|
903 | n/a | self.assertIn('mod1.py', names) |
---|
904 | n/a | self.assertNotIn('mod1.pyc', names) |
---|
905 | n/a | |
---|
906 | n/a | finally: |
---|
907 | n/a | rmtree(TESTFN2) |
---|
908 | n/a | |
---|
909 | n/a | |
---|
910 | n/a | class ExtractTests(unittest.TestCase): |
---|
911 | n/a | def test_extract(self): |
---|
912 | n/a | with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: |
---|
913 | n/a | for fpath, fdata in SMALL_TEST_DATA: |
---|
914 | n/a | zipfp.writestr(fpath, fdata) |
---|
915 | n/a | |
---|
916 | n/a | with zipfile.ZipFile(TESTFN2, "r") as zipfp: |
---|
917 | n/a | for fpath, fdata in SMALL_TEST_DATA: |
---|
918 | n/a | writtenfile = zipfp.extract(fpath) |
---|
919 | n/a | |
---|
920 | n/a | # make sure it was written to the right place |
---|
921 | n/a | correctfile = os.path.join(os.getcwd(), fpath) |
---|
922 | n/a | correctfile = os.path.normpath(correctfile) |
---|
923 | n/a | |
---|
924 | n/a | self.assertEqual(writtenfile, correctfile) |
---|
925 | n/a | |
---|
926 | n/a | # make sure correct data is in correct file |
---|
927 | n/a | with open(writtenfile, "rb") as f: |
---|
928 | n/a | self.assertEqual(fdata.encode(), f.read()) |
---|
929 | n/a | |
---|
930 | n/a | unlink(writtenfile) |
---|
931 | n/a | |
---|
932 | n/a | # remove the test file subdirectories |
---|
933 | n/a | rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) |
---|
934 | n/a | |
---|
935 | n/a | def test_extract_all(self): |
---|
936 | n/a | with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: |
---|
937 | n/a | for fpath, fdata in SMALL_TEST_DATA: |
---|
938 | n/a | zipfp.writestr(fpath, fdata) |
---|
939 | n/a | |
---|
940 | n/a | with zipfile.ZipFile(TESTFN2, "r") as zipfp: |
---|
941 | n/a | zipfp.extractall() |
---|
942 | n/a | for fpath, fdata in SMALL_TEST_DATA: |
---|
943 | n/a | outfile = os.path.join(os.getcwd(), fpath) |
---|
944 | n/a | |
---|
945 | n/a | with open(outfile, "rb") as f: |
---|
946 | n/a | self.assertEqual(fdata.encode(), f.read()) |
---|
947 | n/a | |
---|
948 | n/a | unlink(outfile) |
---|
949 | n/a | |
---|
950 | n/a | # remove the test file subdirectories |
---|
951 | n/a | rmtree(os.path.join(os.getcwd(), 'ziptest2dir')) |
---|
952 | n/a | |
---|
953 | n/a | def check_file(self, filename, content): |
---|
954 | n/a | self.assertTrue(os.path.isfile(filename)) |
---|
955 | n/a | with open(filename, 'rb') as f: |
---|
956 | n/a | self.assertEqual(f.read(), content) |
---|
957 | n/a | |
---|
958 | n/a | def test_sanitize_windows_name(self): |
---|
959 | n/a | san = zipfile.ZipFile._sanitize_windows_name |
---|
960 | n/a | # Passing pathsep in allows this test to work regardless of platform. |
---|
961 | n/a | self.assertEqual(san(r',,?,C:,foo,bar/z', ','), r'_,C_,foo,bar/z') |
---|
962 | n/a | self.assertEqual(san(r'a\b,c<d>e|f"g?h*i', ','), r'a\b,c_d_e_f_g_h_i') |
---|
963 | n/a | self.assertEqual(san('../../foo../../ba..r', '/'), r'foo/ba..r') |
---|
964 | n/a | |
---|
965 | n/a | def test_extract_hackers_arcnames_common_cases(self): |
---|
966 | n/a | common_hacknames = [ |
---|
967 | n/a | ('../foo/bar', 'foo/bar'), |
---|
968 | n/a | ('foo/../bar', 'foo/bar'), |
---|
969 | n/a | ('foo/../../bar', 'foo/bar'), |
---|
970 | n/a | ('foo/bar/..', 'foo/bar'), |
---|
971 | n/a | ('./../foo/bar', 'foo/bar'), |
---|
972 | n/a | ('/foo/bar', 'foo/bar'), |
---|
973 | n/a | ('/foo/../bar', 'foo/bar'), |
---|
974 | n/a | ('/foo/../../bar', 'foo/bar'), |
---|
975 | n/a | ] |
---|
976 | n/a | self._test_extract_hackers_arcnames(common_hacknames) |
---|
977 | n/a | |
---|
978 | n/a | @unittest.skipIf(os.path.sep != '\\', 'Requires \\ as path separator.') |
---|
979 | n/a | def test_extract_hackers_arcnames_windows_only(self): |
---|
980 | n/a | """Test combination of path fixing and windows name sanitization.""" |
---|
981 | n/a | windows_hacknames = [ |
---|
982 | n/a | (r'..\foo\bar', 'foo/bar'), |
---|
983 | n/a | (r'..\/foo\/bar', 'foo/bar'), |
---|
984 | n/a | (r'foo/\..\/bar', 'foo/bar'), |
---|
985 | n/a | (r'foo\/../\bar', 'foo/bar'), |
---|
986 | n/a | (r'C:foo/bar', 'foo/bar'), |
---|
987 | n/a | (r'C:/foo/bar', 'foo/bar'), |
---|
988 | n/a | (r'C://foo/bar', 'foo/bar'), |
---|
989 | n/a | (r'C:\foo\bar', 'foo/bar'), |
---|
990 | n/a | (r'//conky/mountpoint/foo/bar', 'foo/bar'), |
---|
991 | n/a | (r'\\conky\mountpoint\foo\bar', 'foo/bar'), |
---|
992 | n/a | (r'///conky/mountpoint/foo/bar', 'conky/mountpoint/foo/bar'), |
---|
993 | n/a | (r'\\\conky\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'), |
---|
994 | n/a | (r'//conky//mountpoint/foo/bar', 'conky/mountpoint/foo/bar'), |
---|
995 | n/a | (r'\\conky\\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'), |
---|
996 | n/a | (r'//?/C:/foo/bar', 'foo/bar'), |
---|
997 | n/a | (r'\\?\C:\foo\bar', 'foo/bar'), |
---|
998 | n/a | (r'C:/../C:/foo/bar', 'C_/foo/bar'), |
---|
999 | n/a | (r'a:b\c<d>e|f"g?h*i', 'b/c_d_e_f_g_h_i'), |
---|
1000 | n/a | ('../../foo../../ba..r', 'foo/ba..r'), |
---|
1001 | n/a | ] |
---|
1002 | n/a | self._test_extract_hackers_arcnames(windows_hacknames) |
---|
1003 | n/a | |
---|
1004 | n/a | @unittest.skipIf(os.path.sep != '/', r'Requires / as path separator.') |
---|
1005 | n/a | def test_extract_hackers_arcnames_posix_only(self): |
---|
1006 | n/a | posix_hacknames = [ |
---|
1007 | n/a | ('//foo/bar', 'foo/bar'), |
---|
1008 | n/a | ('../../foo../../ba..r', 'foo../ba..r'), |
---|
1009 | n/a | (r'foo/..\bar', r'foo/..\bar'), |
---|
1010 | n/a | ] |
---|
1011 | n/a | self._test_extract_hackers_arcnames(posix_hacknames) |
---|
1012 | n/a | |
---|
1013 | n/a | def _test_extract_hackers_arcnames(self, hacknames): |
---|
1014 | n/a | for arcname, fixedname in hacknames: |
---|
1015 | n/a | content = b'foobar' + arcname.encode() |
---|
1016 | n/a | with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipfp: |
---|
1017 | n/a | zinfo = zipfile.ZipInfo() |
---|
1018 | n/a | # preserve backslashes |
---|
1019 | n/a | zinfo.filename = arcname |
---|
1020 | n/a | zinfo.external_attr = 0o600 << 16 |
---|
1021 | n/a | zipfp.writestr(zinfo, content) |
---|
1022 | n/a | |
---|
1023 | n/a | arcname = arcname.replace(os.sep, "/") |
---|
1024 | n/a | targetpath = os.path.join('target', 'subdir', 'subsub') |
---|
1025 | n/a | correctfile = os.path.join(targetpath, *fixedname.split('/')) |
---|
1026 | n/a | |
---|
1027 | n/a | with zipfile.ZipFile(TESTFN2, 'r') as zipfp: |
---|
1028 | n/a | writtenfile = zipfp.extract(arcname, targetpath) |
---|
1029 | n/a | self.assertEqual(writtenfile, correctfile, |
---|
1030 | n/a | msg='extract %r: %r != %r' % |
---|
1031 | n/a | (arcname, writtenfile, correctfile)) |
---|
1032 | n/a | self.check_file(correctfile, content) |
---|
1033 | n/a | rmtree('target') |
---|
1034 | n/a | |
---|
1035 | n/a | with zipfile.ZipFile(TESTFN2, 'r') as zipfp: |
---|
1036 | n/a | zipfp.extractall(targetpath) |
---|
1037 | n/a | self.check_file(correctfile, content) |
---|
1038 | n/a | rmtree('target') |
---|
1039 | n/a | |
---|
1040 | n/a | correctfile = os.path.join(os.getcwd(), *fixedname.split('/')) |
---|
1041 | n/a | |
---|
1042 | n/a | with zipfile.ZipFile(TESTFN2, 'r') as zipfp: |
---|
1043 | n/a | writtenfile = zipfp.extract(arcname) |
---|
1044 | n/a | self.assertEqual(writtenfile, correctfile, |
---|
1045 | n/a | msg="extract %r" % arcname) |
---|
1046 | n/a | self.check_file(correctfile, content) |
---|
1047 | n/a | rmtree(fixedname.split('/')[0]) |
---|
1048 | n/a | |
---|
1049 | n/a | with zipfile.ZipFile(TESTFN2, 'r') as zipfp: |
---|
1050 | n/a | zipfp.extractall() |
---|
1051 | n/a | self.check_file(correctfile, content) |
---|
1052 | n/a | rmtree(fixedname.split('/')[0]) |
---|
1053 | n/a | |
---|
1054 | n/a | unlink(TESTFN2) |
---|
1055 | n/a | |
---|
1056 | n/a | |
---|
1057 | n/a | class OtherTests(unittest.TestCase): |
---|
1058 | n/a | def test_open_via_zip_info(self): |
---|
1059 | n/a | # Create the ZIP archive |
---|
1060 | n/a | with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: |
---|
1061 | n/a | zipfp.writestr("name", "foo") |
---|
1062 | n/a | with self.assertWarns(UserWarning): |
---|
1063 | n/a | zipfp.writestr("name", "bar") |
---|
1064 | n/a | self.assertEqual(zipfp.namelist(), ["name"] * 2) |
---|
1065 | n/a | |
---|
1066 | n/a | with zipfile.ZipFile(TESTFN2, "r") as zipfp: |
---|
1067 | n/a | infos = zipfp.infolist() |
---|
1068 | n/a | data = b"" |
---|
1069 | n/a | for info in infos: |
---|
1070 | n/a | with zipfp.open(info) as zipopen: |
---|
1071 | n/a | data += zipopen.read() |
---|
1072 | n/a | self.assertIn(data, {b"foobar", b"barfoo"}) |
---|
1073 | n/a | data = b"" |
---|
1074 | n/a | for info in infos: |
---|
1075 | n/a | data += zipfp.read(info) |
---|
1076 | n/a | self.assertIn(data, {b"foobar", b"barfoo"}) |
---|
1077 | n/a | |
---|
1078 | n/a | def test_writestr_extended_local_header_issue1202(self): |
---|
1079 | n/a | with zipfile.ZipFile(TESTFN2, 'w') as orig_zip: |
---|
1080 | n/a | for data in 'abcdefghijklmnop': |
---|
1081 | n/a | zinfo = zipfile.ZipInfo(data) |
---|
1082 | n/a | zinfo.flag_bits |= 0x08 # Include an extended local header. |
---|
1083 | n/a | orig_zip.writestr(zinfo, data) |
---|
1084 | n/a | |
---|
1085 | n/a | def test_close(self): |
---|
1086 | n/a | """Check that the zipfile is closed after the 'with' block.""" |
---|
1087 | n/a | with zipfile.ZipFile(TESTFN2, "w") as zipfp: |
---|
1088 | n/a | for fpath, fdata in SMALL_TEST_DATA: |
---|
1089 | n/a | zipfp.writestr(fpath, fdata) |
---|
1090 | n/a | self.assertIsNotNone(zipfp.fp, 'zipfp is not open') |
---|
1091 | n/a | self.assertIsNone(zipfp.fp, 'zipfp is not closed') |
---|
1092 | n/a | |
---|
1093 | n/a | with zipfile.ZipFile(TESTFN2, "r") as zipfp: |
---|
1094 | n/a | self.assertIsNotNone(zipfp.fp, 'zipfp is not open') |
---|
1095 | n/a | self.assertIsNone(zipfp.fp, 'zipfp is not closed') |
---|
1096 | n/a | |
---|
1097 | n/a | def test_close_on_exception(self): |
---|
1098 | n/a | """Check that the zipfile is closed if an exception is raised in the |
---|
1099 | n/a | 'with' block.""" |
---|
1100 | n/a | with zipfile.ZipFile(TESTFN2, "w") as zipfp: |
---|
1101 | n/a | for fpath, fdata in SMALL_TEST_DATA: |
---|
1102 | n/a | zipfp.writestr(fpath, fdata) |
---|
1103 | n/a | |
---|
1104 | n/a | try: |
---|
1105 | n/a | with zipfile.ZipFile(TESTFN2, "r") as zipfp2: |
---|
1106 | n/a | raise zipfile.BadZipFile() |
---|
1107 | n/a | except zipfile.BadZipFile: |
---|
1108 | n/a | self.assertIsNone(zipfp2.fp, 'zipfp is not closed') |
---|
1109 | n/a | |
---|
1110 | n/a | def test_unsupported_version(self): |
---|
1111 | n/a | # File has an extract_version of 120 |
---|
1112 | n/a | data = (b'PK\x03\x04x\x00\x00\x00\x00\x00!p\xa1@\x00\x00\x00\x00\x00\x00' |
---|
1113 | n/a | b'\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00xPK\x01\x02x\x03x\x00\x00\x00\x00' |
---|
1114 | n/a | b'\x00!p\xa1@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00' |
---|
1115 | n/a | b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00\x00xPK\x05\x06' |
---|
1116 | n/a | b'\x00\x00\x00\x00\x01\x00\x01\x00/\x00\x00\x00\x1f\x00\x00\x00\x00\x00') |
---|
1117 | n/a | |
---|
1118 | n/a | self.assertRaises(NotImplementedError, zipfile.ZipFile, |
---|
1119 | n/a | io.BytesIO(data), 'r') |
---|
1120 | n/a | |
---|
1121 | n/a | @requires_zlib |
---|
1122 | n/a | def test_read_unicode_filenames(self): |
---|
1123 | n/a | # bug #10801 |
---|
1124 | n/a | fname = findfile('zip_cp437_header.zip') |
---|
1125 | n/a | with zipfile.ZipFile(fname) as zipfp: |
---|
1126 | n/a | for name in zipfp.namelist(): |
---|
1127 | n/a | zipfp.open(name).close() |
---|
1128 | n/a | |
---|
1129 | n/a | def test_write_unicode_filenames(self): |
---|
1130 | n/a | with zipfile.ZipFile(TESTFN, "w") as zf: |
---|
1131 | n/a | zf.writestr("foo.txt", "Test for unicode filename") |
---|
1132 | n/a | zf.writestr("\xf6.txt", "Test for unicode filename") |
---|
1133 | n/a | self.assertIsInstance(zf.infolist()[0].filename, str) |
---|
1134 | n/a | |
---|
1135 | n/a | with zipfile.ZipFile(TESTFN, "r") as zf: |
---|
1136 | n/a | self.assertEqual(zf.filelist[0].filename, "foo.txt") |
---|
1137 | n/a | self.assertEqual(zf.filelist[1].filename, "\xf6.txt") |
---|
1138 | n/a | |
---|
1139 | n/a | def test_exclusive_create_zip_file(self): |
---|
1140 | n/a | """Test exclusive creating a new zipfile.""" |
---|
1141 | n/a | unlink(TESTFN2) |
---|
1142 | n/a | filename = 'testfile.txt' |
---|
1143 | n/a | content = b'hello, world. this is some content.' |
---|
1144 | n/a | with zipfile.ZipFile(TESTFN2, "x", zipfile.ZIP_STORED) as zipfp: |
---|
1145 | n/a | zipfp.writestr(filename, content) |
---|
1146 | n/a | with self.assertRaises(FileExistsError): |
---|
1147 | n/a | zipfile.ZipFile(TESTFN2, "x", zipfile.ZIP_STORED) |
---|
1148 | n/a | with zipfile.ZipFile(TESTFN2, "r") as zipfp: |
---|
1149 | n/a | self.assertEqual(zipfp.namelist(), [filename]) |
---|
1150 | n/a | self.assertEqual(zipfp.read(filename), content) |
---|
1151 | n/a | |
---|
1152 | n/a | def test_create_non_existent_file_for_append(self): |
---|
1153 | n/a | if os.path.exists(TESTFN): |
---|
1154 | n/a | os.unlink(TESTFN) |
---|
1155 | n/a | |
---|
1156 | n/a | filename = 'testfile.txt' |
---|
1157 | n/a | content = b'hello, world. this is some content.' |
---|
1158 | n/a | |
---|
1159 | n/a | try: |
---|
1160 | n/a | with zipfile.ZipFile(TESTFN, 'a') as zf: |
---|
1161 | n/a | zf.writestr(filename, content) |
---|
1162 | n/a | except OSError: |
---|
1163 | n/a | self.fail('Could not append data to a non-existent zip file.') |
---|
1164 | n/a | |
---|
1165 | n/a | self.assertTrue(os.path.exists(TESTFN)) |
---|
1166 | n/a | |
---|
1167 | n/a | with zipfile.ZipFile(TESTFN, 'r') as zf: |
---|
1168 | n/a | self.assertEqual(zf.read(filename), content) |
---|
1169 | n/a | |
---|
1170 | n/a | def test_close_erroneous_file(self): |
---|
1171 | n/a | # This test checks that the ZipFile constructor closes the file object |
---|
1172 | n/a | # it opens if there's an error in the file. If it doesn't, the |
---|
1173 | n/a | # traceback holds a reference to the ZipFile object and, indirectly, |
---|
1174 | n/a | # the file object. |
---|
1175 | n/a | # On Windows, this causes the os.unlink() call to fail because the |
---|
1176 | n/a | # underlying file is still open. This is SF bug #412214. |
---|
1177 | n/a | # |
---|
1178 | n/a | with open(TESTFN, "w") as fp: |
---|
1179 | n/a | fp.write("this is not a legal zip file\n") |
---|
1180 | n/a | try: |
---|
1181 | n/a | zf = zipfile.ZipFile(TESTFN) |
---|
1182 | n/a | except zipfile.BadZipFile: |
---|
1183 | n/a | pass |
---|
1184 | n/a | |
---|
1185 | n/a | def test_is_zip_erroneous_file(self): |
---|
1186 | n/a | """Check that is_zipfile() correctly identifies non-zip files.""" |
---|
1187 | n/a | # - passing a filename |
---|
1188 | n/a | with open(TESTFN, "w") as fp: |
---|
1189 | n/a | fp.write("this is not a legal zip file\n") |
---|
1190 | n/a | self.assertFalse(zipfile.is_zipfile(TESTFN)) |
---|
1191 | n/a | # - passing a file object |
---|
1192 | n/a | with open(TESTFN, "rb") as fp: |
---|
1193 | n/a | self.assertFalse(zipfile.is_zipfile(fp)) |
---|
1194 | n/a | # - passing a file-like object |
---|
1195 | n/a | fp = io.BytesIO() |
---|
1196 | n/a | fp.write(b"this is not a legal zip file\n") |
---|
1197 | n/a | self.assertFalse(zipfile.is_zipfile(fp)) |
---|
1198 | n/a | fp.seek(0, 0) |
---|
1199 | n/a | self.assertFalse(zipfile.is_zipfile(fp)) |
---|
1200 | n/a | |
---|
1201 | n/a | def test_damaged_zipfile(self): |
---|
1202 | n/a | """Check that zipfiles with missing bytes at the end raise BadZipFile.""" |
---|
1203 | n/a | # - Create a valid zip file |
---|
1204 | n/a | fp = io.BytesIO() |
---|
1205 | n/a | with zipfile.ZipFile(fp, mode="w") as zipf: |
---|
1206 | n/a | zipf.writestr("foo.txt", b"O, for a Muse of Fire!") |
---|
1207 | n/a | zipfiledata = fp.getvalue() |
---|
1208 | n/a | |
---|
1209 | n/a | # - Now create copies of it missing the last N bytes and make sure |
---|
1210 | n/a | # a BadZipFile exception is raised when we try to open it |
---|
1211 | n/a | for N in range(len(zipfiledata)): |
---|
1212 | n/a | fp = io.BytesIO(zipfiledata[:N]) |
---|
1213 | n/a | self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, fp) |
---|
1214 | n/a | |
---|
1215 | n/a | def test_is_zip_valid_file(self): |
---|
1216 | n/a | """Check that is_zipfile() correctly identifies zip files.""" |
---|
1217 | n/a | # - passing a filename |
---|
1218 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1219 | n/a | zipf.writestr("foo.txt", b"O, for a Muse of Fire!") |
---|
1220 | n/a | |
---|
1221 | n/a | self.assertTrue(zipfile.is_zipfile(TESTFN)) |
---|
1222 | n/a | # - passing a file object |
---|
1223 | n/a | with open(TESTFN, "rb") as fp: |
---|
1224 | n/a | self.assertTrue(zipfile.is_zipfile(fp)) |
---|
1225 | n/a | fp.seek(0, 0) |
---|
1226 | n/a | zip_contents = fp.read() |
---|
1227 | n/a | # - passing a file-like object |
---|
1228 | n/a | fp = io.BytesIO() |
---|
1229 | n/a | fp.write(zip_contents) |
---|
1230 | n/a | self.assertTrue(zipfile.is_zipfile(fp)) |
---|
1231 | n/a | fp.seek(0, 0) |
---|
1232 | n/a | self.assertTrue(zipfile.is_zipfile(fp)) |
---|
1233 | n/a | |
---|
1234 | n/a | def test_non_existent_file_raises_OSError(self): |
---|
1235 | n/a | # make sure we don't raise an AttributeError when a partially-constructed |
---|
1236 | n/a | # ZipFile instance is finalized; this tests for regression on SF tracker |
---|
1237 | n/a | # bug #403871. |
---|
1238 | n/a | |
---|
1239 | n/a | # The bug we're testing for caused an AttributeError to be raised |
---|
1240 | n/a | # when a ZipFile instance was created for a file that did not |
---|
1241 | n/a | # exist; the .fp member was not initialized but was needed by the |
---|
1242 | n/a | # __del__() method. Since the AttributeError is in the __del__(), |
---|
1243 | n/a | # it is ignored, but the user should be sufficiently annoyed by |
---|
1244 | n/a | # the message on the output that regression will be noticed |
---|
1245 | n/a | # quickly. |
---|
1246 | n/a | self.assertRaises(OSError, zipfile.ZipFile, TESTFN) |
---|
1247 | n/a | |
---|
1248 | n/a | def test_empty_file_raises_BadZipFile(self): |
---|
1249 | n/a | f = open(TESTFN, 'w') |
---|
1250 | n/a | f.close() |
---|
1251 | n/a | self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) |
---|
1252 | n/a | |
---|
1253 | n/a | with open(TESTFN, 'w') as fp: |
---|
1254 | n/a | fp.write("short file") |
---|
1255 | n/a | self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) |
---|
1256 | n/a | |
---|
1257 | n/a | def test_closed_zip_raises_ValueError(self): |
---|
1258 | n/a | """Verify that testzip() doesn't swallow inappropriate exceptions.""" |
---|
1259 | n/a | data = io.BytesIO() |
---|
1260 | n/a | with zipfile.ZipFile(data, mode="w") as zipf: |
---|
1261 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1262 | n/a | |
---|
1263 | n/a | # This is correct; calling .read on a closed ZipFile should raise |
---|
1264 | n/a | # a ValueError, and so should calling .testzip. An earlier |
---|
1265 | n/a | # version of .testzip would swallow this exception (and any other) |
---|
1266 | n/a | # and report that the first file in the archive was corrupt. |
---|
1267 | n/a | self.assertRaises(ValueError, zipf.read, "foo.txt") |
---|
1268 | n/a | self.assertRaises(ValueError, zipf.open, "foo.txt") |
---|
1269 | n/a | self.assertRaises(ValueError, zipf.testzip) |
---|
1270 | n/a | self.assertRaises(ValueError, zipf.writestr, "bogus.txt", "bogus") |
---|
1271 | n/a | with open(TESTFN, 'w') as f: |
---|
1272 | n/a | f.write('zipfile test data') |
---|
1273 | n/a | self.assertRaises(ValueError, zipf.write, TESTFN) |
---|
1274 | n/a | |
---|
1275 | n/a | def test_bad_constructor_mode(self): |
---|
1276 | n/a | """Check that bad modes passed to ZipFile constructor are caught.""" |
---|
1277 | n/a | self.assertRaises(ValueError, zipfile.ZipFile, TESTFN, "q") |
---|
1278 | n/a | |
---|
1279 | n/a | def test_bad_open_mode(self): |
---|
1280 | n/a | """Check that bad modes passed to ZipFile.open are caught.""" |
---|
1281 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1282 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1283 | n/a | |
---|
1284 | n/a | with zipfile.ZipFile(TESTFN, mode="r") as zipf: |
---|
1285 | n/a | # read the data to make sure the file is there |
---|
1286 | n/a | zipf.read("foo.txt") |
---|
1287 | n/a | self.assertRaises(ValueError, zipf.open, "foo.txt", "q") |
---|
1288 | n/a | # universal newlines support is removed |
---|
1289 | n/a | self.assertRaises(ValueError, zipf.open, "foo.txt", "U") |
---|
1290 | n/a | self.assertRaises(ValueError, zipf.open, "foo.txt", "rU") |
---|
1291 | n/a | |
---|
1292 | n/a | def test_read0(self): |
---|
1293 | n/a | """Check that calling read(0) on a ZipExtFile object returns an empty |
---|
1294 | n/a | string and doesn't advance file pointer.""" |
---|
1295 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1296 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1297 | n/a | # read the data to make sure the file is there |
---|
1298 | n/a | with zipf.open("foo.txt") as f: |
---|
1299 | n/a | for i in range(FIXEDTEST_SIZE): |
---|
1300 | n/a | self.assertEqual(f.read(0), b'') |
---|
1301 | n/a | |
---|
1302 | n/a | self.assertEqual(f.read(), b"O, for a Muse of Fire!") |
---|
1303 | n/a | |
---|
1304 | n/a | def test_open_non_existent_item(self): |
---|
1305 | n/a | """Check that attempting to call open() for an item that doesn't |
---|
1306 | n/a | exist in the archive raises a RuntimeError.""" |
---|
1307 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1308 | n/a | self.assertRaises(KeyError, zipf.open, "foo.txt", "r") |
---|
1309 | n/a | |
---|
1310 | n/a | def test_bad_compression_mode(self): |
---|
1311 | n/a | """Check that bad compression methods passed to ZipFile.open are |
---|
1312 | n/a | caught.""" |
---|
1313 | n/a | self.assertRaises(NotImplementedError, zipfile.ZipFile, TESTFN, "w", -1) |
---|
1314 | n/a | |
---|
1315 | n/a | def test_unsupported_compression(self): |
---|
1316 | n/a | # data is declared as shrunk, but actually deflated |
---|
1317 | n/a | data = (b'PK\x03\x04.\x00\x00\x00\x01\x00\xe4C\xa1@\x00\x00\x00' |
---|
1318 | n/a | b'\x00\x02\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00x\x03\x00PK\x01' |
---|
1319 | n/a | b'\x02.\x03.\x00\x00\x00\x01\x00\xe4C\xa1@\x00\x00\x00\x00\x02\x00\x00' |
---|
1320 | n/a | b'\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' |
---|
1321 | n/a | b'\x80\x01\x00\x00\x00\x00xPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x00' |
---|
1322 | n/a | b'/\x00\x00\x00!\x00\x00\x00\x00\x00') |
---|
1323 | n/a | with zipfile.ZipFile(io.BytesIO(data), 'r') as zipf: |
---|
1324 | n/a | self.assertRaises(NotImplementedError, zipf.open, 'x') |
---|
1325 | n/a | |
---|
1326 | n/a | def test_null_byte_in_filename(self): |
---|
1327 | n/a | """Check that a filename containing a null byte is properly |
---|
1328 | n/a | terminated.""" |
---|
1329 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1330 | n/a | zipf.writestr("foo.txt\x00qqq", b"O, for a Muse of Fire!") |
---|
1331 | n/a | self.assertEqual(zipf.namelist(), ['foo.txt']) |
---|
1332 | n/a | |
---|
1333 | n/a | def test_struct_sizes(self): |
---|
1334 | n/a | """Check that ZIP internal structure sizes are calculated correctly.""" |
---|
1335 | n/a | self.assertEqual(zipfile.sizeEndCentDir, 22) |
---|
1336 | n/a | self.assertEqual(zipfile.sizeCentralDir, 46) |
---|
1337 | n/a | self.assertEqual(zipfile.sizeEndCentDir64, 56) |
---|
1338 | n/a | self.assertEqual(zipfile.sizeEndCentDir64Locator, 20) |
---|
1339 | n/a | |
---|
1340 | n/a | def test_comments(self): |
---|
1341 | n/a | """Check that comments on the archive are handled properly.""" |
---|
1342 | n/a | |
---|
1343 | n/a | # check default comment is empty |
---|
1344 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1345 | n/a | self.assertEqual(zipf.comment, b'') |
---|
1346 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1347 | n/a | |
---|
1348 | n/a | with zipfile.ZipFile(TESTFN, mode="r") as zipfr: |
---|
1349 | n/a | self.assertEqual(zipfr.comment, b'') |
---|
1350 | n/a | |
---|
1351 | n/a | # check a simple short comment |
---|
1352 | n/a | comment = b'Bravely taking to his feet, he beat a very brave retreat.' |
---|
1353 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1354 | n/a | zipf.comment = comment |
---|
1355 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1356 | n/a | with zipfile.ZipFile(TESTFN, mode="r") as zipfr: |
---|
1357 | n/a | self.assertEqual(zipf.comment, comment) |
---|
1358 | n/a | |
---|
1359 | n/a | # check a comment of max length |
---|
1360 | n/a | comment2 = ''.join(['%d' % (i**3 % 10) for i in range((1 << 16)-1)]) |
---|
1361 | n/a | comment2 = comment2.encode("ascii") |
---|
1362 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1363 | n/a | zipf.comment = comment2 |
---|
1364 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1365 | n/a | |
---|
1366 | n/a | with zipfile.ZipFile(TESTFN, mode="r") as zipfr: |
---|
1367 | n/a | self.assertEqual(zipfr.comment, comment2) |
---|
1368 | n/a | |
---|
1369 | n/a | # check a comment that is too long is truncated |
---|
1370 | n/a | with zipfile.ZipFile(TESTFN, mode="w") as zipf: |
---|
1371 | n/a | with self.assertWarns(UserWarning): |
---|
1372 | n/a | zipf.comment = comment2 + b'oops' |
---|
1373 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1374 | n/a | with zipfile.ZipFile(TESTFN, mode="r") as zipfr: |
---|
1375 | n/a | self.assertEqual(zipfr.comment, comment2) |
---|
1376 | n/a | |
---|
1377 | n/a | # check that comments are correctly modified in append mode |
---|
1378 | n/a | with zipfile.ZipFile(TESTFN,mode="w") as zipf: |
---|
1379 | n/a | zipf.comment = b"original comment" |
---|
1380 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1381 | n/a | with zipfile.ZipFile(TESTFN,mode="a") as zipf: |
---|
1382 | n/a | zipf.comment = b"an updated comment" |
---|
1383 | n/a | with zipfile.ZipFile(TESTFN,mode="r") as zipf: |
---|
1384 | n/a | self.assertEqual(zipf.comment, b"an updated comment") |
---|
1385 | n/a | |
---|
1386 | n/a | # check that comments are correctly shortened in append mode |
---|
1387 | n/a | with zipfile.ZipFile(TESTFN,mode="w") as zipf: |
---|
1388 | n/a | zipf.comment = b"original comment that's longer" |
---|
1389 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1390 | n/a | with zipfile.ZipFile(TESTFN,mode="a") as zipf: |
---|
1391 | n/a | zipf.comment = b"shorter comment" |
---|
1392 | n/a | with zipfile.ZipFile(TESTFN,mode="r") as zipf: |
---|
1393 | n/a | self.assertEqual(zipf.comment, b"shorter comment") |
---|
1394 | n/a | |
---|
1395 | n/a | def test_unicode_comment(self): |
---|
1396 | n/a | with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf: |
---|
1397 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1398 | n/a | with self.assertRaises(TypeError): |
---|
1399 | n/a | zipf.comment = "this is an error" |
---|
1400 | n/a | |
---|
1401 | n/a | def test_change_comment_in_empty_archive(self): |
---|
1402 | n/a | with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf: |
---|
1403 | n/a | self.assertFalse(zipf.filelist) |
---|
1404 | n/a | zipf.comment = b"this is a comment" |
---|
1405 | n/a | with zipfile.ZipFile(TESTFN, "r") as zipf: |
---|
1406 | n/a | self.assertEqual(zipf.comment, b"this is a comment") |
---|
1407 | n/a | |
---|
1408 | n/a | def test_change_comment_in_nonempty_archive(self): |
---|
1409 | n/a | with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf: |
---|
1410 | n/a | zipf.writestr("foo.txt", "O, for a Muse of Fire!") |
---|
1411 | n/a | with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf: |
---|
1412 | n/a | self.assertTrue(zipf.filelist) |
---|
1413 | n/a | zipf.comment = b"this is a comment" |
---|
1414 | n/a | with zipfile.ZipFile(TESTFN, "r") as zipf: |
---|
1415 | n/a | self.assertEqual(zipf.comment, b"this is a comment") |
---|
1416 | n/a | |
---|
1417 | n/a | def test_empty_zipfile(self): |
---|
1418 | n/a | # Check that creating a file in 'w' or 'a' mode and closing without |
---|
1419 | n/a | # adding any files to the archives creates a valid empty ZIP file |
---|
1420 | n/a | zipf = zipfile.ZipFile(TESTFN, mode="w") |
---|
1421 | n/a | zipf.close() |
---|
1422 | n/a | try: |
---|
1423 | n/a | zipf = zipfile.ZipFile(TESTFN, mode="r") |
---|
1424 | n/a | except zipfile.BadZipFile: |
---|
1425 | n/a | self.fail("Unable to create empty ZIP file in 'w' mode") |
---|
1426 | n/a | |
---|
1427 | n/a | zipf = zipfile.ZipFile(TESTFN, mode="a") |
---|
1428 | n/a | zipf.close() |
---|
1429 | n/a | try: |
---|
1430 | n/a | zipf = zipfile.ZipFile(TESTFN, mode="r") |
---|
1431 | n/a | except: |
---|
1432 | n/a | self.fail("Unable to create empty ZIP file in 'a' mode") |
---|
1433 | n/a | |
---|
1434 | n/a | def test_open_empty_file(self): |
---|
1435 | n/a | # Issue 1710703: Check that opening a file with less than 22 bytes |
---|
1436 | n/a | # raises a BadZipFile exception (rather than the previously unhelpful |
---|
1437 | n/a | # OSError) |
---|
1438 | n/a | f = open(TESTFN, 'w') |
---|
1439 | n/a | f.close() |
---|
1440 | n/a | self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN, 'r') |
---|
1441 | n/a | |
---|
1442 | n/a | def test_create_zipinfo_before_1980(self): |
---|
1443 | n/a | self.assertRaises(ValueError, |
---|
1444 | n/a | zipfile.ZipInfo, 'seventies', (1979, 1, 1, 0, 0, 0)) |
---|
1445 | n/a | |
---|
1446 | n/a | def test_zipfile_with_short_extra_field(self): |
---|
1447 | n/a | """If an extra field in the header is less than 4 bytes, skip it.""" |
---|
1448 | n/a | zipdata = ( |
---|
1449 | n/a | b'PK\x03\x04\x14\x00\x00\x00\x00\x00\x93\x9b\xad@\x8b\x9e' |
---|
1450 | n/a | b'\xd9\xd3\x01\x00\x00\x00\x01\x00\x00\x00\x03\x00\x03\x00ab' |
---|
1451 | n/a | b'c\x00\x00\x00APK\x01\x02\x14\x03\x14\x00\x00\x00\x00' |
---|
1452 | n/a | b'\x00\x93\x9b\xad@\x8b\x9e\xd9\xd3\x01\x00\x00\x00\x01\x00\x00' |
---|
1453 | n/a | b'\x00\x03\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa4\x81\x00' |
---|
1454 | n/a | b'\x00\x00\x00abc\x00\x00PK\x05\x06\x00\x00\x00\x00' |
---|
1455 | n/a | b'\x01\x00\x01\x003\x00\x00\x00%\x00\x00\x00\x00\x00' |
---|
1456 | n/a | ) |
---|
1457 | n/a | with zipfile.ZipFile(io.BytesIO(zipdata), 'r') as zipf: |
---|
1458 | n/a | # testzip returns the name of the first corrupt file, or None |
---|
1459 | n/a | self.assertIsNone(zipf.testzip()) |
---|
1460 | n/a | |
---|
1461 | n/a | def test_open_conflicting_handles(self): |
---|
1462 | n/a | # It's only possible to open one writable file handle at a time |
---|
1463 | n/a | msg1 = b"It's fun to charter an accountant!" |
---|
1464 | n/a | msg2 = b"And sail the wide accountant sea" |
---|
1465 | n/a | msg3 = b"To find, explore the funds offshore" |
---|
1466 | n/a | with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipf: |
---|
1467 | n/a | with zipf.open('foo', mode='w') as w2: |
---|
1468 | n/a | w2.write(msg1) |
---|
1469 | n/a | with zipf.open('bar', mode='w') as w1: |
---|
1470 | n/a | with self.assertRaises(ValueError): |
---|
1471 | n/a | zipf.open('handle', mode='w') |
---|
1472 | n/a | with self.assertRaises(ValueError): |
---|
1473 | n/a | zipf.open('foo', mode='r') |
---|
1474 | n/a | with self.assertRaises(ValueError): |
---|
1475 | n/a | zipf.writestr('str', 'abcde') |
---|
1476 | n/a | with self.assertRaises(ValueError): |
---|
1477 | n/a | zipf.write(__file__, 'file') |
---|
1478 | n/a | with self.assertRaises(ValueError): |
---|
1479 | n/a | zipf.close() |
---|
1480 | n/a | w1.write(msg2) |
---|
1481 | n/a | with zipf.open('baz', mode='w') as w2: |
---|
1482 | n/a | w2.write(msg3) |
---|
1483 | n/a | |
---|
1484 | n/a | with zipfile.ZipFile(TESTFN2, 'r') as zipf: |
---|
1485 | n/a | self.assertEqual(zipf.read('foo'), msg1) |
---|
1486 | n/a | self.assertEqual(zipf.read('bar'), msg2) |
---|
1487 | n/a | self.assertEqual(zipf.read('baz'), msg3) |
---|
1488 | n/a | self.assertEqual(zipf.namelist(), ['foo', 'bar', 'baz']) |
---|
1489 | n/a | |
---|
1490 | n/a | def tearDown(self): |
---|
1491 | n/a | unlink(TESTFN) |
---|
1492 | n/a | unlink(TESTFN2) |
---|
1493 | n/a | |
---|
1494 | n/a | |
---|
1495 | n/a | class AbstractBadCrcTests: |
---|
1496 | n/a | def test_testzip_with_bad_crc(self): |
---|
1497 | n/a | """Tests that files with bad CRCs return their name from testzip.""" |
---|
1498 | n/a | zipdata = self.zip_with_bad_crc |
---|
1499 | n/a | |
---|
1500 | n/a | with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: |
---|
1501 | n/a | # testzip returns the name of the first corrupt file, or None |
---|
1502 | n/a | self.assertEqual('afile', zipf.testzip()) |
---|
1503 | n/a | |
---|
1504 | n/a | def test_read_with_bad_crc(self): |
---|
1505 | n/a | """Tests that files with bad CRCs raise a BadZipFile exception when read.""" |
---|
1506 | n/a | zipdata = self.zip_with_bad_crc |
---|
1507 | n/a | |
---|
1508 | n/a | # Using ZipFile.read() |
---|
1509 | n/a | with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: |
---|
1510 | n/a | self.assertRaises(zipfile.BadZipFile, zipf.read, 'afile') |
---|
1511 | n/a | |
---|
1512 | n/a | # Using ZipExtFile.read() |
---|
1513 | n/a | with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: |
---|
1514 | n/a | with zipf.open('afile', 'r') as corrupt_file: |
---|
1515 | n/a | self.assertRaises(zipfile.BadZipFile, corrupt_file.read) |
---|
1516 | n/a | |
---|
1517 | n/a | # Same with small reads (in order to exercise the buffering logic) |
---|
1518 | n/a | with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: |
---|
1519 | n/a | with zipf.open('afile', 'r') as corrupt_file: |
---|
1520 | n/a | corrupt_file.MIN_READ_SIZE = 2 |
---|
1521 | n/a | with self.assertRaises(zipfile.BadZipFile): |
---|
1522 | n/a | while corrupt_file.read(2): |
---|
1523 | n/a | pass |
---|
1524 | n/a | |
---|
1525 | n/a | |
---|
1526 | n/a | class StoredBadCrcTests(AbstractBadCrcTests, unittest.TestCase): |
---|
1527 | n/a | compression = zipfile.ZIP_STORED |
---|
1528 | n/a | zip_with_bad_crc = ( |
---|
1529 | n/a | b'PK\003\004\024\0\0\0\0\0 \213\212;:r' |
---|
1530 | n/a | b'\253\377\f\0\0\0\f\0\0\0\005\0\0\000af' |
---|
1531 | n/a | b'ilehello,AworldP' |
---|
1532 | n/a | b'K\001\002\024\003\024\0\0\0\0\0 \213\212;:' |
---|
1533 | n/a | b'r\253\377\f\0\0\0\f\0\0\0\005\0\0\0\0' |
---|
1534 | n/a | b'\0\0\0\0\0\0\0\200\001\0\0\0\000afi' |
---|
1535 | n/a | b'lePK\005\006\0\0\0\0\001\0\001\0003\000' |
---|
1536 | n/a | b'\0\0/\0\0\0\0\0') |
---|
1537 | n/a | |
---|
1538 | n/a | @requires_zlib |
---|
1539 | n/a | class DeflateBadCrcTests(AbstractBadCrcTests, unittest.TestCase): |
---|
1540 | n/a | compression = zipfile.ZIP_DEFLATED |
---|
1541 | n/a | zip_with_bad_crc = ( |
---|
1542 | n/a | b'PK\x03\x04\x14\x00\x00\x00\x08\x00n}\x0c=FA' |
---|
1543 | n/a | b'KE\x10\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' |
---|
1544 | n/a | b'ile\xcbH\xcd\xc9\xc9W(\xcf/\xcaI\xc9\xa0' |
---|
1545 | n/a | b'=\x13\x00PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00n' |
---|
1546 | n/a | b'}\x0c=FAKE\x10\x00\x00\x00n\x00\x00\x00\x05' |
---|
1547 | n/a | b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00' |
---|
1548 | n/a | b'\x00afilePK\x05\x06\x00\x00\x00\x00\x01\x00' |
---|
1549 | n/a | b'\x01\x003\x00\x00\x003\x00\x00\x00\x00\x00') |
---|
1550 | n/a | |
---|
1551 | n/a | @requires_bz2 |
---|
1552 | n/a | class Bzip2BadCrcTests(AbstractBadCrcTests, unittest.TestCase): |
---|
1553 | n/a | compression = zipfile.ZIP_BZIP2 |
---|
1554 | n/a | zip_with_bad_crc = ( |
---|
1555 | n/a | b'PK\x03\x04\x14\x03\x00\x00\x0c\x00nu\x0c=FA' |
---|
1556 | n/a | b'KE8\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' |
---|
1557 | n/a | b'ileBZh91AY&SY\xd4\xa8\xca' |
---|
1558 | n/a | b'\x7f\x00\x00\x0f\x11\x80@\x00\x06D\x90\x80 \x00 \xa5' |
---|
1559 | n/a | b'P\xd9!\x03\x03\x13\x13\x13\x89\xa9\xa9\xc2u5:\x9f' |
---|
1560 | n/a | b'\x8b\xb9"\x9c(HjTe?\x80PK\x01\x02\x14' |
---|
1561 | n/a | b'\x03\x14\x03\x00\x00\x0c\x00nu\x0c=FAKE8' |
---|
1562 | n/a | b'\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\x00' |
---|
1563 | n/a | b'\x00 \x80\x80\x81\x00\x00\x00\x00afilePK' |
---|
1564 | n/a | b'\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00\x00[\x00' |
---|
1565 | n/a | b'\x00\x00\x00\x00') |
---|
1566 | n/a | |
---|
1567 | n/a | @requires_lzma |
---|
1568 | n/a | class LzmaBadCrcTests(AbstractBadCrcTests, unittest.TestCase): |
---|
1569 | n/a | compression = zipfile.ZIP_LZMA |
---|
1570 | n/a | zip_with_bad_crc = ( |
---|
1571 | n/a | b'PK\x03\x04\x14\x03\x00\x00\x0e\x00nu\x0c=FA' |
---|
1572 | n/a | b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' |
---|
1573 | n/a | b'ile\t\x04\x05\x00]\x00\x00\x00\x04\x004\x19I' |
---|
1574 | n/a | b'\xee\x8d\xe9\x17\x89:3`\tq!.8\x00PK' |
---|
1575 | n/a | b'\x01\x02\x14\x03\x14\x03\x00\x00\x0e\x00nu\x0c=FA' |
---|
1576 | n/a | b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00' |
---|
1577 | n/a | b'\x00\x00\x00\x00 \x80\x80\x81\x00\x00\x00\x00afil' |
---|
1578 | n/a | b'ePK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00' |
---|
1579 | n/a | b'\x00>\x00\x00\x00\x00\x00') |
---|
1580 | n/a | |
---|
1581 | n/a | |
---|
1582 | n/a | class DecryptionTests(unittest.TestCase): |
---|
1583 | n/a | """Check that ZIP decryption works. Since the library does not |
---|
1584 | n/a | support encryption at the moment, we use a pre-generated encrypted |
---|
1585 | n/a | ZIP file.""" |
---|
1586 | n/a | |
---|
1587 | n/a | data = ( |
---|
1588 | n/a | b'PK\x03\x04\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00\x1a\x00' |
---|
1589 | n/a | b'\x00\x00\x08\x00\x00\x00test.txt\xfa\x10\xa0gly|\xfa-\xc5\xc0=\xf9y' |
---|
1590 | n/a | b'\x18\xe0\xa8r\xb3Z}Lg\xbc\xae\xf9|\x9b\x19\xe4\x8b\xba\xbb)\x8c\xb0\xdbl' |
---|
1591 | n/a | b'PK\x01\x02\x14\x00\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00' |
---|
1592 | n/a | b'\x1a\x00\x00\x00\x08\x00\x00\x00\x00\x00\x00\x00\x01\x00 \x00\xb6\x81' |
---|
1593 | n/a | b'\x00\x00\x00\x00test.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x006\x00' |
---|
1594 | n/a | b'\x00\x00L\x00\x00\x00\x00\x00' ) |
---|
1595 | n/a | data2 = ( |
---|
1596 | n/a | b'PK\x03\x04\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02' |
---|
1597 | n/a | b'\x00\x00\x04\x00\x15\x00zeroUT\t\x00\x03\xd6\x8b\x92G\xda\x8b\x92GUx\x04' |
---|
1598 | n/a | b'\x00\xe8\x03\xe8\x03\xc7<M\xb5a\xceX\xa3Y&\x8b{oE\xd7\x9d\x8c\x98\x02\xc0' |
---|
1599 | n/a | b'PK\x07\x08xu\xaa\xb2\x14\x00\x00\x00\x00\x02\x00\x00PK\x01\x02\x17\x03' |
---|
1600 | n/a | b'\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02\x00\x00' |
---|
1601 | n/a | b'\x04\x00\r\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa4\x81\x00\x00\x00\x00ze' |
---|
1602 | n/a | b'roUT\x05\x00\x03\xd6\x8b\x92GUx\x00\x00PK\x05\x06\x00\x00\x00\x00\x01' |
---|
1603 | n/a | b'\x00\x01\x00?\x00\x00\x00[\x00\x00\x00\x00\x00' ) |
---|
1604 | n/a | |
---|
1605 | n/a | plain = b'zipfile.py encryption test' |
---|
1606 | n/a | plain2 = b'\x00'*512 |
---|
1607 | n/a | |
---|
1608 | n/a | def setUp(self): |
---|
1609 | n/a | with open(TESTFN, "wb") as fp: |
---|
1610 | n/a | fp.write(self.data) |
---|
1611 | n/a | self.zip = zipfile.ZipFile(TESTFN, "r") |
---|
1612 | n/a | with open(TESTFN2, "wb") as fp: |
---|
1613 | n/a | fp.write(self.data2) |
---|
1614 | n/a | self.zip2 = zipfile.ZipFile(TESTFN2, "r") |
---|
1615 | n/a | |
---|
1616 | n/a | def tearDown(self): |
---|
1617 | n/a | self.zip.close() |
---|
1618 | n/a | os.unlink(TESTFN) |
---|
1619 | n/a | self.zip2.close() |
---|
1620 | n/a | os.unlink(TESTFN2) |
---|
1621 | n/a | |
---|
1622 | n/a | def test_no_password(self): |
---|
1623 | n/a | # Reading the encrypted file without password |
---|
1624 | n/a | # must generate a RunTime exception |
---|
1625 | n/a | self.assertRaises(RuntimeError, self.zip.read, "test.txt") |
---|
1626 | n/a | self.assertRaises(RuntimeError, self.zip2.read, "zero") |
---|
1627 | n/a | |
---|
1628 | n/a | def test_bad_password(self): |
---|
1629 | n/a | self.zip.setpassword(b"perl") |
---|
1630 | n/a | self.assertRaises(RuntimeError, self.zip.read, "test.txt") |
---|
1631 | n/a | self.zip2.setpassword(b"perl") |
---|
1632 | n/a | self.assertRaises(RuntimeError, self.zip2.read, "zero") |
---|
1633 | n/a | |
---|
1634 | n/a | @requires_zlib |
---|
1635 | n/a | def test_good_password(self): |
---|
1636 | n/a | self.zip.setpassword(b"python") |
---|
1637 | n/a | self.assertEqual(self.zip.read("test.txt"), self.plain) |
---|
1638 | n/a | self.zip2.setpassword(b"12345") |
---|
1639 | n/a | self.assertEqual(self.zip2.read("zero"), self.plain2) |
---|
1640 | n/a | |
---|
1641 | n/a | def test_unicode_password(self): |
---|
1642 | n/a | self.assertRaises(TypeError, self.zip.setpassword, "unicode") |
---|
1643 | n/a | self.assertRaises(TypeError, self.zip.read, "test.txt", "python") |
---|
1644 | n/a | self.assertRaises(TypeError, self.zip.open, "test.txt", pwd="python") |
---|
1645 | n/a | self.assertRaises(TypeError, self.zip.extract, "test.txt", pwd="python") |
---|
1646 | n/a | |
---|
1647 | n/a | class AbstractTestsWithRandomBinaryFiles: |
---|
1648 | n/a | @classmethod |
---|
1649 | n/a | def setUpClass(cls): |
---|
1650 | n/a | datacount = randint(16, 64)*1024 + randint(1, 1024) |
---|
1651 | n/a | cls.data = b''.join(struct.pack('<f', random()*randint(-1000, 1000)) |
---|
1652 | n/a | for i in range(datacount)) |
---|
1653 | n/a | |
---|
1654 | n/a | def setUp(self): |
---|
1655 | n/a | # Make a source file with some lines |
---|
1656 | n/a | with open(TESTFN, "wb") as fp: |
---|
1657 | n/a | fp.write(self.data) |
---|
1658 | n/a | |
---|
1659 | n/a | def tearDown(self): |
---|
1660 | n/a | unlink(TESTFN) |
---|
1661 | n/a | unlink(TESTFN2) |
---|
1662 | n/a | |
---|
1663 | n/a | def make_test_archive(self, f, compression): |
---|
1664 | n/a | # Create the ZIP archive |
---|
1665 | n/a | with zipfile.ZipFile(f, "w", compression) as zipfp: |
---|
1666 | n/a | zipfp.write(TESTFN, "another.name") |
---|
1667 | n/a | zipfp.write(TESTFN, TESTFN) |
---|
1668 | n/a | |
---|
1669 | n/a | def zip_test(self, f, compression): |
---|
1670 | n/a | self.make_test_archive(f, compression) |
---|
1671 | n/a | |
---|
1672 | n/a | # Read the ZIP archive |
---|
1673 | n/a | with zipfile.ZipFile(f, "r", compression) as zipfp: |
---|
1674 | n/a | testdata = zipfp.read(TESTFN) |
---|
1675 | n/a | self.assertEqual(len(testdata), len(self.data)) |
---|
1676 | n/a | self.assertEqual(testdata, self.data) |
---|
1677 | n/a | self.assertEqual(zipfp.read("another.name"), self.data) |
---|
1678 | n/a | |
---|
1679 | n/a | def test_read(self): |
---|
1680 | n/a | for f in get_files(self): |
---|
1681 | n/a | self.zip_test(f, self.compression) |
---|
1682 | n/a | |
---|
1683 | n/a | def zip_open_test(self, f, compression): |
---|
1684 | n/a | self.make_test_archive(f, compression) |
---|
1685 | n/a | |
---|
1686 | n/a | # Read the ZIP archive |
---|
1687 | n/a | with zipfile.ZipFile(f, "r", compression) as zipfp: |
---|
1688 | n/a | zipdata1 = [] |
---|
1689 | n/a | with zipfp.open(TESTFN) as zipopen1: |
---|
1690 | n/a | while True: |
---|
1691 | n/a | read_data = zipopen1.read(256) |
---|
1692 | n/a | if not read_data: |
---|
1693 | n/a | break |
---|
1694 | n/a | zipdata1.append(read_data) |
---|
1695 | n/a | |
---|
1696 | n/a | zipdata2 = [] |
---|
1697 | n/a | with zipfp.open("another.name") as zipopen2: |
---|
1698 | n/a | while True: |
---|
1699 | n/a | read_data = zipopen2.read(256) |
---|
1700 | n/a | if not read_data: |
---|
1701 | n/a | break |
---|
1702 | n/a | zipdata2.append(read_data) |
---|
1703 | n/a | |
---|
1704 | n/a | testdata1 = b''.join(zipdata1) |
---|
1705 | n/a | self.assertEqual(len(testdata1), len(self.data)) |
---|
1706 | n/a | self.assertEqual(testdata1, self.data) |
---|
1707 | n/a | |
---|
1708 | n/a | testdata2 = b''.join(zipdata2) |
---|
1709 | n/a | self.assertEqual(len(testdata2), len(self.data)) |
---|
1710 | n/a | self.assertEqual(testdata2, self.data) |
---|
1711 | n/a | |
---|
1712 | n/a | def test_open(self): |
---|
1713 | n/a | for f in get_files(self): |
---|
1714 | n/a | self.zip_open_test(f, self.compression) |
---|
1715 | n/a | |
---|
1716 | n/a | def zip_random_open_test(self, f, compression): |
---|
1717 | n/a | self.make_test_archive(f, compression) |
---|
1718 | n/a | |
---|
1719 | n/a | # Read the ZIP archive |
---|
1720 | n/a | with zipfile.ZipFile(f, "r", compression) as zipfp: |
---|
1721 | n/a | zipdata1 = [] |
---|
1722 | n/a | with zipfp.open(TESTFN) as zipopen1: |
---|
1723 | n/a | while True: |
---|
1724 | n/a | read_data = zipopen1.read(randint(1, 1024)) |
---|
1725 | n/a | if not read_data: |
---|
1726 | n/a | break |
---|
1727 | n/a | zipdata1.append(read_data) |
---|
1728 | n/a | |
---|
1729 | n/a | testdata = b''.join(zipdata1) |
---|
1730 | n/a | self.assertEqual(len(testdata), len(self.data)) |
---|
1731 | n/a | self.assertEqual(testdata, self.data) |
---|
1732 | n/a | |
---|
1733 | n/a | def test_random_open(self): |
---|
1734 | n/a | for f in get_files(self): |
---|
1735 | n/a | self.zip_random_open_test(f, self.compression) |
---|
1736 | n/a | |
---|
1737 | n/a | |
---|
1738 | n/a | class StoredTestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles, |
---|
1739 | n/a | unittest.TestCase): |
---|
1740 | n/a | compression = zipfile.ZIP_STORED |
---|
1741 | n/a | |
---|
1742 | n/a | @requires_zlib |
---|
1743 | n/a | class DeflateTestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles, |
---|
1744 | n/a | unittest.TestCase): |
---|
1745 | n/a | compression = zipfile.ZIP_DEFLATED |
---|
1746 | n/a | |
---|
1747 | n/a | @requires_bz2 |
---|
1748 | n/a | class Bzip2TestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles, |
---|
1749 | n/a | unittest.TestCase): |
---|
1750 | n/a | compression = zipfile.ZIP_BZIP2 |
---|
1751 | n/a | |
---|
1752 | n/a | @requires_lzma |
---|
1753 | n/a | class LzmaTestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles, |
---|
1754 | n/a | unittest.TestCase): |
---|
1755 | n/a | compression = zipfile.ZIP_LZMA |
---|
1756 | n/a | |
---|
1757 | n/a | |
---|
1758 | n/a | # Privide the tell() method but not seek() |
---|
1759 | n/a | class Tellable: |
---|
1760 | n/a | def __init__(self, fp): |
---|
1761 | n/a | self.fp = fp |
---|
1762 | n/a | self.offset = 0 |
---|
1763 | n/a | |
---|
1764 | n/a | def write(self, data): |
---|
1765 | n/a | n = self.fp.write(data) |
---|
1766 | n/a | self.offset += n |
---|
1767 | n/a | return n |
---|
1768 | n/a | |
---|
1769 | n/a | def tell(self): |
---|
1770 | n/a | return self.offset |
---|
1771 | n/a | |
---|
1772 | n/a | def flush(self): |
---|
1773 | n/a | self.fp.flush() |
---|
1774 | n/a | |
---|
1775 | n/a | class Unseekable: |
---|
1776 | n/a | def __init__(self, fp): |
---|
1777 | n/a | self.fp = fp |
---|
1778 | n/a | |
---|
1779 | n/a | def write(self, data): |
---|
1780 | n/a | return self.fp.write(data) |
---|
1781 | n/a | |
---|
1782 | n/a | def flush(self): |
---|
1783 | n/a | self.fp.flush() |
---|
1784 | n/a | |
---|
1785 | n/a | class UnseekableTests(unittest.TestCase): |
---|
1786 | n/a | def test_writestr(self): |
---|
1787 | n/a | for wrapper in (lambda f: f), Tellable, Unseekable: |
---|
1788 | n/a | with self.subTest(wrapper=wrapper): |
---|
1789 | n/a | f = io.BytesIO() |
---|
1790 | n/a | f.write(b'abc') |
---|
1791 | n/a | bf = io.BufferedWriter(f) |
---|
1792 | n/a | with zipfile.ZipFile(wrapper(bf), 'w', zipfile.ZIP_STORED) as zipfp: |
---|
1793 | n/a | zipfp.writestr('ones', b'111') |
---|
1794 | n/a | zipfp.writestr('twos', b'222') |
---|
1795 | n/a | self.assertEqual(f.getvalue()[:5], b'abcPK') |
---|
1796 | n/a | with zipfile.ZipFile(f, mode='r') as zipf: |
---|
1797 | n/a | with zipf.open('ones') as zopen: |
---|
1798 | n/a | self.assertEqual(zopen.read(), b'111') |
---|
1799 | n/a | with zipf.open('twos') as zopen: |
---|
1800 | n/a | self.assertEqual(zopen.read(), b'222') |
---|
1801 | n/a | |
---|
1802 | n/a | def test_write(self): |
---|
1803 | n/a | for wrapper in (lambda f: f), Tellable, Unseekable: |
---|
1804 | n/a | with self.subTest(wrapper=wrapper): |
---|
1805 | n/a | f = io.BytesIO() |
---|
1806 | n/a | f.write(b'abc') |
---|
1807 | n/a | bf = io.BufferedWriter(f) |
---|
1808 | n/a | with zipfile.ZipFile(wrapper(bf), 'w', zipfile.ZIP_STORED) as zipfp: |
---|
1809 | n/a | self.addCleanup(unlink, TESTFN) |
---|
1810 | n/a | with open(TESTFN, 'wb') as f2: |
---|
1811 | n/a | f2.write(b'111') |
---|
1812 | n/a | zipfp.write(TESTFN, 'ones') |
---|
1813 | n/a | with open(TESTFN, 'wb') as f2: |
---|
1814 | n/a | f2.write(b'222') |
---|
1815 | n/a | zipfp.write(TESTFN, 'twos') |
---|
1816 | n/a | self.assertEqual(f.getvalue()[:5], b'abcPK') |
---|
1817 | n/a | with zipfile.ZipFile(f, mode='r') as zipf: |
---|
1818 | n/a | with zipf.open('ones') as zopen: |
---|
1819 | n/a | self.assertEqual(zopen.read(), b'111') |
---|
1820 | n/a | with zipf.open('twos') as zopen: |
---|
1821 | n/a | self.assertEqual(zopen.read(), b'222') |
---|
1822 | n/a | |
---|
1823 | n/a | def test_open_write(self): |
---|
1824 | n/a | for wrapper in (lambda f: f), Tellable, Unseekable: |
---|
1825 | n/a | with self.subTest(wrapper=wrapper): |
---|
1826 | n/a | f = io.BytesIO() |
---|
1827 | n/a | f.write(b'abc') |
---|
1828 | n/a | bf = io.BufferedWriter(f) |
---|
1829 | n/a | with zipfile.ZipFile(wrapper(bf), 'w', zipfile.ZIP_STORED) as zipf: |
---|
1830 | n/a | with zipf.open('ones', 'w') as zopen: |
---|
1831 | n/a | zopen.write(b'111') |
---|
1832 | n/a | with zipf.open('twos', 'w') as zopen: |
---|
1833 | n/a | zopen.write(b'222') |
---|
1834 | n/a | self.assertEqual(f.getvalue()[:5], b'abcPK') |
---|
1835 | n/a | with zipfile.ZipFile(f) as zipf: |
---|
1836 | n/a | self.assertEqual(zipf.read('ones'), b'111') |
---|
1837 | n/a | self.assertEqual(zipf.read('twos'), b'222') |
---|
1838 | n/a | |
---|
1839 | n/a | |
---|
1840 | n/a | @requires_zlib |
---|
1841 | n/a | class TestsWithMultipleOpens(unittest.TestCase): |
---|
1842 | n/a | @classmethod |
---|
1843 | n/a | def setUpClass(cls): |
---|
1844 | n/a | cls.data1 = b'111' + getrandbytes(10000) |
---|
1845 | n/a | cls.data2 = b'222' + getrandbytes(10000) |
---|
1846 | n/a | |
---|
1847 | n/a | def make_test_archive(self, f): |
---|
1848 | n/a | # Create the ZIP archive |
---|
1849 | n/a | with zipfile.ZipFile(f, "w", zipfile.ZIP_DEFLATED) as zipfp: |
---|
1850 | n/a | zipfp.writestr('ones', self.data1) |
---|
1851 | n/a | zipfp.writestr('twos', self.data2) |
---|
1852 | n/a | |
---|
1853 | n/a | def test_same_file(self): |
---|
1854 | n/a | # Verify that (when the ZipFile is in control of creating file objects) |
---|
1855 | n/a | # multiple open() calls can be made without interfering with each other. |
---|
1856 | n/a | for f in get_files(self): |
---|
1857 | n/a | self.make_test_archive(f) |
---|
1858 | n/a | with zipfile.ZipFile(f, mode="r") as zipf: |
---|
1859 | n/a | with zipf.open('ones') as zopen1, zipf.open('ones') as zopen2: |
---|
1860 | n/a | data1 = zopen1.read(500) |
---|
1861 | n/a | data2 = zopen2.read(500) |
---|
1862 | n/a | data1 += zopen1.read() |
---|
1863 | n/a | data2 += zopen2.read() |
---|
1864 | n/a | self.assertEqual(data1, data2) |
---|
1865 | n/a | self.assertEqual(data1, self.data1) |
---|
1866 | n/a | |
---|
1867 | n/a | def test_different_file(self): |
---|
1868 | n/a | # Verify that (when the ZipFile is in control of creating file objects) |
---|
1869 | n/a | # multiple open() calls can be made without interfering with each other. |
---|
1870 | n/a | for f in get_files(self): |
---|
1871 | n/a | self.make_test_archive(f) |
---|
1872 | n/a | with zipfile.ZipFile(f, mode="r") as zipf: |
---|
1873 | n/a | with zipf.open('ones') as zopen1, zipf.open('twos') as zopen2: |
---|
1874 | n/a | data1 = zopen1.read(500) |
---|
1875 | n/a | data2 = zopen2.read(500) |
---|
1876 | n/a | data1 += zopen1.read() |
---|
1877 | n/a | data2 += zopen2.read() |
---|
1878 | n/a | self.assertEqual(data1, self.data1) |
---|
1879 | n/a | self.assertEqual(data2, self.data2) |
---|
1880 | n/a | |
---|
1881 | n/a | def test_interleaved(self): |
---|
1882 | n/a | # Verify that (when the ZipFile is in control of creating file objects) |
---|
1883 | n/a | # multiple open() calls can be made without interfering with each other. |
---|
1884 | n/a | for f in get_files(self): |
---|
1885 | n/a | self.make_test_archive(f) |
---|
1886 | n/a | with zipfile.ZipFile(f, mode="r") as zipf: |
---|
1887 | n/a | with zipf.open('ones') as zopen1: |
---|
1888 | n/a | data1 = zopen1.read(500) |
---|
1889 | n/a | with zipf.open('twos') as zopen2: |
---|
1890 | n/a | data2 = zopen2.read(500) |
---|
1891 | n/a | data1 += zopen1.read() |
---|
1892 | n/a | data2 += zopen2.read() |
---|
1893 | n/a | self.assertEqual(data1, self.data1) |
---|
1894 | n/a | self.assertEqual(data2, self.data2) |
---|
1895 | n/a | |
---|
1896 | n/a | def test_read_after_close(self): |
---|
1897 | n/a | for f in get_files(self): |
---|
1898 | n/a | self.make_test_archive(f) |
---|
1899 | n/a | with contextlib.ExitStack() as stack: |
---|
1900 | n/a | with zipfile.ZipFile(f, 'r') as zipf: |
---|
1901 | n/a | zopen1 = stack.enter_context(zipf.open('ones')) |
---|
1902 | n/a | zopen2 = stack.enter_context(zipf.open('twos')) |
---|
1903 | n/a | data1 = zopen1.read(500) |
---|
1904 | n/a | data2 = zopen2.read(500) |
---|
1905 | n/a | data1 += zopen1.read() |
---|
1906 | n/a | data2 += zopen2.read() |
---|
1907 | n/a | self.assertEqual(data1, self.data1) |
---|
1908 | n/a | self.assertEqual(data2, self.data2) |
---|
1909 | n/a | |
---|
1910 | n/a | def test_read_after_write(self): |
---|
1911 | n/a | for f in get_files(self): |
---|
1912 | n/a | with zipfile.ZipFile(f, 'w', zipfile.ZIP_DEFLATED) as zipf: |
---|
1913 | n/a | zipf.writestr('ones', self.data1) |
---|
1914 | n/a | zipf.writestr('twos', self.data2) |
---|
1915 | n/a | with zipf.open('ones') as zopen1: |
---|
1916 | n/a | data1 = zopen1.read(500) |
---|
1917 | n/a | self.assertEqual(data1, self.data1[:500]) |
---|
1918 | n/a | with zipfile.ZipFile(f, 'r') as zipf: |
---|
1919 | n/a | data1 = zipf.read('ones') |
---|
1920 | n/a | data2 = zipf.read('twos') |
---|
1921 | n/a | self.assertEqual(data1, self.data1) |
---|
1922 | n/a | self.assertEqual(data2, self.data2) |
---|
1923 | n/a | |
---|
1924 | n/a | def test_write_after_read(self): |
---|
1925 | n/a | for f in get_files(self): |
---|
1926 | n/a | with zipfile.ZipFile(f, "w", zipfile.ZIP_DEFLATED) as zipf: |
---|
1927 | n/a | zipf.writestr('ones', self.data1) |
---|
1928 | n/a | with zipf.open('ones') as zopen1: |
---|
1929 | n/a | zopen1.read(500) |
---|
1930 | n/a | zipf.writestr('twos', self.data2) |
---|
1931 | n/a | with zipfile.ZipFile(f, 'r') as zipf: |
---|
1932 | n/a | data1 = zipf.read('ones') |
---|
1933 | n/a | data2 = zipf.read('twos') |
---|
1934 | n/a | self.assertEqual(data1, self.data1) |
---|
1935 | n/a | self.assertEqual(data2, self.data2) |
---|
1936 | n/a | |
---|
1937 | n/a | def test_many_opens(self): |
---|
1938 | n/a | # Verify that read() and open() promptly close the file descriptor, |
---|
1939 | n/a | # and don't rely on the garbage collector to free resources. |
---|
1940 | n/a | self.make_test_archive(TESTFN2) |
---|
1941 | n/a | with zipfile.ZipFile(TESTFN2, mode="r") as zipf: |
---|
1942 | n/a | for x in range(100): |
---|
1943 | n/a | zipf.read('ones') |
---|
1944 | n/a | with zipf.open('ones') as zopen1: |
---|
1945 | n/a | pass |
---|
1946 | n/a | with open(os.devnull) as f: |
---|
1947 | n/a | self.assertLess(f.fileno(), 100) |
---|
1948 | n/a | |
---|
1949 | n/a | def test_write_while_reading(self): |
---|
1950 | n/a | with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_DEFLATED) as zipf: |
---|
1951 | n/a | zipf.writestr('ones', self.data1) |
---|
1952 | n/a | with zipfile.ZipFile(TESTFN2, 'a', zipfile.ZIP_DEFLATED) as zipf: |
---|
1953 | n/a | with zipf.open('ones', 'r') as r1: |
---|
1954 | n/a | data1 = r1.read(500) |
---|
1955 | n/a | with zipf.open('twos', 'w') as w1: |
---|
1956 | n/a | w1.write(self.data2) |
---|
1957 | n/a | data1 += r1.read() |
---|
1958 | n/a | self.assertEqual(data1, self.data1) |
---|
1959 | n/a | with zipfile.ZipFile(TESTFN2) as zipf: |
---|
1960 | n/a | self.assertEqual(zipf.read('twos'), self.data2) |
---|
1961 | n/a | |
---|
1962 | n/a | def tearDown(self): |
---|
1963 | n/a | unlink(TESTFN2) |
---|
1964 | n/a | |
---|
1965 | n/a | |
---|
1966 | n/a | class TestWithDirectory(unittest.TestCase): |
---|
1967 | n/a | def setUp(self): |
---|
1968 | n/a | os.mkdir(TESTFN2) |
---|
1969 | n/a | |
---|
1970 | n/a | def test_extract_dir(self): |
---|
1971 | n/a | with zipfile.ZipFile(findfile("zipdir.zip")) as zipf: |
---|
1972 | n/a | zipf.extractall(TESTFN2) |
---|
1973 | n/a | self.assertTrue(os.path.isdir(os.path.join(TESTFN2, "a"))) |
---|
1974 | n/a | self.assertTrue(os.path.isdir(os.path.join(TESTFN2, "a", "b"))) |
---|
1975 | n/a | self.assertTrue(os.path.exists(os.path.join(TESTFN2, "a", "b", "c"))) |
---|
1976 | n/a | |
---|
1977 | n/a | def test_bug_6050(self): |
---|
1978 | n/a | # Extraction should succeed if directories already exist |
---|
1979 | n/a | os.mkdir(os.path.join(TESTFN2, "a")) |
---|
1980 | n/a | self.test_extract_dir() |
---|
1981 | n/a | |
---|
1982 | n/a | def test_write_dir(self): |
---|
1983 | n/a | dirpath = os.path.join(TESTFN2, "x") |
---|
1984 | n/a | os.mkdir(dirpath) |
---|
1985 | n/a | mode = os.stat(dirpath).st_mode & 0xFFFF |
---|
1986 | n/a | with zipfile.ZipFile(TESTFN, "w") as zipf: |
---|
1987 | n/a | zipf.write(dirpath) |
---|
1988 | n/a | zinfo = zipf.filelist[0] |
---|
1989 | n/a | self.assertTrue(zinfo.filename.endswith("/x/")) |
---|
1990 | n/a | self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10) |
---|
1991 | n/a | zipf.write(dirpath, "y") |
---|
1992 | n/a | zinfo = zipf.filelist[1] |
---|
1993 | n/a | self.assertTrue(zinfo.filename, "y/") |
---|
1994 | n/a | self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10) |
---|
1995 | n/a | with zipfile.ZipFile(TESTFN, "r") as zipf: |
---|
1996 | n/a | zinfo = zipf.filelist[0] |
---|
1997 | n/a | self.assertTrue(zinfo.filename.endswith("/x/")) |
---|
1998 | n/a | self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10) |
---|
1999 | n/a | zinfo = zipf.filelist[1] |
---|
2000 | n/a | self.assertTrue(zinfo.filename, "y/") |
---|
2001 | n/a | self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10) |
---|
2002 | n/a | target = os.path.join(TESTFN2, "target") |
---|
2003 | n/a | os.mkdir(target) |
---|
2004 | n/a | zipf.extractall(target) |
---|
2005 | n/a | self.assertTrue(os.path.isdir(os.path.join(target, "y"))) |
---|
2006 | n/a | self.assertEqual(len(os.listdir(target)), 2) |
---|
2007 | n/a | |
---|
2008 | n/a | def test_writestr_dir(self): |
---|
2009 | n/a | os.mkdir(os.path.join(TESTFN2, "x")) |
---|
2010 | n/a | with zipfile.ZipFile(TESTFN, "w") as zipf: |
---|
2011 | n/a | zipf.writestr("x/", b'') |
---|
2012 | n/a | zinfo = zipf.filelist[0] |
---|
2013 | n/a | self.assertEqual(zinfo.filename, "x/") |
---|
2014 | n/a | self.assertEqual(zinfo.external_attr, (0o40775 << 16) | 0x10) |
---|
2015 | n/a | with zipfile.ZipFile(TESTFN, "r") as zipf: |
---|
2016 | n/a | zinfo = zipf.filelist[0] |
---|
2017 | n/a | self.assertTrue(zinfo.filename.endswith("x/")) |
---|
2018 | n/a | self.assertEqual(zinfo.external_attr, (0o40775 << 16) | 0x10) |
---|
2019 | n/a | target = os.path.join(TESTFN2, "target") |
---|
2020 | n/a | os.mkdir(target) |
---|
2021 | n/a | zipf.extractall(target) |
---|
2022 | n/a | self.assertTrue(os.path.isdir(os.path.join(target, "x"))) |
---|
2023 | n/a | self.assertEqual(os.listdir(target), ["x"]) |
---|
2024 | n/a | |
---|
2025 | n/a | def tearDown(self): |
---|
2026 | n/a | rmtree(TESTFN2) |
---|
2027 | n/a | if os.path.exists(TESTFN): |
---|
2028 | n/a | unlink(TESTFN) |
---|
2029 | n/a | |
---|
2030 | n/a | |
---|
2031 | n/a | class ZipInfoTests(unittest.TestCase): |
---|
2032 | n/a | def test_from_file(self): |
---|
2033 | n/a | zi = zipfile.ZipInfo.from_file(__file__) |
---|
2034 | n/a | self.assertEqual(posixpath.basename(zi.filename), 'test_zipfile.py') |
---|
2035 | n/a | self.assertFalse(zi.is_dir()) |
---|
2036 | n/a | |
---|
2037 | n/a | def test_from_dir(self): |
---|
2038 | n/a | dirpath = os.path.dirname(os.path.abspath(__file__)) |
---|
2039 | n/a | zi = zipfile.ZipInfo.from_file(dirpath, 'stdlib_tests') |
---|
2040 | n/a | self.assertEqual(zi.filename, 'stdlib_tests/') |
---|
2041 | n/a | self.assertTrue(zi.is_dir()) |
---|
2042 | n/a | self.assertEqual(zi.compress_type, zipfile.ZIP_STORED) |
---|
2043 | n/a | self.assertEqual(zi.file_size, 0) |
---|
2044 | n/a | |
---|
2045 | n/a | |
---|
2046 | n/a | class CommandLineTest(unittest.TestCase): |
---|
2047 | n/a | |
---|
2048 | n/a | def zipfilecmd(self, *args, **kwargs): |
---|
2049 | n/a | rc, out, err = script_helper.assert_python_ok('-m', 'zipfile', *args, |
---|
2050 | n/a | **kwargs) |
---|
2051 | n/a | return out.replace(os.linesep.encode(), b'\n') |
---|
2052 | n/a | |
---|
2053 | n/a | def zipfilecmd_failure(self, *args): |
---|
2054 | n/a | return script_helper.assert_python_failure('-m', 'zipfile', *args) |
---|
2055 | n/a | |
---|
2056 | n/a | def test_test_command(self): |
---|
2057 | n/a | zip_name = findfile('zipdir.zip') |
---|
2058 | n/a | for opt in '-t', '--test': |
---|
2059 | n/a | out = self.zipfilecmd(opt, zip_name) |
---|
2060 | n/a | self.assertEqual(out.rstrip(), b'Done testing') |
---|
2061 | n/a | zip_name = findfile('testtar.tar') |
---|
2062 | n/a | rc, out, err = self.zipfilecmd_failure('-t', zip_name) |
---|
2063 | n/a | self.assertEqual(out, b'') |
---|
2064 | n/a | |
---|
2065 | n/a | def test_list_command(self): |
---|
2066 | n/a | zip_name = findfile('zipdir.zip') |
---|
2067 | n/a | t = io.StringIO() |
---|
2068 | n/a | with zipfile.ZipFile(zip_name, 'r') as tf: |
---|
2069 | n/a | tf.printdir(t) |
---|
2070 | n/a | expected = t.getvalue().encode('ascii', 'backslashreplace') |
---|
2071 | n/a | for opt in '-l', '--list': |
---|
2072 | n/a | out = self.zipfilecmd(opt, zip_name, |
---|
2073 | n/a | PYTHONIOENCODING='ascii:backslashreplace') |
---|
2074 | n/a | self.assertEqual(out, expected) |
---|
2075 | n/a | |
---|
2076 | n/a | @requires_zlib |
---|
2077 | n/a | def test_create_command(self): |
---|
2078 | n/a | self.addCleanup(unlink, TESTFN) |
---|
2079 | n/a | with open(TESTFN, 'w') as f: |
---|
2080 | n/a | f.write('test 1') |
---|
2081 | n/a | os.mkdir(TESTFNDIR) |
---|
2082 | n/a | self.addCleanup(rmtree, TESTFNDIR) |
---|
2083 | n/a | with open(os.path.join(TESTFNDIR, 'file.txt'), 'w') as f: |
---|
2084 | n/a | f.write('test 2') |
---|
2085 | n/a | files = [TESTFN, TESTFNDIR] |
---|
2086 | n/a | namelist = [TESTFN, TESTFNDIR + '/', TESTFNDIR + '/file.txt'] |
---|
2087 | n/a | for opt in '-c', '--create': |
---|
2088 | n/a | try: |
---|
2089 | n/a | out = self.zipfilecmd(opt, TESTFN2, *files) |
---|
2090 | n/a | self.assertEqual(out, b'') |
---|
2091 | n/a | with zipfile.ZipFile(TESTFN2) as zf: |
---|
2092 | n/a | self.assertEqual(zf.namelist(), namelist) |
---|
2093 | n/a | self.assertEqual(zf.read(namelist[0]), b'test 1') |
---|
2094 | n/a | self.assertEqual(zf.read(namelist[2]), b'test 2') |
---|
2095 | n/a | finally: |
---|
2096 | n/a | unlink(TESTFN2) |
---|
2097 | n/a | |
---|
2098 | n/a | def test_extract_command(self): |
---|
2099 | n/a | zip_name = findfile('zipdir.zip') |
---|
2100 | n/a | for opt in '-e', '--extract': |
---|
2101 | n/a | with temp_dir() as extdir: |
---|
2102 | n/a | out = self.zipfilecmd(opt, zip_name, extdir) |
---|
2103 | n/a | self.assertEqual(out, b'') |
---|
2104 | n/a | with zipfile.ZipFile(zip_name) as zf: |
---|
2105 | n/a | for zi in zf.infolist(): |
---|
2106 | n/a | path = os.path.join(extdir, |
---|
2107 | n/a | zi.filename.replace('/', os.sep)) |
---|
2108 | n/a | if zi.is_dir(): |
---|
2109 | n/a | self.assertTrue(os.path.isdir(path)) |
---|
2110 | n/a | else: |
---|
2111 | n/a | self.assertTrue(os.path.isfile(path)) |
---|
2112 | n/a | with open(path, 'rb') as f: |
---|
2113 | n/a | self.assertEqual(f.read(), zf.read(zi)) |
---|
2114 | n/a | |
---|
2115 | n/a | if __name__ == "__main__": |
---|
2116 | n/a | unittest.main() |
---|