RustPython · youknowone · Apr 23, 2024
diff --git a/Lib/test/archiver_tests.py b/Lib/test/archiver_tests.py
@@ -0,0 +1,155 @@
+"""Tests common to tarfile and zipfile."""
+
+import os
+import sys
+
+from test.support import os_helper
+
+class OverwriteTests:
+
+    def setUp(self):
+        os.makedirs(self.testdir)
+        self.addCleanup(os_helper.rmtree, self.testdir)
+
+    def create_file(self, path, content=b''):
+        with open(path, 'wb') as f:
+            f.write(content)
+
+    def open(self, path):
+        raise NotImplementedError
+
+    def extractall(self, ar):
+        raise NotImplementedError
+
+
+    def test_overwrite_file_as_file(self):
+        target = os.path.join(self.testdir, 'test')
+        self.create_file(target, b'content')
+        with self.open(self.ar_with_file) as ar:
+            self.extractall(ar)
+        self.assertTrue(os.path.isfile(target))
+        with open(target, 'rb') as f:
+            self.assertEqual(f.read(), b'newcontent')
+
+    def test_overwrite_dir_as_dir(self):
+        target = os.path.join(self.testdir, 'test')
+        os.mkdir(target)
+        with self.open(self.ar_with_dir) as ar:
+            self.extractall(ar)
+        self.assertTrue(os.path.isdir(target))
+
+    def test_overwrite_dir_as_implicit_dir(self):
+        target = os.path.join(self.testdir, 'test')
+        os.mkdir(target)
+        with self.open(self.ar_with_implicit_dir) as ar:
+            self.extractall(ar)
+        self.assertTrue(os.path.isdir(target))
+        self.assertTrue(os.path.isfile(os.path.join(target, 'file')))
+        with open(os.path.join(target, 'file'), 'rb') as f:
+            self.assertEqual(f.read(), b'newcontent')
+
+    def test_overwrite_dir_as_file(self):
+        target = os.path.join(self.testdir, 'test')
+        os.mkdir(target)
+        with self.open(self.ar_with_file) as ar:
+            with self.assertRaises(PermissionError if sys.platform == 'win32'
+                                   else IsADirectoryError):
+                self.extractall(ar)
+        self.assertTrue(os.path.isdir(target))
+
+    def test_overwrite_file_as_dir(self):
+        target = os.path.join(self.testdir, 'test')
+        self.create_file(target, b'content')
+        with self.open(self.ar_with_dir) as ar:
+            with self.assertRaises(FileExistsError):
+                self.extractall(ar)
+        self.assertTrue(os.path.isfile(target))
+        with open(target, 'rb') as f:
+            self.assertEqual(f.read(), b'content')
+
+    def test_overwrite_file_as_implicit_dir(self):
+        target = os.path.join(self.testdir, 'test')
+        self.create_file(target, b'content')
+        with self.open(self.ar_with_implicit_dir) as ar:
+            with self.assertRaises(FileNotFoundError if sys.platform == 'win32'
+                                   else NotADirectoryError):
+                self.extractall(ar)
+        self.assertTrue(os.path.isfile(target))
+        with open(target, 'rb') as f:
+            self.assertEqual(f.read(), b'content')
+
+    @os_helper.skip_unless_symlink
+    def test_overwrite_file_symlink_as_file(self):
+        # XXX: It is potential security vulnerability.
+        target = os.path.join(self.testdir, 'test')
+        target2 = os.path.join(self.testdir, 'test2')
+        self.create_file(target2, b'content')
+        os.symlink('test2', target)
+        with self.open(self.ar_with_file) as ar:
+            self.extractall(ar)
+        self.assertTrue(os.path.islink(target))
+        self.assertTrue(os.path.isfile(target2))
+        with open(target2, 'rb') as f:
+            self.assertEqual(f.read(), b'newcontent')
+
+    @os_helper.skip_unless_symlink
+    def test_overwrite_broken_file_symlink_as_file(self):
+        # XXX: It is potential security vulnerability.
+        target = os.path.join(self.testdir, 'test')
+        target2 = os.path.join(self.testdir, 'test2')
+        os.symlink('test2', target)
+        with self.open(self.ar_with_file) as ar:
+            self.extractall(ar)
+        self.assertTrue(os.path.islink(target))
+        self.assertTrue(os.path.isfile(target2))
+        with open(target2, 'rb') as f:
+            self.assertEqual(f.read(), b'newcontent')
+
+    @os_helper.skip_unless_symlink
+    def test_overwrite_dir_symlink_as_dir(self):
+        # XXX: It is potential security vulnerability.
+        target = os.path.join(self.testdir, 'test')
+        target2 = os.path.join(self.testdir, 'test2')
+        os.mkdir(target2)
+        os.symlink('test2', target, target_is_directory=True)
+        with self.open(self.ar_with_dir) as ar:
+            self.extractall(ar)
+        self.assertTrue(os.path.islink(target))
+        self.assertTrue(os.path.isdir(target2))
+
+    @os_helper.skip_unless_symlink
+    def test_overwrite_dir_symlink_as_implicit_dir(self):
+        # XXX: It is potential security vulnerability.
+        target = os.path.join(self.testdir, 'test')
+        target2 = os.path.join(self.testdir, 'test2')
+        os.mkdir(target2)
+        os.symlink('test2', target, target_is_directory=True)
+        with self.open(self.ar_with_implicit_dir) as ar:
+            self.extractall(ar)
+        self.assertTrue(os.path.islink(target))
+        self.assertTrue(os.path.isdir(target2))
+        self.assertTrue(os.path.isfile(os.path.join(target2, 'file')))
+        with open(os.path.join(target2, 'file'), 'rb') as f:
+            self.assertEqual(f.read(), b'newcontent')
+
+    @os_helper.skip_unless_symlink
+    def test_overwrite_broken_dir_symlink_as_dir(self):
+        target = os.path.join(self.testdir, 'test')
+        target2 = os.path.join(self.testdir, 'test2')
+        os.symlink('test2', target, target_is_directory=True)
+        with self.open(self.ar_with_dir) as ar:
+            with self.assertRaises(FileExistsError):
+                self.extractall(ar)
+        self.assertTrue(os.path.islink(target))
+        self.assertFalse(os.path.exists(target2))
+
+    @os_helper.skip_unless_symlink
+    def test_overwrite_broken_dir_symlink_as_implicit_dir(self):
+        target = os.path.join(self.testdir, 'test')
+        target2 = os.path.join(self.testdir, 'test2')
+        os.symlink('test2', target, target_is_directory=True)
+        with self.open(self.ar_with_implicit_dir) as ar:
+            with self.assertRaises(FileExistsError):
+                self.extractall(ar)
+        self.assertTrue(os.path.islink(target))
+        self.assertFalse(os.path.exists(target2))
diff --git a/Lib/test/test_zipfile/__init__.py b/Lib/test/test_zipfile/__init__.py
@@ -0,0 +1,5 @@
+import os
+from test.support import load_package_tests
+
+def load_tests(*args):
+    return load_package_tests(os.path.dirname(__file__), *args)
diff --git a/Lib/test/test_zipfile/__main__.py b/Lib/test/test_zipfile/__main__.py
@@ -0,0 +1,7 @@
+import unittest
+
+from . import load_tests  # noqa: F401
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/Lib/test/test_zipfile/_path/__init__.py b/Lib/test/test_zipfile/_path/__init__.py
diff --git a/Lib/test/test_zipfile/_path/_functools.py b/Lib/test/test_zipfile/_path/_functools.py
@@ -0,0 +1,9 @@
+import functools
+
+
+# from jaraco.functools 3.5.2
+def compose(*funcs):
+    def compose_two(f1, f2):
+        return lambda *args, **kwargs: f1(f2(*args, **kwargs))
+
+    return functools.reduce(compose_two, funcs)
diff --git a/Lib/test/test_zipfile/_path/_itertools.py b/Lib/test/test_zipfile/_path/_itertools.py
@@ -0,0 +1,79 @@
+import itertools
+from collections import deque
+from itertools import islice
+
+
+# from jaraco.itertools 6.3.0
+class Counter:
+    """
+    Wrap an iterable in an object that stores the count of items
+    that pass through it.
+
+    >>> items = Counter(range(20))
+    >>> items.count
+    0
+    >>> values = list(items)
+    >>> items.count
+    20
+    """
+
+    def __init__(self, i):
+        self.count = 0
+        self.iter = zip(itertools.count(1), i)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        self.count, result = next(self.iter)
+        return result
+
+
+# from more_itertools v8.13.0
+def always_iterable(obj, base_type=(str, bytes)):
+    if obj is None:
+        return iter(())
+
+    if (base_type is not None) and isinstance(obj, base_type):
+        return iter((obj,))
+
+    try:
+        return iter(obj)
+    except TypeError:
+        return iter((obj,))
+
+
+# from more_itertools v9.0.0
+def consume(iterator, n=None):
+    """Advance *iterable* by *n* steps. If *n* is ``None``, consume it
+    entirely.
+    Efficiently exhausts an iterator without returning values. Defaults to
+    consuming the whole iterator, but an optional second argument may be
+    provided to limit consumption.
+        >>> i = (x for x in range(10))
+        >>> next(i)
+        0
+        >>> consume(i, 3)
+        >>> next(i)
+        4
+        >>> consume(i)
+        >>> next(i)
+        Traceback (most recent call last):
+          File "<stdin>", line 1, in <module>
+        StopIteration
+    If the iterator has fewer items remaining than the provided limit, the
+    whole iterator will be consumed.
+        >>> i = (x for x in range(3))
+        >>> consume(i, 5)
+        >>> next(i)
+        Traceback (most recent call last):
+          File "<stdin>", line 1, in <module>
+        StopIteration
+    """
+    # Use functions that consume iterators at C speed.
+    if n is None:
+        # feed the entire iterator into a zero-length deque
+        deque(iterator, maxlen=0)
+    else:
+        # advance to the empty slice starting at position n
+        next(islice(iterator, n, n), None)
diff --git a/Lib/test/test_zipfile/_path/_support.py b/Lib/test/test_zipfile/_path/_support.py
@@ -0,0 +1,9 @@
+import importlib
+import unittest
+
+
+def import_or_skip(name):
+    try:
+        return importlib.import_module(name)
+    except ImportError:  # pragma: no cover
+        raise unittest.SkipTest(f'Unable to import {name}')
diff --git a/Lib/test/test_zipfile/_path/_test_params.py b/Lib/test/test_zipfile/_path/_test_params.py
@@ -0,0 +1,39 @@
+import types
+import functools
+
+from ._itertools import always_iterable
+
+
+def parameterize(names, value_groups):
+    """
+    Decorate a test method to run it as a set of subtests.
+
+    Modeled after pytest.parametrize.
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapped(self):
+            for values in value_groups:
+                resolved = map(Invoked.eval, always_iterable(values))
+                params = dict(zip(always_iterable(names), resolved))
+                with self.subTest(**params):
+                    func(self, **params)
+
+        return wrapped
+
+    return decorator
+
+
+class Invoked(types.SimpleNamespace):
+    """
+    Wrap a function to be invoked for each usage.
+    """
+
+    @classmethod
+    def wrap(cls, func):
+        return cls(func=func)
+
+    @classmethod
+    def eval(cls, cand):
+        return cand.func() if isinstance(cand, cls) else cand