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>
This commit is contained in:
Barney Gale 2023-09-30 15:45:01 +01:00 committed by GitHub
parent 0449fe999d
commit 89966a694b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 687 additions and 161 deletions

View file

@ -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)