Skip to content

Commit 5e1fc93

Browse files
jackoconnordevJack O'Connor
jackoconnordev
and
Jack O'Connor
authored
Update zipfile to 3.13.5 (#6069)
* Add Lib/test/archiver_tests.py @ 3.13.5 Needed for updated zipfile tests. * Update zipfile to 3.13.5 Notes: - I have to skip some brand new tests due to shift_jis encoding not being supported - `test_write_filtered_python_package` marked as `expectedFailure` with "AttributeError: module 'os' has no attribute 'supports_effective_ids'" - I didn't want to do a partial or full update to os module in this PR --------- Co-authored-by: Jack O'Connor <jack@jackoconnor.dev>
1 parent 1464d5c commit 5e1fc93

File tree

16 files changed

+2890
-827
lines changed

16 files changed

+2890
-827
lines changed

‎Lib/test/archiver_tests.py‎

Lines changed: 177 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,177 @@
1+
"""Tests common to tarfile and zipfile."""
2+
3+
import os
4+
import sys
5+
6+
from test.support import swap_attr
7+
from test.support import os_helper
8+
9+
class OverwriteTests:
10+
11+
def setUp(self):
12+
os.makedirs(self.testdir)
13+
self.addCleanup(os_helper.rmtree, self.testdir)
14+
15+
def create_file(self, path, content=b''):
16+
with open(path, 'wb') as f:
17+
f.write(content)
18+
19+
def open(self, path):
20+
raise NotImplementedError
21+
22+
def extractall(self, ar):
23+
raise NotImplementedError
24+
25+
26+
def test_overwrite_file_as_file(self):
27+
target = os.path.join(self.testdir, 'test')
28+
self.create_file(target, b'content')
29+
with self.open(self.ar_with_file) as ar:
30+
self.extractall(ar)
31+
self.assertTrue(os.path.isfile(target))
32+
with open(target, 'rb') as f:
33+
self.assertEqual(f.read(), b'newcontent')
34+
35+
def test_overwrite_dir_as_dir(self):
36+
target = os.path.join(self.testdir, 'test')
37+
os.mkdir(target)
38+
with self.open(self.ar_with_dir) as ar:
39+
self.extractall(ar)
40+
self.assertTrue(os.path.isdir(target))
41+
42+
def test_overwrite_dir_as_implicit_dir(self):
43+
target = os.path.join(self.testdir, 'test')
44+
os.mkdir(target)
45+
with self.open(self.ar_with_implicit_dir) as ar:
46+
self.extractall(ar)
47+
self.assertTrue(os.path.isdir(target))
48+
self.assertTrue(os.path.isfile(os.path.join(target, 'file')))
49+
with open(os.path.join(target, 'file'), 'rb') as f:
50+
self.assertEqual(f.read(), b'newcontent')
51+
52+
def test_overwrite_dir_as_file(self):
53+
target = os.path.join(self.testdir, 'test')
54+
os.mkdir(target)
55+
with self.open(self.ar_with_file) as ar:
56+
with self.assertRaises(PermissionError if sys.platform == 'win32'
57+
else IsADirectoryError):
58+
self.extractall(ar)
59+
self.assertTrue(os.path.isdir(target))
60+
61+
def test_overwrite_file_as_dir(self):
62+
target = os.path.join(self.testdir, 'test')
63+
self.create_file(target, b'content')
64+
with self.open(self.ar_with_dir) as ar:
65+
with self.assertRaises(FileExistsError):
66+
self.extractall(ar)
67+
self.assertTrue(os.path.isfile(target))
68+
with open(target, 'rb') as f:
69+
self.assertEqual(f.read(), b'content')
70+
71+
def test_overwrite_file_as_implicit_dir(self):
72+
target = os.path.join(self.testdir, 'test')
73+
self.create_file(target, b'content')
74+
with self.open(self.ar_with_implicit_dir) as ar:
75+
with self.assertRaises(FileNotFoundError if sys.platform == 'win32'
76+
else NotADirectoryError):
77+
self.extractall(ar)
78+
self.assertTrue(os.path.isfile(target))
79+
with open(target, 'rb') as f:
80+
self.assertEqual(f.read(), b'content')
81+
82+
@os_helper.skip_unless_symlink
83+
def test_overwrite_file_symlink_as_file(self):
84+
# XXX: It is potential security vulnerability.
85+
target = os.path.join(self.testdir, 'test')
86+
target2 = os.path.join(self.testdir, 'test2')
87+
self.create_file(target2, b'content')
88+
os.symlink('test2', target)
89+
with self.open(self.ar_with_file) as ar:
90+
self.extractall(ar)
91+
self.assertTrue(os.path.islink(target))
92+
self.assertTrue(os.path.isfile(target2))
93+
with open(target2, 'rb') as f:
94+
self.assertEqual(f.read(), b'newcontent')
95+
96+
@os_helper.skip_unless_symlink
97+
def test_overwrite_broken_file_symlink_as_file(self):
98+
# XXX: It is potential security vulnerability.
99+
target = os.path.join(self.testdir, 'test')
100+
target2 = os.path.join(self.testdir, 'test2')
101+
os.symlink('test2', target)
102+
with self.open(self.ar_with_file) as ar:
103+
self.extractall(ar)
104+
self.assertTrue(os.path.islink(target))
105+
self.assertTrue(os.path.isfile(target2))
106+
with open(target2, 'rb') as f:
107+
self.assertEqual(f.read(), b'newcontent')
108+
109+
@os_helper.skip_unless_symlink
110+
def test_overwrite_dir_symlink_as_dir(self):
111+
# XXX: It is potential security vulnerability.
112+
target = os.path.join(self.testdir, 'test')
113+
target2 = os.path.join(self.testdir, 'test2')
114+
os.mkdir(target2)
115+
os.symlink('test2', target, target_is_directory=True)
116+
with self.open(self.ar_with_dir) as ar:
117+
self.extractall(ar)
118+
self.assertTrue(os.path.islink(target))
119+
self.assertTrue(os.path.isdir(target2))
120+
121+
@os_helper.skip_unless_symlink
122+
def test_overwrite_dir_symlink_as_implicit_dir(self):
123+
# XXX: It is potential security vulnerability.
124+
target = os.path.join(self.testdir, 'test')
125+
target2 = os.path.join(self.testdir, 'test2')
126+
os.mkdir(target2)
127+
os.symlink('test2', target, target_is_directory=True)
128+
with self.open(self.ar_with_implicit_dir) as ar:
129+
self.extractall(ar)
130+
self.assertTrue(os.path.islink(target))
131+
self.assertTrue(os.path.isdir(target2))
132+
self.assertTrue(os.path.isfile(os.path.join(target2, 'file')))
133+
with open(os.path.join(target2, 'file'), 'rb') as f:
134+
self.assertEqual(f.read(), b'newcontent')
135+
136+
@os_helper.skip_unless_symlink
137+
def test_overwrite_broken_dir_symlink_as_dir(self):
138+
target = os.path.join(self.testdir, 'test')
139+
target2 = os.path.join(self.testdir, 'test2')
140+
os.symlink('test2', target, target_is_directory=True)
141+
with self.open(self.ar_with_dir) as ar:
142+
with self.assertRaises(FileExistsError):
143+
self.extractall(ar)
144+
self.assertTrue(os.path.islink(target))
145+
self.assertFalse(os.path.exists(target2))
146+
147+
@os_helper.skip_unless_symlink
148+
def test_overwrite_broken_dir_symlink_as_implicit_dir(self):
149+
target = os.path.join(self.testdir, 'test')
150+
target2 = os.path.join(self.testdir, 'test2')
151+
os.symlink('test2', target, target_is_directory=True)
152+
with self.open(self.ar_with_implicit_dir) as ar:
153+
with self.assertRaises(FileExistsError):
154+
self.extractall(ar)
155+
self.assertTrue(os.path.islink(target))
156+
self.assertFalse(os.path.exists(target2))
157+
158+
def test_concurrent_extract_dir(self):
159+
target = os.path.join(self.testdir, 'test')
160+
def concurrent_mkdir(*args, **kwargs):
161+
orig_mkdir(*args, **kwargs)
162+
orig_mkdir(*args, **kwargs)
163+
with swap_attr(os, 'mkdir', concurrent_mkdir) as orig_mkdir:
164+
with self.open(self.ar_with_dir) as ar:
165+
self.extractall(ar)
166+
self.assertTrue(os.path.isdir(target))
167+
168+
def test_concurrent_extract_implicit_dir(self):
169+
target = os.path.join(self.testdir, 'test')
170+
def concurrent_mkdir(*args, **kwargs):
171+
orig_mkdir(*args, **kwargs)
172+
orig_mkdir(*args, **kwargs)
173+
with swap_attr(os, 'mkdir', concurrent_mkdir) as orig_mkdir:
174+
with self.open(self.ar_with_implicit_dir) as ar:
175+
self.extractall(ar)
176+
self.assertTrue(os.path.isdir(target))
177+
self.assertTrue(os.path.isfile(os.path.join(target, 'file')))

‎Lib/test/test_zipfile/__init__.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
import os
2+
from test.support import load_package_tests
3+
4+
def load_tests(*args):
5+
return load_package_tests(os.path.dirname(__file__), *args)

‎Lib/test/test_zipfile/__main__.py‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
import unittest
2+
3+
from . import load_tests # noqa: F401
4+
5+
6+
if __name__ == "__main__":
7+
unittest.main()

‎Lib/test/test_zipfile/_path/__init__.py‎

Whitespace-only changes.

‎Lib/test/test_zipfile/_path/_functools.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
import functools
2+
3+
4+
# from jaraco.functools 3.5.2
5+
def compose(*funcs):
6+
def compose_two(f1, f2):
7+
return lambda *args, **kwargs: f1(f2(*args, **kwargs))
8+
9+
return functools.reduce(compose_two, funcs)

‎Lib/test/test_zipfile/_path/_itertools.py‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
import itertools
2+
from collections import deque
3+
from itertools import islice
4+
5+
6+
# from jaraco.itertools 6.3.0
7+
class Counter:
8+
"""
9+
Wrap an iterable in an object that stores the count of items
10+
that pass through it.
11+
12+
>>> items = Counter(range(20))
13+
>>> items.count
14+
0
15+
>>> values = list(items)
16+
>>> items.count
17+
20
18+
"""
19+
20+
def __init__(self, i):
21+
self.count = 0
22+
self.iter = zip(itertools.count(1), i)
23+
24+
def __iter__(self):
25+
return self
26+
27+
def __next__(self):
28+
self.count, result = next(self.iter)
29+
return result
30+
31+
32+
# from more_itertools v8.13.0
33+
def always_iterable(obj, base_type=(str, bytes)):
34+
if obj is None:
35+
return iter(())
36+
37+
if (base_type is not None) and isinstance(obj, base_type):
38+
return iter((obj,))
39+
40+
try:
41+
return iter(obj)
42+
except TypeError:
43+
return iter((obj,))
44+
45+
46+
# from more_itertools v9.0.0
47+
def consume(iterator, n=None):
48+
"""Advance *iterable* by *n* steps. If *n* is ``None``, consume it
49+
entirely.
50+
Efficiently exhausts an iterator without returning values. Defaults to
51+
consuming the whole iterator, but an optional second argument may be
52+
provided to limit consumption.
53+
>>> i = (x for x in range(10))
54+
>>> next(i)
55+
0
56+
>>> consume(i, 3)
57+
>>> next(i)
58+
4
59+
>>> consume(i)
60+
>>> next(i)
61+
Traceback (most recent call last):
62+
File "<stdin>", line 1, in <module>
63+
StopIteration
64+
If the iterator has fewer items remaining than the provided limit, the
65+
whole iterator will be consumed.
66+
>>> i = (x for x in range(3))
67+
>>> consume(i, 5)
68+
>>> next(i)
69+
Traceback (most recent call last):
70+
File "<stdin>", line 1, in <module>
71+
StopIteration
72+
"""
73+
# Use functions that consume iterators at C speed.
74+
if n is None:
75+
# feed the entire iterator into a zero-length deque
76+
deque(iterator, maxlen=0)
77+
else:
78+
# advance to the empty slice starting at position n
79+
next(islice(iterator, n, n), None)

‎Lib/test/test_zipfile/_path/_support.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
import importlib
2+
import unittest
3+
4+
5+
def import_or_skip(name):
6+
try:
7+
return importlib.import_module(name)
8+
except ImportError: # pragma: no cover
9+
raise unittest.SkipTest(f'Unable to import {name}')

‎Lib/test/test_zipfile/_path/_test_params.py‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
import functools
2+
import types
3+
4+
from ._itertools import always_iterable
5+
6+
7+
def parameterize(names, value_groups):
8+
"""
9+
Decorate a test method to run it as a set of subtests.
10+
11+
Modeled after pytest.parametrize.
12+
"""
13+
14+
def decorator(func):
15+
@functools.wraps(func)
16+
def wrapped(self):
17+
for values in value_groups:
18+
resolved = map(Invoked.eval, always_iterable(values))
19+
params = dict(zip(always_iterable(names), resolved))
20+
with self.subTest(**params):
21+
func(self, **params)
22+
23+
return wrapped
24+
25+
return decorator
26+
27+
28+
class Invoked(types.SimpleNamespace):
29+
"""
30+
Wrap a function to be invoked for each usage.
31+
"""
32+
33+
@classmethod
34+
def wrap(cls, func):
35+
return cls(func=func)
36+
37+
@classmethod
38+
def eval(cls, cand):
39+
return cand.func() if isinstance(cand, cls) else cand

0 commit comments

Comments
 (0)