Merge pull request #3 from jaraco/experiment/tests

Ingest the tests from CPython

.coveragerc

@@ -2,6 +2,9 @@
 omit =
 	# leading `*/` for pytest-dev/pytest-cov#456
 	*/.tox/*
+
+	# local
+	*/compat/*
 disable_warnings =
 	couldnt-parse
 

backports/tarfile.py → backports/tarfile/__init__.py

@@ -48,6 +48,8 @@
 import re
 import warnings
 
+from .compat.py38 import removesuffix
+
 try:
     import pwd
 except ImportError:
@@ -1365,7 +1367,7 @@ def _proc_gnulong(self, tarfile):
         # Remove redundant slashes from directories. This is to be consistent
         # with frombuf().
         if next.isdir():
-            next.name = next.name.removesuffix("/")
+            next.name = removesuffix(next.name, "/")
 
         return next
 

backports/tarfile/__main__.py

@@ -0,0 +1,5 @@
+from . import main
+
+
+if __name__ == '__main__':
+    main()

backports/tarfile/compat/py38.py

@@ -0,0 +1,24 @@
+import sys
+
+
+if sys.version_info < (3, 9):
+
+    def removesuffix(self, suffix):
+        # suffix='' should not call self[:-0].
+        if suffix and self.endswith(suffix):
+            return self[: -len(suffix)]
+        else:
+            return self[:]
+
+    def removeprefix(self, prefix):
+        if self.startswith(prefix):
+            return self[len(prefix) :]
+        else:
+            return self[:]
+else:
+
+    def removesuffix(self, suffix):
+        return self.removesuffix(suffix)
+
+    def removeprefix(self, prefix):
+        return self.removeprefix(prefix)

conftest.py

@@ -0,0 +1,38 @@
+import pathlib
+import sys
+from test import support
+
+import pytest
+
+
+def find_file(name, subdir=None):
+    return str(pathlib.Path(*filter(None, ('tests', subdir, name))).absolute())
+
+
+def patch_findfile():
+    """
+    Early hook to ensure findfile behaves differently before test_tarfile is imported.
+    """
+    support.findfile = find_file
+
+
+def backport_as_std():
+    """
+    Make sure 'import tarfile' gets the backport.
+    """
+    from backports import tarfile
+    sys.modules['tarfile'] = tarfile
+
+
+def pytest_configure():
+    patch_findfile()
+    backport_as_std()
+
+
+@pytest.fixture(scope='module', autouse=True)
+def setup_and_teardown_module(request):
+    request.module.setUpModule()
+    try:
+        yield
+    finally:
+        request.module.tearDownModule()

newsfragments/+df7ee3db.bugfix.rst

@@ -0,0 +1 @@
+Fixed a bug in _proc_gnulong on Python 3.8 where removesuffix was used.

newsfragments/2.feature.rst

@@ -0,0 +1 @@
+Backported tests from CPython.

setup.cfg

@@ -28,6 +28,7 @@ testing =
 	# pytest-ruff >= 0.2.1
 
 	# local
+	jaraco.test
 
 docs =
 	# upstream

tests/compat/archiver_tests.py

@@ -0,0 +1,155 @@
+"""Tests common to tarfile and zipfile."""
+
+import os
+import sys
+
+from .py39 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))

tests/compat/py310.py

@@ -0,0 +1,48 @@
+import contextlib
+import sys
+from .py39 import support as std_support
+from .py39 import os_helper as std_os_helper
+import types
+import warnings as std_warnings
+
+
+
+try:
+    from test import archiver_tests
+except ImportError:
+    from . import archiver_tests  # noqa: F401
+
+
+class os_helper_compat:
+    def skip_unless_working_chmod(test):
+        """Never skip"""
+        return test
+
+    def can_chmod():
+        return True
+
+
+os_helper = types.SimpleNamespace(**{**vars(os_helper_compat), **vars(std_os_helper)})
+
+
+class support_compat:
+    def is_emscripten():
+        return False
+
+    def is_wasi():
+        return False
+
+
+support = types.SimpleNamespace(**{**vars(support_compat), **vars(std_support)})
+
+
+class warnings_compat:
+    if sys.version_info < (3, 11):
+        @contextlib.contextmanager
+        def catch_warnings(*, record=False, module=None, action=None, **kwargs):
+            with std_warnings.catch_warnings(record=record, module=module) as val:
+                if action:
+                    std_warnings.simplefilter(action, **kwargs)
+                yield val
+
+warnings = types.SimpleNamespace(**{**vars(std_warnings), **vars(warnings_compat)})

tests/compat/py38.py

@@ -0,0 +1,77 @@
+import random
+import sys
+import test.support
+import types
+import unittest
+
+from jaraco.test.cpython import from_test_support, try_import
+
+
+warnings_helper = try_import('warnings_helper') or from_test_support('check_warnings')
+
+
+class support_compat:
+    if sys.version_info < (3, 9):
+        def requires_zlib(reason='requires zlib'):
+            try:
+                import zlib
+            except ImportError:
+                zlib = None
+            return unittest.skipUnless(zlib, reason)
+
+        def requires_gzip(reason='requires gzip'):
+            try:
+                import gzip
+            except ImportError:
+                gzip = None
+            return unittest.skipUnless(gzip, reason)
+
+        def requires_bz2(reason='requires bz2'):
+            try:
+                import bz2
+            except ImportError:
+                bz2 = None
+            return unittest.skipUnless(bz2, reason)
+
+        def requires_lzma(reason='requires lzma'):
+            try:
+                import lzma
+            except ImportError:
+                lzma = None
+            return unittest.skipUnless(lzma, reason)
+
+
+support = types.SimpleNamespace(**{**vars(test.support), **vars(support_compat)})
+
+
+class RandomCompat(random.Random):
+    def randbytes(self, n):
+        """Generate n random bytes."""
+        return self.getrandbits(n * 8).to_bytes(n, 'little')
+
+
+Random = RandomCompat if sys.version_info < (3, 9) else random.Random
+
+
+if sys.version_info < (3, 9):
+
+    def removesuffix(self, suffix):
+        # suffix='' should not call self[:-0].
+        if suffix and self.endswith(suffix):
+            return self[: -len(suffix)]
+        else:
+            return self[:]
+
+    def removeprefix(self, prefix):
+        if self.startswith(prefix):
+            return self[len(prefix) :]
+        else:
+            return self[:]
+else:
+
+    def removesuffix(self, suffix):
+        return self.removesuffix(suffix)
+
+    def removeprefix(self, prefix):
+        return self.removeprefix(prefix)
+