GH-89812: Add `pathlib._PathBase` (#106337)

Add private `pathlib._PathBase` class. This will be used by an experimental PyPI package to incubate a `tarfile.TarPath` class.

Co-authored-by: Adam Turner <9087854+AA-Turner@users.noreply.github.com>

1 files changed, 326 insertions, 74 deletions

diff --git a/Lib/test/test_pathlib.py b/Lib/test/test_pathlib.py
index 484a5e6c3bd..319148e9065 100644
--- a/Lib/test/test_pathlib.py
+++ b/Lib/test/test_pathlib.py
@@ -1582,14 +1582,172 @@ class WindowsPathAsPureTest(PureWindowsPathTest):
 
 
 #
-# Tests for the concrete classes.
+# Tests for the virtual classes.
 #
 
-class PathTest(unittest.TestCase):
-    """Tests for the FS-accessing functionalities of the Path classes."""
+class PathBaseTest(PurePathTest):
+    cls = pathlib._PathBase
 
-    cls = pathlib.Path
-    can_symlink = os_helper.can_symlink()
+    def test_unsupported_operation(self):
+        P = self.cls
+        p = self.cls()
+        e = pathlib.UnsupportedOperation
+        self.assertRaises(e, p.stat)
+        self.assertRaises(e, p.lstat)
+        self.assertRaises(e, p.exists)
+        self.assertRaises(e, p.samefile, 'foo')
+        self.assertRaises(e, p.is_dir)
+        self.assertRaises(e, p.is_file)
+        self.assertRaises(e, p.is_mount)
+        self.assertRaises(e, p.is_symlink)
+        self.assertRaises(e, p.is_block_device)
+        self.assertRaises(e, p.is_char_device)
+        self.assertRaises(e, p.is_fifo)
+        self.assertRaises(e, p.is_socket)
+        self.assertRaises(e, p.open)
+        self.assertRaises(e, p.read_bytes)
+        self.assertRaises(e, p.read_text)
+        self.assertRaises(e, p.write_bytes, b'foo')
+        self.assertRaises(e, p.write_text, 'foo')
+        self.assertRaises(e, p.iterdir)
+        self.assertRaises(e, p.glob, '*')
+        self.assertRaises(e, p.rglob, '*')
+        self.assertRaises(e, lambda: list(p.walk()))
+        self.assertRaises(e, p.absolute)
+        self.assertRaises(e, P.cwd)
+        self.assertRaises(e, p.expanduser)
+        self.assertRaises(e, p.home)
+        self.assertRaises(e, p.readlink)
+        self.assertRaises(e, p.symlink_to, 'foo')
+        self.assertRaises(e, p.hardlink_to, 'foo')
+        self.assertRaises(e, p.mkdir)
+        self.assertRaises(e, p.touch)
+        self.assertRaises(e, p.rename, 'foo')
+        self.assertRaises(e, p.replace, 'foo')
+        self.assertRaises(e, p.chmod, 0o755)
+        self.assertRaises(e, p.lchmod, 0o755)
+        self.assertRaises(e, p.unlink)
+        self.assertRaises(e, p.rmdir)
+        self.assertRaises(e, p.owner)
+        self.assertRaises(e, p.group)
+        self.assertRaises(e, p.as_uri)
+
+    def test_as_uri_common(self):
+        e = pathlib.UnsupportedOperation
+        self.assertRaises(e, self.cls().as_uri)
+
+    def test_fspath_common(self):
+        self.assertRaises(TypeError, os.fspath, self.cls())
+
+    def test_as_bytes_common(self):
+        self.assertRaises(TypeError, bytes, self.cls())
+
+    def test_matches_path_api(self):
+        our_names = {name for name in dir(self.cls) if name[0] != '_'}
+        path_names = {name for name in dir(pathlib.Path) if name[0] != '_'}
+        self.assertEqual(our_names, path_names)
+        for attr_name in our_names:
+            our_attr = getattr(self.cls, attr_name)
+            path_attr = getattr(pathlib.Path, attr_name)
+            self.assertEqual(our_attr.__doc__, path_attr.__doc__)
+
+
+class DummyPathIO(io.BytesIO):
+    """
+    Used by DummyPath to implement `open('w')`
+    """
+
+    def __init__(self, files, path):
+        super().__init__()
+        self.files = files
+        self.path = path
+
+    def close(self):
+        self.files[self.path] = self.getvalue()
+        super().close()
+
+
+class DummyPath(pathlib._PathBase):
+    """
+    Simple implementation of PathBase that keeps files and directories in
+    memory.
+    """
+    _files = {}
+    _directories = {}
+    _symlinks = {}
+
+    def stat(self, *, follow_symlinks=True):
+        if follow_symlinks:
+            path = str(self.resolve())
+        else:
+            path = str(self.parent.resolve() / self.name)
+        if path in self._files:
+            st_mode = stat.S_IFREG
+        elif path in self._directories:
+            st_mode = stat.S_IFDIR
+        elif path in self._symlinks:
+            st_mode = stat.S_IFLNK
+        else:
+            raise FileNotFoundError(errno.ENOENT, "Not found", str(self))
+        return os.stat_result((st_mode, hash(str(self)), 0, 0, 0, 0, 0, 0, 0, 0))
+
+    def open(self, mode='r', buffering=-1, encoding=None,
+             errors=None, newline=None):
+        if buffering != -1:
+            raise NotImplementedError
+        path_obj = self.resolve()
+        path = str(path_obj)
+        name = path_obj.name
+        parent = str(path_obj.parent)
+        if path in self._directories:
+            raise IsADirectoryError(errno.EISDIR, "Is a directory", path)
+
+        text = 'b' not in mode
+        mode = ''.join(c for c in mode if c not in 'btU')
+        if mode == 'r':
+            if path not in self._files:
+                raise FileNotFoundError(errno.ENOENT, "File not found", path)
+            stream = io.BytesIO(self._files[path])
+        elif mode == 'w':
+            if parent not in self._directories:
+                raise FileNotFoundError(errno.ENOENT, "File not found", parent)
+            stream = DummyPathIO(self._files, path)
+            self._files[path] = b''
+            self._directories[parent].add(name)
+        else:
+            raise NotImplementedError
+        if text:
+            stream = io.TextIOWrapper(stream, encoding=encoding, errors=errors, newline=newline)
+        return stream
+
+    def iterdir(self):
+        path = str(self.resolve())
+        if path in self._files:
+            raise NotADirectoryError(errno.ENOTDIR, "Not a directory", path)
+        elif path in self._directories:
+            return (self / name for name in self._directories[path])
+        else:
+            raise FileNotFoundError(errno.ENOENT, "File not found", path)
+
+    def mkdir(self, mode=0o777, parents=False, exist_ok=False):
+        try:
+            self._directories[str(self.parent)].add(self.name)
+            self._directories[str(self)] = set()
+        except KeyError:
+            if not parents or self.parent == self:
+                raise FileNotFoundError(errno.ENOENT, "File not found", str(self.parent)) from None
+            self.parent.mkdir(parents=True, exist_ok=True)
+            self.mkdir(mode, parents=False, exist_ok=exist_ok)
+        except FileExistsError:
+            if not exist_ok:
+                raise
+
+
+class DummyPathTest(unittest.TestCase):
+    """Tests for PathBase methods that use stat(), open() and iterdir()."""
+
+    cls = DummyPath
+    can_symlink = False
 
     # (BASE)
     #  |
@@ -1612,37 +1770,38 @@ class PathTest(unittest.TestCase):
     #
 
     def setUp(self):
-        def cleanup():
-            os.chmod(join('dirE'), 0o777)
-            os_helper.rmtree(BASE)
-        self.addCleanup(cleanup)
-        os.mkdir(BASE)
-        os.mkdir(join('dirA'))
-        os.mkdir(join('dirB'))
-        os.mkdir(join('dirC'))
-        os.mkdir(join('dirC', 'dirD'))
-        os.mkdir(join('dirE'))
-        with open(join('fileA'), 'wb') as f:
-            f.write(b"this is file A\n")
-        with open(join('dirB', 'fileB'), 'wb') as f:
-            f.write(b"this is file B\n")
-        with open(join('dirC', 'fileC'), 'wb') as f:
-            f.write(b"this is file C\n")
-        with open(join('dirC', 'novel.txt'), 'wb') as f:
-            f.write(b"this is a novel\n")
-        with open(join('dirC', 'dirD', 'fileD'), 'wb') as f:
-            f.write(b"this is file D\n")
-        os.chmod(join('dirE'), 0)
-        if self.can_symlink:
-            # Relative symlinks.
-            os.symlink('fileA', join('linkA'))
-            os.symlink('non-existing', join('brokenLink'))
-            os.symlink('dirB', join('linkB'), target_is_directory=True)
-            os.symlink(os.path.join('..', 'dirB'), join('dirA', 'linkC'), target_is_directory=True)
-            # This one goes upwards, creating a loop.
-            os.symlink(os.path.join('..', 'dirB'), join('dirB', 'linkD'), target_is_directory=True)
-            # Broken symlink (pointing to itself).
-            os.symlink('brokenLinkLoop',  join('brokenLinkLoop'))
+        # note: this must be kept in sync with `PathTest.setUp()`
+        cls = self.cls
+        cls._files.clear()
+        cls._directories.clear()
+        cls._symlinks.clear()
+        join = cls.pathmod.join
+        cls._files.update({
+            join(BASE, 'fileA'): b'this is file A\n',
+            join(BASE, 'dirB', 'fileB'): b'this is file B\n',
+            join(BASE, 'dirC', 'fileC'): b'this is file C\n',
+            join(BASE, 'dirC', 'dirD', 'fileD'): b'this is file D\n',
+            join(BASE, 'dirC', 'novel.txt'): b'this is a novel\n',
+        })
+        cls._directories.update({
+            BASE: {'dirA', 'dirB', 'dirC', 'dirE', 'fileA'},
+            join(BASE, 'dirA'): set(),
+            join(BASE, 'dirB'): {'fileB'},
+            join(BASE, 'dirC'): {'dirD', 'fileC', 'novel.txt'},
+            join(BASE, 'dirC', 'dirD'): {'fileD'},
+            join(BASE, 'dirE'): {},
+        })
+        dirname = BASE
+        while True:
+            dirname, basename = cls.pathmod.split(dirname)
+            if not basename:
+                break
+            cls._directories[dirname] = {basename}
+
+    def tempdir(self):
+        path = self.cls(BASE).with_name('tmp-dirD')
+        path.mkdir()
+        return path
 
     def assertFileNotFound(self, func, *args, **kwargs):
         with self.assertRaises(FileNotFoundError) as cm:
@@ -1991,9 +2150,11 @@ class PathTest(unittest.TestCase):
     def test_glob_many_open_files(self):
         depth = 30
         P = self.cls
-        base = P(BASE) / 'deep'
-        p = P(base, *(['d']*depth))
-        p.mkdir(parents=True)
+        p = base = P(BASE) / 'deep'
+        p.mkdir()
+        for _ in range(depth):
+            p /= 'd'
+            p.mkdir()
         pattern = '/'.join(['*'] * depth)
         iters = [base.glob(pattern) for j in range(100)]
         for it in iters:
@@ -2080,6 +2241,7 @@ class PathTest(unittest.TestCase):
         self.assertEqual((P / 'brokenLink').readlink(),
                          self.cls('non-existing'))
         self.assertEqual((P / 'linkB').readlink(), self.cls('dirB'))
+        self.assertEqual((P / 'linkB' / 'linkD').readlink(), self.cls('../dirB'))
         with self.assertRaises(OSError):
             (P / 'fileA').readlink()
 
@@ -2128,7 +2290,7 @@ class PathTest(unittest.TestCase):
         self._check_resolve_relative(p, P(BASE, 'dirB', 'fileB', 'foo', 'in',
                                           'spam'), False)
         p = P(BASE, 'dirA', 'linkC', '..', 'foo', 'in', 'spam')
-        if os.name == 'nt':
+        if os.name == 'nt' and isinstance(p, pathlib.Path):
             # In Windows, if linkY points to dirB, 'dirA\linkY\..'
             # resolves to 'dirA' without resolving linkY first.
             self._check_resolve_relative(p, P(BASE, 'dirA', 'foo', 'in',
@@ -2138,9 +2300,7 @@ class PathTest(unittest.TestCase):
             # resolves to 'dirB/..' first before resolving to parent of dirB.
             self._check_resolve_relative(p, P(BASE, 'foo', 'in', 'spam'), False)
         # Now create absolute symlinks.
-        d = os_helper._longpath(tempfile.mkdtemp(suffix='-dirD',
-                                                 dir=os.getcwd()))
-        self.addCleanup(os_helper.rmtree, d)
+        d = self.tempdir()
         P(BASE, 'dirA', 'linkX').symlink_to(d)
         P(BASE, str(d), 'linkY').symlink_to(join('dirB'))
         p = P(BASE, 'dirA', 'linkX', 'linkY', 'fileB')
@@ -2150,7 +2310,7 @@ class PathTest(unittest.TestCase):
         self._check_resolve_relative(p, P(BASE, 'dirB', 'foo', 'in', 'spam'),
                                      False)
         p = P(BASE, 'dirA', 'linkX', 'linkY', '..', 'foo', 'in', 'spam')
-        if os.name == 'nt':
+        if os.name == 'nt' and isinstance(p, pathlib.Path):
             # In Windows, if linkY points to dirB, 'dirA\linkY\..'
             # resolves to 'dirA' without resolving linkY first.
             self._check_resolve_relative(p, P(d, 'foo', 'in', 'spam'), False)
@@ -2174,6 +2334,38 @@ class PathTest(unittest.TestCase):
         # Non-strict
         self.assertEqual(r.resolve(strict=False), p / '3' / '4')
 
+    def _check_symlink_loop(self, *args):
+        path = self.cls(*args)
+        with self.assertRaises(OSError) as cm:
+            path.resolve(strict=True)
+        self.assertEqual(cm.exception.errno, errno.ELOOP)
+
+    def test_resolve_loop(self):
+        if not self.can_symlink:
+            self.skipTest("symlinks required")
+        if os.name == 'nt' and issubclass(self.cls, pathlib.Path):
+            self.skipTest("symlink loops work differently with concrete Windows paths")
+        # Loops with relative symlinks.
+        self.cls(BASE, 'linkX').symlink_to('linkX/inside')
+        self._check_symlink_loop(BASE, 'linkX')
+        self.cls(BASE, 'linkY').symlink_to('linkY')
+        self._check_symlink_loop(BASE, 'linkY')
+        self.cls(BASE, 'linkZ').symlink_to('linkZ/../linkZ')
+        self._check_symlink_loop(BASE, 'linkZ')
+        # Non-strict
+        p = self.cls(BASE, 'linkZ', 'foo')
+        self.assertEqual(p.resolve(strict=False), p)
+        # Loops with absolute symlinks.
+        self.cls(BASE, 'linkU').symlink_to(join('linkU/inside'))
+        self._check_symlink_loop(BASE, 'linkU')
+        self.cls(BASE, 'linkV').symlink_to(join('linkV'))
+        self._check_symlink_loop(BASE, 'linkV')
+        self.cls(BASE, 'linkW').symlink_to(join('linkW/../linkW'))
+        self._check_symlink_loop(BASE, 'linkW')
+        # Non-strict
+        q = self.cls(BASE, 'linkW', 'foo')
+        self.assertEqual(q.resolve(strict=False), q)
+
     def test_stat(self):
         statA = self.cls(BASE).joinpath('fileA').stat()
         statB = self.cls(BASE).joinpath('dirB', 'fileB').stat()
@@ -2382,6 +2574,10 @@ class PathTest(unittest.TestCase):
         self.assertEqualNormCase(str(p), BASE)
 
         # Resolve relative paths.
+        try:
+            self.cls().absolute()
+        except pathlib.UnsupportedOperation:
+            return
         old_path = os.getcwd()
         os.chdir(BASE)
         try:
@@ -2409,6 +2605,92 @@ class PathTest(unittest.TestCase):
     def test_complex_symlinks_relative_dot_dot(self):
         self._check_complex_symlinks(os.path.join('dirA', '..'))
 
+
+class DummyPathWithSymlinks(DummyPath):
+    def readlink(self):
+        path = str(self.parent.resolve() / self.name)
+        if path in self._symlinks:
+            return self.with_segments(self._symlinks[path])
+        elif path in self._files or path in self._directories:
+            raise OSError(errno.EINVAL, "Not a symlink", path)
+        else:
+            raise FileNotFoundError(errno.ENOENT, "File not found", path)
+
+    def symlink_to(self, target, target_is_directory=False):
+        self._directories[str(self.parent)].add(self.name)
+        self._symlinks[str(self)] = str(target)
+
+
+class DummyPathWithSymlinksTest(DummyPathTest):
+    cls = DummyPathWithSymlinks
+    can_symlink = True
+
+    def setUp(self):
+        super().setUp()
+        cls = self.cls
+        join = cls.pathmod.join
+        cls._symlinks.update({
+            join(BASE, 'linkA'): 'fileA',
+            join(BASE, 'linkB'): 'dirB',
+            join(BASE, 'dirA', 'linkC'): join('..', 'dirB'),
+            join(BASE, 'dirB', 'linkD'): join('..', 'dirB'),
+            join(BASE, 'brokenLink'): 'non-existing',
+            join(BASE, 'brokenLinkLoop'): 'brokenLinkLoop',
+        })
+        cls._directories[BASE].update({'linkA', 'linkB', 'brokenLink', 'brokenLinkLoop'})
+        cls._directories[join(BASE, 'dirA')].add('linkC')
+        cls._directories[join(BASE, 'dirB')].add('linkD')
+
+
+#
+# Tests for the concrete classes.
+#
+
+class PathTest(DummyPathTest):
+    """Tests for the FS-accessing functionalities of the Path classes."""
+    cls = pathlib.Path
+    can_symlink = os_helper.can_symlink()
+
+    def setUp(self):
+        # note: this must be kept in sync with `DummyPathTest.setUp()`
+        def cleanup():
+            os.chmod(join('dirE'), 0o777)
+            os_helper.rmtree(BASE)
+        self.addCleanup(cleanup)
+        os.mkdir(BASE)
+        os.mkdir(join('dirA'))
+        os.mkdir(join('dirB'))
+        os.mkdir(join('dirC'))
+        os.mkdir(join('dirC', 'dirD'))
+        os.mkdir(join('dirE'))
+        with open(join('fileA'), 'wb') as f:
+            f.write(b"this is file A\n")
+        with open(join('dirB', 'fileB'), 'wb') as f:
+            f.write(b"this is file B\n")
+        with open(join('dirC', 'fileC'), 'wb') as f:
+            f.write(b"this is file C\n")
+        with open(join('dirC', 'novel.txt'), 'wb') as f:
+            f.write(b"this is a novel\n")
+        with open(join('dirC', 'dirD', 'fileD'), 'wb') as f:
+            f.write(b"this is file D\n")
+        os.chmod(join('dirE'), 0)
+        if self.can_symlink:
+            # Relative symlinks.
+            os.symlink('fileA', join('linkA'))
+            os.symlink('non-existing', join('brokenLink'))
+            os.symlink('dirB', join('linkB'), target_is_directory=True)
+            os.symlink(os.path.join('..', 'dirB'), join('dirA', 'linkC'), target_is_directory=True)
+            # This one goes upwards, creating a loop.
+            os.symlink(os.path.join('..', 'dirB'), join('dirB', 'linkD'), target_is_directory=True)
+            # Broken symlink (pointing to itself).
+            os.symlink('brokenLinkLoop',  join('brokenLinkLoop'))
+
+    def tempdir(self):
+        d = os_helper._longpath(tempfile.mkdtemp(suffix='-dirD',
+                                                 dir=os.getcwd()))
+        self.addCleanup(os_helper.rmtree, d)
+        return d
+
     def test_concrete_class(self):
         if self.cls is pathlib.Path:
             expected = pathlib.WindowsPath if os.name == 'nt' else pathlib.PosixPath
@@ -3178,12 +3460,6 @@ class PosixPathTest(PathTest):
         self.assertEqual(str(P('//a').absolute()), '//a')
         self.assertEqual(str(P('//a/b').absolute()), '//a/b')
 
-    def _check_symlink_loop(self, *args):
-        path = self.cls(*args)
-        with self.assertRaises(OSError) as cm:
-            path.resolve(strict=True)
-        self.assertEqual(cm.exception.errno, errno.ELOOP)
-
     @unittest.skipIf(
         is_emscripten or is_wasi,
         "umask is not implemented on Emscripten/WASI."
@@ -3230,30 +3506,6 @@ class PosixPathTest(PathTest):
         st = os.stat(join('masked_new_file'))
         self.assertEqual(stat.S_IMODE(st.st_mode), 0o750)
 
-    def test_resolve_loop(self):
-        if not self.can_symlink:
-            self.skipTest("symlinks required")
-        # Loops with relative symlinks.
-        os.symlink('linkX/inside', join('linkX'))
-        self._check_symlink_loop(BASE, 'linkX')
-        os.symlink('linkY', join('linkY'))
-        self._check_symlink_loop(BASE, 'linkY')
-        os.symlink('linkZ/../linkZ', join('linkZ'))
-        self._check_symlink_loop(BASE, 'linkZ')
-        # Non-strict
-        p = self.cls(BASE, 'linkZ', 'foo')
-        self.assertEqual(p.resolve(strict=False), p)
-        # Loops with absolute symlinks.
-        os.symlink(join('linkU/inside'), join('linkU'))
-        self._check_symlink_loop(BASE, 'linkU')
-        os.symlink(join('linkV'), join('linkV'))
-        self._check_symlink_loop(BASE, 'linkV')
-        os.symlink(join('linkW/../linkW'), join('linkW'))
-        self._check_symlink_loop(BASE, 'linkW')
-        # Non-strict
-        q = self.cls(BASE, 'linkW', 'foo')
-        self.assertEqual(q.resolve(strict=False), q)
-
     def test_glob(self):
         P = self.cls
         p = P(BASE)