From 042cb6c1ac0bd78c8235dd254919181261b910c0 Mon Sep 17 00:00:00 2001 From: barneygale Date: Sun, 7 May 2023 20:46:28 +0100 Subject: [PATCH 1/2] Add `pathlib._AbstractPath` --- Lib/pathlib.py | 113 +++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 90 insertions(+), 23 deletions(-) diff --git a/Lib/pathlib.py b/Lib/pathlib.py index 68255aa3e511ec..dcf0191eb1d8d7 100644 --- a/Lib/pathlib.py +++ b/Lib/pathlib.py @@ -5,6 +5,7 @@ operating systems. """ +import contextlib import fnmatch import functools import io @@ -729,14 +730,11 @@ class PureWindowsPath(PurePath): # Filesystem-accessing classes -class Path(PurePath): - """PurePath subclass that can make system calls. +class _AbstractPath(PurePath): + """PurePath subclass that can traverse directories and open files. - Path represents a filesystem path but unlike PurePath, also offers - methods to do system calls on path objects. Depending on your system, - instantiating a Path will return either a PosixPath or a WindowsPath - object. You can also instantiate a PosixPath or WindowsPath directly, - but cannot instantiate a WindowsPath on a POSIX system or vice versa. + The `stat()`, `open()` and `iterdir()` methods are abstract; all other + methods are defined in terms of these three. """ __slots__ = () @@ -745,7 +743,7 @@ def stat(self, *, follow_symlinks=True): Return the result of the stat() system call on this path, like os.stat() does. """ - return os.stat(self, follow_symlinks=follow_symlinks) + raise NotImplementedError def lstat(self): """ @@ -775,12 +773,12 @@ def exists(self, *, follow_symlinks=True): return False return True - def is_dir(self): + def is_dir(self, *, follow_symlinks=True): """ Whether this path is a directory. """ try: - return S_ISDIR(self.stat().st_mode) + return S_ISDIR(self.stat(follow_symlinks=follow_symlinks).st_mode) except OSError as e: if not _ignore_error(e): raise @@ -812,7 +810,21 @@ def is_mount(self): """ Check if this path is a mount point """ - return self._flavour.ismount(self) + # Need to exist and be a dir + if not self.exists() or not self.is_dir(): + return False + + try: + parent_dev = self.parent.stat().st_dev + except OSError: + return False + + dev = self.stat().st_dev + if dev != parent_dev: + return True + ino = self.stat().st_ino + parent_ino = self.parent.stat().st_ino + return ino == parent_ino def is_symlink(self): """ @@ -833,7 +845,15 @@ def is_junction(self): """ Whether this path is a junction. """ - return self._flavour.isjunction(self) + import stat + try: + return self.lstat().st_reparse_tag == stat.IO_REPARSE_TAG_MOUNT_POINT + except OSError as e: + if not _ignore_error(e): + raise + return False + except (ValueError, AttributeError): + return False def is_block_device(self): """ @@ -916,9 +936,7 @@ def open(self, mode='r', buffering=-1, encoding=None, Open the file pointed by this path and return a file object, as the built-in open() function does. """ - if "b" not in mode: - encoding = io.text_encoding(encoding) - return io.open(self, mode, buffering, encoding, errors, newline) + raise NotImplementedError def read_bytes(self): """ @@ -961,14 +979,10 @@ def iterdir(self): The children are yielded in arbitrary order, and the special entries '.' and '..' are not included. """ - for name in os.listdir(self): - yield self._make_child_relpath(name) + raise NotImplementedError def _scandir(self): - # bpo-24132: a future version of pathlib will support subclassing of - # pathlib.Path to customize how the filesystem is accessed. This - # includes scandir(), which is used to implement glob(). - return os.scandir(self) + return contextlib.nullcontext(list(self.iterdir())) def _make_child_relpath(self, name): path_str = str(self) @@ -1034,13 +1048,13 @@ def walk(self, top_down=True, on_error=None, follow_symlinks=False): # blow up for a minor reason when (say) a thousand readable # directories are still left to visit. That logic is copied here. try: - scandir_it = path._scandir() + scandir_obj = path._scandir() except OSError as error: if on_error is not None: on_error(error) continue - with scandir_it: + with scandir_obj as scandir_it: dirnames = [] filenames = [] for entry in scandir_it: @@ -1062,6 +1076,18 @@ def walk(self, top_down=True, on_error=None, follow_symlinks=False): paths += [path._make_child_relpath(d) for d in reversed(dirnames)] + +class Path(_AbstractPath): + """PurePath subclass that can make system calls. + + Path represents a filesystem path but unlike PurePath, also offers + methods to do system calls on path objects. Depending on your system, + instantiating a Path will return either a PosixPath or a WindowsPath + object. You can also instantiate a PosixPath or WindowsPath directly, + but cannot instantiate a WindowsPath on a POSIX system or vice versa. + """ + __slots__ = () + def __init__(self, *args, **kwargs): if kwargs: msg = ("support for supplying keyword arguments to pathlib.PurePath " @@ -1093,6 +1119,47 @@ def __exit__(self, t, v, tb): # Public API + def stat(self, *, follow_symlinks=True): + """ + Return the result of the stat() system call on this path, like + os.stat() does. + """ + return os.stat(self, follow_symlinks=follow_symlinks) + + def is_mount(self): + """ + Check if this path is a mount point + """ + return self._flavour.ismount(self) + + def is_junction(self): + """ + Whether this path is a junction. + """ + return self._flavour.isjunction(self) + + def open(self, mode='r', buffering=-1, encoding=None, + errors=None, newline=None): + """ + Open the file pointed by this path and return a file object, as + the built-in open() function does. + """ + if "b" not in mode: + encoding = io.text_encoding(encoding) + return io.open(self, mode, buffering, encoding, errors, newline) + + def iterdir(self): + """Yield path objects of the directory contents. + + The children are yielded in arbitrary order, and the + special entries '.' and '..' are not included. + """ + for name in os.listdir(self): + yield self._make_child_relpath(name) + + def _scandir(self): + return os.scandir(self) + @classmethod def cwd(cls): """Return a new path pointing to the current working directory.""" From 59cf62b54b6b254790c0d1fbc05b89348b4166cc Mon Sep 17 00:00:00 2001 From: barneygale Date: Sun, 7 May 2023 21:42:09 +0100 Subject: [PATCH 2/2] Add tests for `pathlib._AbstractPath` --- Lib/test/test_pathlib.py | 679 ++++++++++++++++++++------------------- 1 file changed, 350 insertions(+), 329 deletions(-) diff --git a/Lib/test/test_pathlib.py b/Lib/test/test_pathlib.py index e25c77f2ba8af6..273c4bea3d750a 100644 --- a/Lib/test/test_pathlib.py +++ b/Lib/test/test_pathlib.py @@ -1517,7 +1517,7 @@ def test_group(self): P('c:/').group() -class _BasePathTest(object): +class _BaseAbstractPathTest(object): """Tests for the FS-accessing functionalities of the Path classes.""" # (BASE) @@ -1581,86 +1581,10 @@ def dirlink(self, src, dest): def dirlink(self, src, dest): os.symlink(src, dest) - def assertSame(self, path_a, path_b): - self.assertTrue(os.path.samefile(str(path_a), str(path_b)), - "%r and %r don't point to the same file" % - (path_a, path_b)) - - def assertFileNotFound(self, func, *args, **kwargs): - with self.assertRaises(FileNotFoundError) as cm: - func(*args, **kwargs) - self.assertEqual(cm.exception.errno, errno.ENOENT) - - def assertEqualNormCase(self, path_a, path_b): - self.assertEqual(os.path.normcase(path_a), os.path.normcase(path_b)) - - def _test_cwd(self, p): - q = self.cls(os.getcwd()) - self.assertEqual(p, q) - self.assertEqualNormCase(str(p), str(q)) - self.assertIs(type(p), type(q)) - self.assertTrue(p.is_absolute()) - - def test_cwd(self): - p = self.cls.cwd() - self._test_cwd(p) - - def test_absolute_common(self): - P = self.cls - - with mock.patch("os.getcwd") as getcwd: - getcwd.return_value = BASE - - # Simple relative paths. - self.assertEqual(str(P().absolute()), BASE) - self.assertEqual(str(P('.').absolute()), BASE) - self.assertEqual(str(P('a').absolute()), os.path.join(BASE, 'a')) - self.assertEqual(str(P('a', 'b', 'c').absolute()), os.path.join(BASE, 'a', 'b', 'c')) - - # Symlinks should not be resolved. - self.assertEqual(str(P('linkB', 'fileB').absolute()), os.path.join(BASE, 'linkB', 'fileB')) - self.assertEqual(str(P('brokenLink').absolute()), os.path.join(BASE, 'brokenLink')) - self.assertEqual(str(P('brokenLinkLoop').absolute()), os.path.join(BASE, 'brokenLinkLoop')) - - # '..' entries should be preserved and not normalised. - self.assertEqual(str(P('..').absolute()), os.path.join(BASE, '..')) - self.assertEqual(str(P('a', '..').absolute()), os.path.join(BASE, 'a', '..')) - self.assertEqual(str(P('..', 'b').absolute()), os.path.join(BASE, '..', 'b')) - - def _test_home(self, p): - q = self.cls(os.path.expanduser('~')) - self.assertEqual(p, q) - self.assertEqualNormCase(str(p), str(q)) - self.assertIs(type(p), type(q)) - self.assertTrue(p.is_absolute()) - - @unittest.skipIf( - pwd is None, reason="Test requires pwd module to get homedir." - ) - def test_home(self): - with os_helper.EnvironmentVarGuard() as env: - self._test_home(self.cls.home()) - - env.clear() - env['USERPROFILE'] = os.path.join(BASE, 'userprofile') - self._test_home(self.cls.home()) - - # bpo-38883: ignore `HOME` when set on windows - env['HOME'] = os.path.join(BASE, 'home') - self._test_home(self.cls.home()) - - def test_with_segments(self): + def test_with_segments_abstract(self): class P(_BasePurePathSubclass, self.cls): pass p = P(BASE, session_id=42) - self.assertEqual(42, p.absolute().session_id) - self.assertEqual(42, p.resolve().session_id) - if not is_wasi: # WASI has no user accounts. - self.assertEqual(42, p.with_segments('~').expanduser().session_id) - self.assertEqual(42, (p / 'fileA').rename(p / 'fileB').session_id) - self.assertEqual(42, (p / 'fileB').replace(p / 'fileA').session_id) - if os_helper.can_symlink(): - self.assertEqual(42, (p / 'linkA').readlink().session_id) for path in p.iterdir(): self.assertEqual(42, path.session_id) for path in p.glob('*'): @@ -1695,22 +1619,6 @@ def test_empty_path(self): p = self.cls('') self.assertEqual(p.stat(), os.stat('.')) - @unittest.skipIf(is_wasi, "WASI has no user accounts.") - def test_expanduser_common(self): - P = self.cls - p = P('~') - self.assertEqual(p.expanduser(), P(os.path.expanduser('~'))) - p = P('foo') - self.assertEqual(p.expanduser(), p) - p = P('/~') - self.assertEqual(p.expanduser(), p) - p = P('../~') - self.assertEqual(p.expanduser(), p) - p = P(P('').absolute().anchor) / '~' - self.assertEqual(p.expanduser(), p) - p = P('~/a:b') - self.assertEqual(p.expanduser(), P(os.path.expanduser('~'), './a:b')) - def test_exists(self): P = self.cls p = P(BASE) @@ -1909,6 +1817,345 @@ def test_rglob_symlink_loop(self): } self.assertEqual(given, {p / x for x in expect}) + def test_glob_dotdot(self): + # ".." is not special in globs. + P = self.cls + p = P(BASE) + self.assertEqual(set(p.glob("..")), { P(BASE, "..") }) + self.assertEqual(set(p.glob("../..")), { P(BASE, "..", "..") }) + self.assertEqual(set(p.glob("dirA/..")), { P(BASE, "dirA", "..") }) + self.assertEqual(set(p.glob("dirA/../file*")), { P(BASE, "dirA/../fileA") }) + self.assertEqual(set(p.glob("dirA/../file*/..")), set()) + self.assertEqual(set(p.glob("../xyzzy")), set()) + self.assertEqual(set(p.glob("xyzzy/..")), set()) + self.assertEqual(set(p.glob("/".join([".."] * 50))), { P(BASE, *[".."] * 50)}) + + @os_helper.skip_unless_symlink + def test_glob_permissions(self): + return # FIXME + + # See bpo-38894 + P = self.cls + base = P(BASE) / 'permissions' + base.mkdir() + + file1 = base / "file1" + file1.touch() + file2 = base / "file2" + file2.touch() + + subdir = base / "subdir" + + file3 = base / "file3" + file3.symlink_to(subdir / "other") + + # Patching is needed to avoid relying on the filesystem + # to return the order of the files as the error will not + # happen if the symlink is the last item. + real_scandir = os.scandir + def my_scandir(path): + with real_scandir(path) as scandir_it: + entries = list(scandir_it) + entries.sort(key=lambda entry: entry.name) + return contextlib.nullcontext(entries) + + with mock.patch("os.scandir", my_scandir): + self.assertEqual(len(set(base.glob("*"))), 3) + subdir.mkdir() + self.assertEqual(len(set(base.glob("*"))), 4) + subdir.chmod(000) + self.assertEqual(len(set(base.glob("*"))), 4) + + @os_helper.skip_unless_symlink + def test_stat_no_follow_symlinks(self): + p = self.cls(BASE) / 'linkA' + st = p.stat() + self.assertNotEqual(st, p.stat(follow_symlinks=False)) + + def test_stat_no_follow_symlinks_nosymlink(self): + p = self.cls(BASE) / 'fileA' + st = p.stat() + self.assertEqual(st, p.stat(follow_symlinks=False)) + + @os_helper.skip_unless_symlink + def test_lstat(self): + p = self.cls(BASE)/ 'linkA' + st = p.stat() + self.assertNotEqual(st, p.lstat()) + + def test_lstat_nosymlink(self): + p = self.cls(BASE) / 'fileA' + st = p.stat() + self.assertEqual(st, p.lstat()) + + def test_is_dir(self): + P = self.cls(BASE) + self.assertTrue((P / 'dirA').is_dir()) + self.assertFalse((P / 'fileA').is_dir()) + self.assertFalse((P / 'non-existing').is_dir()) + self.assertFalse((P / 'fileA' / 'bah').is_dir()) + if os_helper.can_symlink(): + self.assertFalse((P / 'linkA').is_dir()) + self.assertTrue((P / 'linkB').is_dir()) + self.assertFalse((P/ 'brokenLink').is_dir(), False) + self.assertIs((P / 'dirA\udfff').is_dir(), False) + self.assertIs((P / 'dirA\x00').is_dir(), False) + + def test_is_file(self): + P = self.cls(BASE) + self.assertTrue((P / 'fileA').is_file()) + self.assertFalse((P / 'dirA').is_file()) + self.assertFalse((P / 'non-existing').is_file()) + self.assertFalse((P / 'fileA' / 'bah').is_file()) + if os_helper.can_symlink(): + self.assertTrue((P / 'linkA').is_file()) + self.assertFalse((P / 'linkB').is_file()) + self.assertFalse((P/ 'brokenLink').is_file()) + self.assertIs((P / 'fileA\udfff').is_file(), False) + self.assertIs((P / 'fileA\x00').is_file(), False) + + def test_is_mount(self): + P = self.cls(BASE) + if os.name == 'nt': + R = self.cls('c:\\') + else: + R = self.cls('/') + self.assertFalse((P / 'fileA').is_mount()) + self.assertFalse((P / 'dirA').is_mount()) + self.assertFalse((P / 'non-existing').is_mount()) + self.assertFalse((P / 'fileA' / 'bah').is_mount()) + self.assertTrue(R.is_mount()) + if os_helper.can_symlink(): + self.assertFalse((P / 'linkA').is_mount()) + self.assertIs((R / '\udfff').is_mount(), False) + + def test_is_symlink(self): + P = self.cls(BASE) + self.assertFalse((P / 'fileA').is_symlink()) + self.assertFalse((P / 'dirA').is_symlink()) + self.assertFalse((P / 'non-existing').is_symlink()) + self.assertFalse((P / 'fileA' / 'bah').is_symlink()) + if os_helper.can_symlink(): + self.assertTrue((P / 'linkA').is_symlink()) + self.assertTrue((P / 'linkB').is_symlink()) + self.assertTrue((P/ 'brokenLink').is_symlink()) + self.assertIs((P / 'fileA\udfff').is_file(), False) + self.assertIs((P / 'fileA\x00').is_file(), False) + if os_helper.can_symlink(): + self.assertIs((P / 'linkA\udfff').is_file(), False) + self.assertIs((P / 'linkA\x00').is_file(), False) + + def test_is_fifo_false(self): + P = self.cls(BASE) + self.assertFalse((P / 'fileA').is_fifo()) + self.assertFalse((P / 'dirA').is_fifo()) + self.assertFalse((P / 'non-existing').is_fifo()) + self.assertFalse((P / 'fileA' / 'bah').is_fifo()) + self.assertIs((P / 'fileA\udfff').is_fifo(), False) + self.assertIs((P / 'fileA\x00').is_fifo(), False) + + @unittest.skipUnless(hasattr(os, "mkfifo"), "os.mkfifo() required") + @unittest.skipIf(sys.platform == "vxworks", + "fifo requires special path on VxWorks") + def test_is_fifo_true(self): + P = self.cls(BASE, 'myfifo') + try: + os.mkfifo(str(P)) + except PermissionError as e: + self.skipTest('os.mkfifo(): %s' % e) + self.assertTrue(P.is_fifo()) + self.assertFalse(P.is_socket()) + self.assertFalse(P.is_file()) + self.assertIs(self.cls(BASE, 'myfifo\udfff').is_fifo(), False) + self.assertIs(self.cls(BASE, 'myfifo\x00').is_fifo(), False) + + def test_is_socket_false(self): + P = self.cls(BASE) + self.assertFalse((P / 'fileA').is_socket()) + self.assertFalse((P / 'dirA').is_socket()) + self.assertFalse((P / 'non-existing').is_socket()) + self.assertFalse((P / 'fileA' / 'bah').is_socket()) + self.assertIs((P / 'fileA\udfff').is_socket(), False) + self.assertIs((P / 'fileA\x00').is_socket(), False) + + @unittest.skipUnless(hasattr(socket, "AF_UNIX"), "Unix sockets required") + @unittest.skipIf( + is_emscripten, "Unix sockets are not implemented on Emscripten." + ) + @unittest.skipIf( + is_wasi, "Cannot create socket on WASI." + ) + def test_is_socket_true(self): + P = self.cls(BASE, 'mysock') + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.addCleanup(sock.close) + try: + sock.bind(str(P)) + except OSError as e: + if (isinstance(e, PermissionError) or + "AF_UNIX path too long" in str(e)): + self.skipTest("cannot bind Unix socket: " + str(e)) + self.assertTrue(P.is_socket()) + self.assertFalse(P.is_fifo()) + self.assertFalse(P.is_file()) + self.assertIs(self.cls(BASE, 'mysock\udfff').is_socket(), False) + self.assertIs(self.cls(BASE, 'mysock\x00').is_socket(), False) + + def test_is_block_device_false(self): + P = self.cls(BASE) + self.assertFalse((P / 'fileA').is_block_device()) + self.assertFalse((P / 'dirA').is_block_device()) + self.assertFalse((P / 'non-existing').is_block_device()) + self.assertFalse((P / 'fileA' / 'bah').is_block_device()) + self.assertIs((P / 'fileA\udfff').is_block_device(), False) + self.assertIs((P / 'fileA\x00').is_block_device(), False) + + def test_is_char_device_false(self): + P = self.cls(BASE) + self.assertFalse((P / 'fileA').is_char_device()) + self.assertFalse((P / 'dirA').is_char_device()) + self.assertFalse((P / 'non-existing').is_char_device()) + self.assertFalse((P / 'fileA' / 'bah').is_char_device()) + self.assertIs((P / 'fileA\udfff').is_char_device(), False) + self.assertIs((P / 'fileA\x00').is_char_device(), False) + + def test_is_char_device_true(self): + # Under Unix, /dev/null should generally be a char device. + P = self.cls('/dev/null') + if not P.exists(): + self.skipTest("/dev/null required") + self.assertTrue(P.is_char_device()) + self.assertFalse(P.is_block_device()) + self.assertFalse(P.is_file()) + self.assertIs(self.cls('/dev/null\udfff').is_char_device(), False) + self.assertIs(self.cls('/dev/null\x00').is_char_device(), False) + + def test_pickling_common(self): + p = self.cls(BASE, 'fileA') + for proto in range(0, pickle.HIGHEST_PROTOCOL + 1): + dumped = pickle.dumps(p, proto) + pp = pickle.loads(dumped) + self.assertEqual(pp.stat(), p.stat()) + + def test_parts_interning(self): + P = self.cls + p = P('/usr/bin/foo') + q = P('/usr/local/bin') + # 'usr' + self.assertIs(p.parts[1], q.parts[1]) + # 'bin' + self.assertIs(p.parts[2], q.parts[3]) + + +class AbstractPathTest(_BaseAbstractPathTest, unittest.TestCase): + class cls(pathlib._AbstractPath): + def stat(self, *, follow_symlinks=True): + return os.stat(self, follow_symlinks=follow_symlinks) + + def open(self, mode='r', buffering=-1, encoding=None, + errors=None, newline=None): + if "b" not in mode: + encoding = io.text_encoding(encoding) + return io.open(self, mode, buffering, encoding, errors, newline) + + def iterdir(self): + for name in os.listdir(self): + yield self._make_child_relpath(name) + + +class _BasePathTest(_BaseAbstractPathTest): + def assertFileNotFound(self, func, *args, **kwargs): + with self.assertRaises(FileNotFoundError) as cm: + func(*args, **kwargs) + self.assertEqual(cm.exception.errno, errno.ENOENT) + + def assertEqualNormCase(self, path_a, path_b): + self.assertEqual(os.path.normcase(path_a), os.path.normcase(path_b)) + + def _test_cwd(self, p): + q = self.cls(os.getcwd()) + self.assertEqual(p, q) + self.assertEqualNormCase(str(p), str(q)) + self.assertIs(type(p), type(q)) + self.assertTrue(p.is_absolute()) + + def test_cwd(self): + p = self.cls.cwd() + self._test_cwd(p) + + def test_absolute_common(self): + P = self.cls + + with mock.patch("os.getcwd") as getcwd: + getcwd.return_value = BASE + + # Simple relative paths. + self.assertEqual(str(P().absolute()), BASE) + self.assertEqual(str(P('.').absolute()), BASE) + self.assertEqual(str(P('a').absolute()), os.path.join(BASE, 'a')) + self.assertEqual(str(P('a', 'b', 'c').absolute()), os.path.join(BASE, 'a', 'b', 'c')) + + # Symlinks should not be resolved. + self.assertEqual(str(P('linkB', 'fileB').absolute()), os.path.join(BASE, 'linkB', 'fileB')) + self.assertEqual(str(P('brokenLink').absolute()), os.path.join(BASE, 'brokenLink')) + self.assertEqual(str(P('brokenLinkLoop').absolute()), os.path.join(BASE, 'brokenLinkLoop')) + + # '..' entries should be preserved and not normalised. + self.assertEqual(str(P('..').absolute()), os.path.join(BASE, '..')) + self.assertEqual(str(P('a', '..').absolute()), os.path.join(BASE, 'a', '..')) + self.assertEqual(str(P('..', 'b').absolute()), os.path.join(BASE, '..', 'b')) + + def test_with_segments(self): + class P(_BasePurePathSubclass, self.cls): + pass + p = P(BASE, session_id=42) + self.assertEqual(42, p.absolute().session_id) + self.assertEqual(42, p.resolve().session_id) + if not is_wasi: # WASI has no user accounts. + self.assertEqual(42, p.with_segments('~').expanduser().session_id) + self.assertEqual(42, (p / 'fileA').rename(p / 'fileB').session_id) + self.assertEqual(42, (p / 'fileB').replace(p / 'fileA').session_id) + if os_helper.can_symlink(): + self.assertEqual(42, (p / 'linkA').readlink().session_id) + + def _test_home(self, p): + q = self.cls(os.path.expanduser('~')) + self.assertEqual(p, q) + self.assertEqualNormCase(str(p), str(q)) + self.assertIs(type(p), type(q)) + self.assertTrue(p.is_absolute()) + + @unittest.skipIf( + pwd is None, reason="Test requires pwd module to get homedir." + ) + def test_home(self): + with os_helper.EnvironmentVarGuard() as env: + self._test_home(self.cls.home()) + + env.clear() + env['USERPROFILE'] = os.path.join(BASE, 'userprofile') + self._test_home(self.cls.home()) + + # bpo-38883: ignore `HOME` when set on windows + env['HOME'] = os.path.join(BASE, 'home') + self._test_home(self.cls.home()) + + @unittest.skipIf(is_wasi, "WASI has no user accounts.") + def test_expanduser_common(self): + P = self.cls + p = P('~') + self.assertEqual(p.expanduser(), P(os.path.expanduser('~'))) + p = P('foo') + self.assertEqual(p.expanduser(), p) + p = P('/~') + self.assertEqual(p.expanduser(), p) + p = P('../~') + self.assertEqual(p.expanduser(), p) + p = P(P('').absolute().anchor) / '~' + self.assertEqual(p.expanduser(), p) + p = P('~/a:b') + self.assertEqual(p.expanduser(), P(os.path.expanduser('~'), './a:b')) + def test_glob_many_open_files(self): depth = 30 P = self.cls @@ -1926,52 +2173,15 @@ def test_glob_many_open_files(self): for it in iters: self.assertEqual(next(it), p) - def test_glob_dotdot(self): - # ".." is not special in globs. - P = self.cls - p = P(BASE) - self.assertEqual(set(p.glob("..")), { P(BASE, "..") }) - self.assertEqual(set(p.glob("../..")), { P(BASE, "..", "..") }) - self.assertEqual(set(p.glob("dirA/..")), { P(BASE, "dirA", "..") }) - self.assertEqual(set(p.glob("dirA/../file*")), { P(BASE, "dirA/../fileA") }) - self.assertEqual(set(p.glob("dirA/../file*/..")), set()) - self.assertEqual(set(p.glob("../xyzzy")), set()) - self.assertEqual(set(p.glob("xyzzy/..")), set()) - self.assertEqual(set(p.glob("/".join([".."] * 50))), { P(BASE, *[".."] * 50)}) - - @os_helper.skip_unless_symlink - def test_glob_permissions(self): - # See bpo-38894 - P = self.cls - base = P(BASE) / 'permissions' - base.mkdir() - - file1 = base / "file1" - file1.touch() - file2 = base / "file2" - file2.touch() - - subdir = base / "subdir" - - file3 = base / "file3" - file3.symlink_to(subdir / "other") - - # Patching is needed to avoid relying on the filesystem - # to return the order of the files as the error will not - # happen if the symlink is the last item. - real_scandir = os.scandir - def my_scandir(path): - with real_scandir(path) as scandir_it: - entries = list(scandir_it) - entries.sort(key=lambda entry: entry.name) - return contextlib.nullcontext(entries) - - with mock.patch("os.scandir", my_scandir): - self.assertEqual(len(set(base.glob("*"))), 3) - subdir.mkdir() - self.assertEqual(len(set(base.glob("*"))), 4) - subdir.chmod(000) - self.assertEqual(len(set(base.glob("*"))), 4) + @os_helper.skip_unless_working_chmod + def test_stat(self): + p = self.cls(BASE) / 'fileA' + st = p.stat() + self.assertEqual(p.stat(), st) + # Change file mode by flipping write bit. + p.chmod(st.st_mode ^ 0o222) + self.addCleanup(p.chmod, st.st_mode) + self.assertNotEqual(p.stat(), st) def _check_resolve(self, p, expected, strict=True): q = p.resolve(strict) @@ -2116,37 +2326,6 @@ def test_chmod_follow_symlinks_true(self): # XXX also need a test for lchmod. - @os_helper.skip_unless_working_chmod - def test_stat(self): - p = self.cls(BASE) / 'fileA' - st = p.stat() - self.assertEqual(p.stat(), st) - # Change file mode by flipping write bit. - p.chmod(st.st_mode ^ 0o222) - self.addCleanup(p.chmod, st.st_mode) - self.assertNotEqual(p.stat(), st) - - @os_helper.skip_unless_symlink - def test_stat_no_follow_symlinks(self): - p = self.cls(BASE) / 'linkA' - st = p.stat() - self.assertNotEqual(st, p.stat(follow_symlinks=False)) - - def test_stat_no_follow_symlinks_nosymlink(self): - p = self.cls(BASE) / 'fileA' - st = p.stat() - self.assertEqual(st, p.stat(follow_symlinks=False)) - - @os_helper.skip_unless_symlink - def test_lstat(self): - p = self.cls(BASE)/ 'linkA' - st = p.stat() - self.assertNotEqual(st, p.lstat()) - - def test_lstat_nosymlink(self): - p = self.cls(BASE) / 'fileA' - st = p.stat() - self.assertEqual(st, p.lstat()) @unittest.skipUnless(pwd, "the pwd module is needed for this test") def test_owner(self): @@ -2451,63 +2630,6 @@ def test_symlink_to(self): self.assertTrue(link.is_dir()) self.assertTrue(list(link.iterdir())) - def test_is_dir(self): - P = self.cls(BASE) - self.assertTrue((P / 'dirA').is_dir()) - self.assertFalse((P / 'fileA').is_dir()) - self.assertFalse((P / 'non-existing').is_dir()) - self.assertFalse((P / 'fileA' / 'bah').is_dir()) - if os_helper.can_symlink(): - self.assertFalse((P / 'linkA').is_dir()) - self.assertTrue((P / 'linkB').is_dir()) - self.assertFalse((P/ 'brokenLink').is_dir(), False) - self.assertIs((P / 'dirA\udfff').is_dir(), False) - self.assertIs((P / 'dirA\x00').is_dir(), False) - - def test_is_file(self): - P = self.cls(BASE) - self.assertTrue((P / 'fileA').is_file()) - self.assertFalse((P / 'dirA').is_file()) - self.assertFalse((P / 'non-existing').is_file()) - self.assertFalse((P / 'fileA' / 'bah').is_file()) - if os_helper.can_symlink(): - self.assertTrue((P / 'linkA').is_file()) - self.assertFalse((P / 'linkB').is_file()) - self.assertFalse((P/ 'brokenLink').is_file()) - self.assertIs((P / 'fileA\udfff').is_file(), False) - self.assertIs((P / 'fileA\x00').is_file(), False) - - def test_is_mount(self): - P = self.cls(BASE) - if os.name == 'nt': - R = self.cls('c:\\') - else: - R = self.cls('/') - self.assertFalse((P / 'fileA').is_mount()) - self.assertFalse((P / 'dirA').is_mount()) - self.assertFalse((P / 'non-existing').is_mount()) - self.assertFalse((P / 'fileA' / 'bah').is_mount()) - self.assertTrue(R.is_mount()) - if os_helper.can_symlink(): - self.assertFalse((P / 'linkA').is_mount()) - self.assertIs((R / '\udfff').is_mount(), False) - - def test_is_symlink(self): - P = self.cls(BASE) - self.assertFalse((P / 'fileA').is_symlink()) - self.assertFalse((P / 'dirA').is_symlink()) - self.assertFalse((P / 'non-existing').is_symlink()) - self.assertFalse((P / 'fileA' / 'bah').is_symlink()) - if os_helper.can_symlink(): - self.assertTrue((P / 'linkA').is_symlink()) - self.assertTrue((P / 'linkB').is_symlink()) - self.assertTrue((P/ 'brokenLink').is_symlink()) - self.assertIs((P / 'fileA\udfff').is_file(), False) - self.assertIs((P / 'fileA\x00').is_file(), False) - if os_helper.can_symlink(): - self.assertIs((P / 'linkA\udfff').is_file(), False) - self.assertIs((P / 'linkA\x00').is_file(), False) - def test_is_junction(self): P = self.cls(BASE) @@ -2515,107 +2637,6 @@ def test_is_junction(self): self.assertEqual(P.is_junction(), P._flavour.isjunction.return_value) P._flavour.isjunction.assert_called_once_with(P) - def test_is_fifo_false(self): - P = self.cls(BASE) - self.assertFalse((P / 'fileA').is_fifo()) - self.assertFalse((P / 'dirA').is_fifo()) - self.assertFalse((P / 'non-existing').is_fifo()) - self.assertFalse((P / 'fileA' / 'bah').is_fifo()) - self.assertIs((P / 'fileA\udfff').is_fifo(), False) - self.assertIs((P / 'fileA\x00').is_fifo(), False) - - @unittest.skipUnless(hasattr(os, "mkfifo"), "os.mkfifo() required") - @unittest.skipIf(sys.platform == "vxworks", - "fifo requires special path on VxWorks") - def test_is_fifo_true(self): - P = self.cls(BASE, 'myfifo') - try: - os.mkfifo(str(P)) - except PermissionError as e: - self.skipTest('os.mkfifo(): %s' % e) - self.assertTrue(P.is_fifo()) - self.assertFalse(P.is_socket()) - self.assertFalse(P.is_file()) - self.assertIs(self.cls(BASE, 'myfifo\udfff').is_fifo(), False) - self.assertIs(self.cls(BASE, 'myfifo\x00').is_fifo(), False) - - def test_is_socket_false(self): - P = self.cls(BASE) - self.assertFalse((P / 'fileA').is_socket()) - self.assertFalse((P / 'dirA').is_socket()) - self.assertFalse((P / 'non-existing').is_socket()) - self.assertFalse((P / 'fileA' / 'bah').is_socket()) - self.assertIs((P / 'fileA\udfff').is_socket(), False) - self.assertIs((P / 'fileA\x00').is_socket(), False) - - @unittest.skipUnless(hasattr(socket, "AF_UNIX"), "Unix sockets required") - @unittest.skipIf( - is_emscripten, "Unix sockets are not implemented on Emscripten." - ) - @unittest.skipIf( - is_wasi, "Cannot create socket on WASI." - ) - def test_is_socket_true(self): - P = self.cls(BASE, 'mysock') - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - self.addCleanup(sock.close) - try: - sock.bind(str(P)) - except OSError as e: - if (isinstance(e, PermissionError) or - "AF_UNIX path too long" in str(e)): - self.skipTest("cannot bind Unix socket: " + str(e)) - self.assertTrue(P.is_socket()) - self.assertFalse(P.is_fifo()) - self.assertFalse(P.is_file()) - self.assertIs(self.cls(BASE, 'mysock\udfff').is_socket(), False) - self.assertIs(self.cls(BASE, 'mysock\x00').is_socket(), False) - - def test_is_block_device_false(self): - P = self.cls(BASE) - self.assertFalse((P / 'fileA').is_block_device()) - self.assertFalse((P / 'dirA').is_block_device()) - self.assertFalse((P / 'non-existing').is_block_device()) - self.assertFalse((P / 'fileA' / 'bah').is_block_device()) - self.assertIs((P / 'fileA\udfff').is_block_device(), False) - self.assertIs((P / 'fileA\x00').is_block_device(), False) - - def test_is_char_device_false(self): - P = self.cls(BASE) - self.assertFalse((P / 'fileA').is_char_device()) - self.assertFalse((P / 'dirA').is_char_device()) - self.assertFalse((P / 'non-existing').is_char_device()) - self.assertFalse((P / 'fileA' / 'bah').is_char_device()) - self.assertIs((P / 'fileA\udfff').is_char_device(), False) - self.assertIs((P / 'fileA\x00').is_char_device(), False) - - def test_is_char_device_true(self): - # Under Unix, /dev/null should generally be a char device. - P = self.cls('/dev/null') - if not P.exists(): - self.skipTest("/dev/null required") - self.assertTrue(P.is_char_device()) - self.assertFalse(P.is_block_device()) - self.assertFalse(P.is_file()) - self.assertIs(self.cls('/dev/null\udfff').is_char_device(), False) - self.assertIs(self.cls('/dev/null\x00').is_char_device(), False) - - def test_pickling_common(self): - p = self.cls(BASE, 'fileA') - for proto in range(0, pickle.HIGHEST_PROTOCOL + 1): - dumped = pickle.dumps(p, proto) - pp = pickle.loads(dumped) - self.assertEqual(pp.stat(), p.stat()) - - def test_parts_interning(self): - P = self.cls - p = P('/usr/bin/foo') - q = P('/usr/local/bin') - # 'usr' - self.assertIs(p.parts[1], q.parts[1]) - # 'bin' - self.assertIs(p.parts[2], q.parts[3]) - def _check_complex_symlinks(self, link0_target): # Test solving a non-looping chain of symlinks (issue #19887). P = self.cls(BASE)