1 import io
2 import itertools
3 import contextlib
4 import pathlib
5 import pickle
6 import sys
7 import unittest
8 import zipfile
9
10 from ._functools import compose
11 from ._itertools import Counter
12
13 from ._test_params import parameterize, Invoked
14
15 from test.support.os_helper import temp_dir
16
17
18 class ESC[4;38;5;81mjaraco:
19 class ESC[4;38;5;81mitertools:
20 Counter = Counter
21
22
23 def add_dirs(zf):
24 """
25 Given a writable zip file zf, inject directory entries for
26 any directories implied by the presence of children.
27 """
28 for name in zipfile.CompleteDirs._implied_dirs(zf.namelist()):
29 zf.writestr(name, b"")
30 return zf
31
32
33 def build_alpharep_fixture():
34 """
35 Create a zip file with this structure:
36
37 .
38 ├── a.txt
39 ├── b
40 │ ├── c.txt
41 │ ├── d
42 │ │ └── e.txt
43 │ └── f.txt
44 ├── g
45 │ └── h
46 │ └── i.txt
47 └── j
48 ├── k.bin
49 ├── l.baz
50 └── m.bar
51
52 This fixture has the following key characteristics:
53
54 - a file at the root (a)
55 - a file two levels deep (b/d/e)
56 - multiple files in a directory (b/c, b/f)
57 - a directory containing only a directory (g/h)
58 - a directory with files of different extensions (j/klm)
59
60 "alpha" because it uses alphabet
61 "rep" because it's a representative example
62 """
63 data = io.BytesIO()
64 zf = zipfile.ZipFile(data, "w")
65 zf.writestr("a.txt", b"content of a")
66 zf.writestr("b/c.txt", b"content of c")
67 zf.writestr("b/d/e.txt", b"content of e")
68 zf.writestr("b/f.txt", b"content of f")
69 zf.writestr("g/h/i.txt", b"content of i")
70 zf.writestr("j/k.bin", b"content of k")
71 zf.writestr("j/l.baz", b"content of l")
72 zf.writestr("j/m.bar", b"content of m")
73 zf.filename = "alpharep.zip"
74 return zf
75
76
77 alpharep_generators = [
78 Invoked.wrap(build_alpharep_fixture),
79 Invoked.wrap(compose(add_dirs, build_alpharep_fixture)),
80 ]
81
82 pass_alpharep = parameterize(['alpharep'], alpharep_generators)
83
84
85 class ESC[4;38;5;81mTestPath(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
86 def setUp(self):
87 self.fixtures = contextlib.ExitStack()
88 self.addCleanup(self.fixtures.close)
89
90 def zipfile_ondisk(self, alpharep):
91 tmpdir = pathlib.Path(self.fixtures.enter_context(temp_dir()))
92 buffer = alpharep.fp
93 alpharep.close()
94 path = tmpdir / alpharep.filename
95 with path.open("wb") as strm:
96 strm.write(buffer.getvalue())
97 return path
98
99 @pass_alpharep
100 def test_iterdir_and_types(self, alpharep):
101 root = zipfile.Path(alpharep)
102 assert root.is_dir()
103 a, b, g, j = root.iterdir()
104 assert a.is_file()
105 assert b.is_dir()
106 assert g.is_dir()
107 c, f, d = b.iterdir()
108 assert c.is_file() and f.is_file()
109 (e,) = d.iterdir()
110 assert e.is_file()
111 (h,) = g.iterdir()
112 (i,) = h.iterdir()
113 assert i.is_file()
114
115 @pass_alpharep
116 def test_is_file_missing(self, alpharep):
117 root = zipfile.Path(alpharep)
118 assert not root.joinpath('missing.txt').is_file()
119
120 @pass_alpharep
121 def test_iterdir_on_file(self, alpharep):
122 root = zipfile.Path(alpharep)
123 a, b, g, j = root.iterdir()
124 with self.assertRaises(ValueError):
125 a.iterdir()
126
127 @pass_alpharep
128 def test_subdir_is_dir(self, alpharep):
129 root = zipfile.Path(alpharep)
130 assert (root / 'b').is_dir()
131 assert (root / 'b/').is_dir()
132 assert (root / 'g').is_dir()
133 assert (root / 'g/').is_dir()
134
135 @pass_alpharep
136 def test_open(self, alpharep):
137 root = zipfile.Path(alpharep)
138 a, b, g, j = root.iterdir()
139 with a.open(encoding="utf-8") as strm:
140 data = strm.read()
141 self.assertEqual(data, "content of a")
142 with a.open('r', "utf-8") as strm: # not a kw, no gh-101144 TypeError
143 data = strm.read()
144 self.assertEqual(data, "content of a")
145
146 def test_open_encoding_utf16(self):
147 in_memory_file = io.BytesIO()
148 zf = zipfile.ZipFile(in_memory_file, "w")
149 zf.writestr("path/16.txt", "This was utf-16".encode("utf-16"))
150 zf.filename = "test_open_utf16.zip"
151 root = zipfile.Path(zf)
152 (path,) = root.iterdir()
153 u16 = path.joinpath("16.txt")
154 with u16.open('r', "utf-16") as strm:
155 data = strm.read()
156 assert data == "This was utf-16"
157 with u16.open(encoding="utf-16") as strm:
158 data = strm.read()
159 assert data == "This was utf-16"
160
161 def test_open_encoding_errors(self):
162 in_memory_file = io.BytesIO()
163 zf = zipfile.ZipFile(in_memory_file, "w")
164 zf.writestr("path/bad-utf8.bin", b"invalid utf-8: \xff\xff.")
165 zf.filename = "test_read_text_encoding_errors.zip"
166 root = zipfile.Path(zf)
167 (path,) = root.iterdir()
168 u16 = path.joinpath("bad-utf8.bin")
169
170 # encoding= as a positional argument for gh-101144.
171 data = u16.read_text("utf-8", errors="ignore")
172 assert data == "invalid utf-8: ."
173 with u16.open("r", "utf-8", errors="surrogateescape") as f:
174 assert f.read() == "invalid utf-8: \udcff\udcff."
175
176 # encoding= both positional and keyword is an error; gh-101144.
177 with self.assertRaisesRegex(TypeError, "encoding"):
178 data = u16.read_text("utf-8", encoding="utf-8")
179
180 # both keyword arguments work.
181 with u16.open("r", encoding="utf-8", errors="strict") as f:
182 # error during decoding with wrong codec.
183 with self.assertRaises(UnicodeDecodeError):
184 f.read()
185
186 @unittest.skipIf(
187 not getattr(sys.flags, 'warn_default_encoding', 0),
188 "Requires warn_default_encoding",
189 )
190 @pass_alpharep
191 def test_encoding_warnings(self, alpharep):
192 """EncodingWarning must blame the read_text and open calls."""
193 assert sys.flags.warn_default_encoding
194 root = zipfile.Path(alpharep)
195 with self.assertWarns(EncodingWarning) as wc:
196 root.joinpath("a.txt").read_text()
197 assert __file__ == wc.filename
198 with self.assertWarns(EncodingWarning) as wc:
199 root.joinpath("a.txt").open("r").close()
200 assert __file__ == wc.filename
201
202 def test_open_write(self):
203 """
204 If the zipfile is open for write, it should be possible to
205 write bytes or text to it.
206 """
207 zf = zipfile.Path(zipfile.ZipFile(io.BytesIO(), mode='w'))
208 with zf.joinpath('file.bin').open('wb') as strm:
209 strm.write(b'binary contents')
210 with zf.joinpath('file.txt').open('w', encoding="utf-8") as strm:
211 strm.write('text file')
212
213 def test_open_extant_directory(self):
214 """
215 Attempting to open a directory raises IsADirectoryError.
216 """
217 zf = zipfile.Path(add_dirs(build_alpharep_fixture()))
218 with self.assertRaises(IsADirectoryError):
219 zf.joinpath('b').open()
220
221 @pass_alpharep
222 def test_open_binary_invalid_args(self, alpharep):
223 root = zipfile.Path(alpharep)
224 with self.assertRaises(ValueError):
225 root.joinpath('a.txt').open('rb', encoding='utf-8')
226 with self.assertRaises(ValueError):
227 root.joinpath('a.txt').open('rb', 'utf-8')
228
229 def test_open_missing_directory(self):
230 """
231 Attempting to open a missing directory raises FileNotFoundError.
232 """
233 zf = zipfile.Path(add_dirs(build_alpharep_fixture()))
234 with self.assertRaises(FileNotFoundError):
235 zf.joinpath('z').open()
236
237 @pass_alpharep
238 def test_read(self, alpharep):
239 root = zipfile.Path(alpharep)
240 a, b, g, j = root.iterdir()
241 assert a.read_text(encoding="utf-8") == "content of a"
242 # Also check positional encoding arg (gh-101144).
243 assert a.read_text("utf-8") == "content of a"
244 assert a.read_bytes() == b"content of a"
245
246 @pass_alpharep
247 def test_joinpath(self, alpharep):
248 root = zipfile.Path(alpharep)
249 a = root.joinpath("a.txt")
250 assert a.is_file()
251 e = root.joinpath("b").joinpath("d").joinpath("e.txt")
252 assert e.read_text(encoding="utf-8") == "content of e"
253
254 @pass_alpharep
255 def test_joinpath_multiple(self, alpharep):
256 root = zipfile.Path(alpharep)
257 e = root.joinpath("b", "d", "e.txt")
258 assert e.read_text(encoding="utf-8") == "content of e"
259
260 @pass_alpharep
261 def test_traverse_truediv(self, alpharep):
262 root = zipfile.Path(alpharep)
263 a = root / "a.txt"
264 assert a.is_file()
265 e = root / "b" / "d" / "e.txt"
266 assert e.read_text(encoding="utf-8") == "content of e"
267
268 @pass_alpharep
269 def test_pathlike_construction(self, alpharep):
270 """
271 zipfile.Path should be constructable from a path-like object
272 """
273 zipfile_ondisk = self.zipfile_ondisk(alpharep)
274 pathlike = pathlib.Path(str(zipfile_ondisk))
275 zipfile.Path(pathlike)
276
277 @pass_alpharep
278 def test_traverse_pathlike(self, alpharep):
279 root = zipfile.Path(alpharep)
280 root / pathlib.Path("a")
281
282 @pass_alpharep
283 def test_parent(self, alpharep):
284 root = zipfile.Path(alpharep)
285 assert (root / 'a').parent.at == ''
286 assert (root / 'a' / 'b').parent.at == 'a/'
287
288 @pass_alpharep
289 def test_dir_parent(self, alpharep):
290 root = zipfile.Path(alpharep)
291 assert (root / 'b').parent.at == ''
292 assert (root / 'b/').parent.at == ''
293
294 @pass_alpharep
295 def test_missing_dir_parent(self, alpharep):
296 root = zipfile.Path(alpharep)
297 assert (root / 'missing dir/').parent.at == ''
298
299 @pass_alpharep
300 def test_mutability(self, alpharep):
301 """
302 If the underlying zipfile is changed, the Path object should
303 reflect that change.
304 """
305 root = zipfile.Path(alpharep)
306 a, b, g, j = root.iterdir()
307 alpharep.writestr('foo.txt', 'foo')
308 alpharep.writestr('bar/baz.txt', 'baz')
309 assert any(child.name == 'foo.txt' for child in root.iterdir())
310 assert (root / 'foo.txt').read_text(encoding="utf-8") == 'foo'
311 (baz,) = (root / 'bar').iterdir()
312 assert baz.read_text(encoding="utf-8") == 'baz'
313
314 HUGE_ZIPFILE_NUM_ENTRIES = 2**13
315
316 def huge_zipfile(self):
317 """Create a read-only zipfile with a huge number of entries entries."""
318 strm = io.BytesIO()
319 zf = zipfile.ZipFile(strm, "w")
320 for entry in map(str, range(self.HUGE_ZIPFILE_NUM_ENTRIES)):
321 zf.writestr(entry, entry)
322 zf.mode = 'r'
323 return zf
324
325 def test_joinpath_constant_time(self):
326 """
327 Ensure joinpath on items in zipfile is linear time.
328 """
329 root = zipfile.Path(self.huge_zipfile())
330 entries = jaraco.itertools.Counter(root.iterdir())
331 for entry in entries:
332 entry.joinpath('suffix')
333 # Check the file iterated all items
334 assert entries.count == self.HUGE_ZIPFILE_NUM_ENTRIES
335
336 @pass_alpharep
337 def test_read_does_not_close(self, alpharep):
338 alpharep = self.zipfile_ondisk(alpharep)
339 with zipfile.ZipFile(alpharep) as file:
340 for rep in range(2):
341 zipfile.Path(file, 'a.txt').read_text(encoding="utf-8")
342
343 @pass_alpharep
344 def test_subclass(self, alpharep):
345 class ESC[4;38;5;81mSubclass(ESC[4;38;5;149mzipfileESC[4;38;5;149m.ESC[4;38;5;149mPath):
346 pass
347
348 root = Subclass(alpharep)
349 assert isinstance(root / 'b', Subclass)
350
351 @pass_alpharep
352 def test_filename(self, alpharep):
353 root = zipfile.Path(alpharep)
354 assert root.filename == pathlib.Path('alpharep.zip')
355
356 @pass_alpharep
357 def test_root_name(self, alpharep):
358 """
359 The name of the root should be the name of the zipfile
360 """
361 root = zipfile.Path(alpharep)
362 assert root.name == 'alpharep.zip' == root.filename.name
363
364 @pass_alpharep
365 def test_suffix(self, alpharep):
366 """
367 The suffix of the root should be the suffix of the zipfile.
368 The suffix of each nested file is the final component's last suffix, if any.
369 Includes the leading period, just like pathlib.Path.
370 """
371 root = zipfile.Path(alpharep)
372 assert root.suffix == '.zip' == root.filename.suffix
373
374 b = root / "b.txt"
375 assert b.suffix == ".txt"
376
377 c = root / "c" / "filename.tar.gz"
378 assert c.suffix == ".gz"
379
380 d = root / "d"
381 assert d.suffix == ""
382
383 @pass_alpharep
384 def test_suffixes(self, alpharep):
385 """
386 The suffix of the root should be the suffix of the zipfile.
387 The suffix of each nested file is the final component's last suffix, if any.
388 Includes the leading period, just like pathlib.Path.
389 """
390 root = zipfile.Path(alpharep)
391 assert root.suffixes == ['.zip'] == root.filename.suffixes
392
393 b = root / 'b.txt'
394 assert b.suffixes == ['.txt']
395
396 c = root / 'c' / 'filename.tar.gz'
397 assert c.suffixes == ['.tar', '.gz']
398
399 d = root / 'd'
400 assert d.suffixes == []
401
402 e = root / '.hgrc'
403 assert e.suffixes == []
404
405 @pass_alpharep
406 def test_suffix_no_filename(self, alpharep):
407 alpharep.filename = None
408 root = zipfile.Path(alpharep)
409 assert root.joinpath('example').suffix == ""
410 assert root.joinpath('example').suffixes == []
411
412 @pass_alpharep
413 def test_stem(self, alpharep):
414 """
415 The final path component, without its suffix
416 """
417 root = zipfile.Path(alpharep)
418 assert root.stem == 'alpharep' == root.filename.stem
419
420 b = root / "b.txt"
421 assert b.stem == "b"
422
423 c = root / "c" / "filename.tar.gz"
424 assert c.stem == "filename.tar"
425
426 d = root / "d"
427 assert d.stem == "d"
428
429 assert (root / ".gitignore").stem == ".gitignore"
430
431 @pass_alpharep
432 def test_root_parent(self, alpharep):
433 root = zipfile.Path(alpharep)
434 assert root.parent == pathlib.Path('.')
435 root.root.filename = 'foo/bar.zip'
436 assert root.parent == pathlib.Path('foo')
437
438 @pass_alpharep
439 def test_root_unnamed(self, alpharep):
440 """
441 It is an error to attempt to get the name
442 or parent of an unnamed zipfile.
443 """
444 alpharep.filename = None
445 root = zipfile.Path(alpharep)
446 with self.assertRaises(TypeError):
447 root.name
448 with self.assertRaises(TypeError):
449 root.parent
450
451 # .name and .parent should still work on subs
452 sub = root / "b"
453 assert sub.name == "b"
454 assert sub.parent
455
456 @pass_alpharep
457 def test_match_and_glob(self, alpharep):
458 root = zipfile.Path(alpharep)
459 assert not root.match("*.txt")
460
461 assert list(root.glob("b/c.*")) == [zipfile.Path(alpharep, "b/c.txt")]
462 assert list(root.glob("b/*.txt")) == [
463 zipfile.Path(alpharep, "b/c.txt"),
464 zipfile.Path(alpharep, "b/f.txt"),
465 ]
466
467 @pass_alpharep
468 def test_glob_recursive(self, alpharep):
469 root = zipfile.Path(alpharep)
470 files = root.glob("**/*.txt")
471 assert all(each.match("*.txt") for each in files)
472
473 assert list(root.glob("**/*.txt")) == list(root.rglob("*.txt"))
474
475 @pass_alpharep
476 def test_glob_subdirs(self, alpharep):
477 root = zipfile.Path(alpharep)
478
479 assert list(root.glob("*/i.txt")) == []
480 assert list(root.rglob("*/i.txt")) == [zipfile.Path(alpharep, "g/h/i.txt")]
481
482 @pass_alpharep
483 def test_glob_does_not_overmatch_dot(self, alpharep):
484 root = zipfile.Path(alpharep)
485
486 assert list(root.glob("*.xt")) == []
487
488 @pass_alpharep
489 def test_glob_single_char(self, alpharep):
490 root = zipfile.Path(alpharep)
491
492 assert list(root.glob("a?txt")) == [zipfile.Path(alpharep, "a.txt")]
493 assert list(root.glob("a[.]txt")) == [zipfile.Path(alpharep, "a.txt")]
494 assert list(root.glob("a[?]txt")) == []
495
496 @pass_alpharep
497 def test_glob_chars(self, alpharep):
498 root = zipfile.Path(alpharep)
499
500 assert list(root.glob("j/?.b[ai][nz]")) == [
501 zipfile.Path(alpharep, "j/k.bin"),
502 zipfile.Path(alpharep, "j/l.baz"),
503 ]
504
505 def test_glob_empty(self):
506 root = zipfile.Path(zipfile.ZipFile(io.BytesIO(), 'w'))
507 with self.assertRaises(ValueError):
508 root.glob('')
509
510 @pass_alpharep
511 def test_eq_hash(self, alpharep):
512 root = zipfile.Path(alpharep)
513 assert root == zipfile.Path(alpharep)
514
515 assert root != (root / "a.txt")
516 assert (root / "a.txt") == (root / "a.txt")
517
518 root = zipfile.Path(alpharep)
519 assert root in {root}
520
521 @pass_alpharep
522 def test_is_symlink(self, alpharep):
523 """
524 See python/cpython#82102 for symlink support beyond this object.
525 """
526
527 root = zipfile.Path(alpharep)
528 assert not root.is_symlink()
529
530 @pass_alpharep
531 def test_relative_to(self, alpharep):
532 root = zipfile.Path(alpharep)
533 relative = root.joinpath("b", "c.txt").relative_to(root / "b")
534 assert str(relative) == "c.txt"
535
536 relative = root.joinpath("b", "d", "e.txt").relative_to(root / "b")
537 assert str(relative) == "d/e.txt"
538
539 @pass_alpharep
540 def test_inheritance(self, alpharep):
541 cls = type('PathChild', (zipfile.Path,), {})
542 file = cls(alpharep).joinpath('some dir').parent
543 assert isinstance(file, cls)
544
545 @parameterize(
546 ['alpharep', 'path_type', 'subpath'],
547 itertools.product(
548 alpharep_generators,
549 [str, pathlib.Path],
550 ['', 'b/'],
551 ),
552 )
553 def test_pickle(self, alpharep, path_type, subpath):
554 zipfile_ondisk = path_type(self.zipfile_ondisk(alpharep))
555
556 saved_1 = pickle.dumps(zipfile.Path(zipfile_ondisk, at=subpath))
557 restored_1 = pickle.loads(saved_1)
558 first, *rest = restored_1.iterdir()
559 assert first.read_text(encoding='utf-8').startswith('content of ')
560
561 @pass_alpharep
562 def test_extract_orig_with_implied_dirs(self, alpharep):
563 """
564 A zip file wrapped in a Path should extract even with implied dirs.
565 """
566 source_path = self.zipfile_ondisk(alpharep)
567 zf = zipfile.ZipFile(source_path)
568 # wrap the zipfile for its side effect
569 zipfile.Path(zf)
570 zf.extractall(source_path.parent)
571
572 @pass_alpharep
573 def test_getinfo_missing(self, alpharep):
574 """
575 Validate behavior of getinfo on original zipfile after wrapping.
576 """
577 zipfile.Path(alpharep)
578 with self.assertRaises(KeyError):
579 alpharep.getinfo('does-not-exist')