Align filesystem archives and bytecode compilation with Python 3.15 (#9191)

Refresh the connected pathlib, glob, zipfile, tempfile, py_compile and compileall closure from CPython 3.15.0 while preserving existing RustPython test guards.

Assisted-by: OpenAI Codex:model version unavailable

Lib/compileall.py

@@ -165,6 +165,14 @@ def compile_file(fullname, ddir=None, force=False, rx=None, quiet=0,
     stripdir = os.fspath(stripdir) if stripdir is not None else None
     name = os.path.basename(fullname)
 
+    # Without a cache_tag, we can only create legacy .pyc files. None of our
+    # callers seem to expect this, so the best we can do is fail without raising
+    if not legacy and sys.implementation.cache_tag is None:
+        if not quiet:
+            print("No cache tag is available to generate .pyc path for",
+                  repr(fullname))
+        return False
+
     dfile = None
 
     if ddir is not None:
@@ -223,7 +231,7 @@ def compile_file(fullname, ddir=None, force=False, rx=None, quiet=0,
                     cfile = importlib.util.cache_from_source(fullname)
                     opt_cfiles[opt_level] = cfile
 
-        head, tail = name[:-3], name[-3:]
+        tail = name[-3:]
         if tail == '.py':
             if not force:
                 try:
@@ -318,7 +326,6 @@ def main():
 
     parser = argparse.ArgumentParser(
         description='Utilities to support installing Python libraries.',
-        color=True,
     )
     parser.add_argument('-l', action='store_const', const=0,
                         default=None, dest='maxlevels',
@@ -330,10 +337,10 @@ def main():
     parser.add_argument('-f', action='store_true', dest='force',
                         help='force rebuild even if timestamps are up to date')
     parser.add_argument('-q', action='count', dest='quiet', default=0,
-                        help='output only error messages; -qq will suppress '
+                        help='output only error messages; `-qq` will suppress '
                              'the error messages as well.')
     parser.add_argument('-b', action='store_true', dest='legacy',
-                        help='use legacy (pre-PEP3147) compiled file locations')
+                        help='use legacy (pre-PEP 3147) compiled file locations')
     parser.add_argument('-d', metavar='DESTDIR',  dest='ddir', default=None,
                         help=('directory to prepend to file paths for use in '
                               'compile-time tracebacks and in runtime '
@@ -359,28 +366,28 @@ def main():
                               'of each file considered for compilation'))
     parser.add_argument('-i', metavar='FILE', dest='flist',
                         help=('add all the files and directories listed in '
-                              'FILE to the list considered for compilation; '
-                              'if "-", names are read from stdin'))
+                              '`FILE` to the list considered for compilation; '
+                              'if `"-"`, names are read from `stdin`'))
     parser.add_argument('compile_dest', metavar='FILE|DIR', nargs='*',
                         help=('zero or more file and directory names '
                               'to compile; if no arguments given, defaults '
-                              'to the equivalent of -l sys.path'))
+                              'to the equivalent of `-l` `sys.path`'))
     parser.add_argument('-j', '--workers', default=1,
                         type=int, help='Run compileall concurrently')
     invalidation_modes = [mode.name.lower().replace('_', '-')
                           for mode in py_compile.PycInvalidationMode]
     parser.add_argument('--invalidation-mode',
                         choices=sorted(invalidation_modes),
-                        help=('set .pyc invalidation mode; defaults to '
-                              '"checked-hash" if the SOURCE_DATE_EPOCH '
+                        help=('set `.pyc` invalidation mode; defaults to '
+                              '`"checked-hash"` if the `SOURCE_DATE_EPOCH` '
                               'environment variable is set, and '
-                              '"timestamp" otherwise.'))
+                              '`"timestamp"` otherwise.'))
     parser.add_argument('-o', action='append', type=int, dest='opt_levels',
                         help=('Optimization levels to run compilation with. '
-                              'Default is -1 which uses the optimization level '
-                              'of the Python interpreter itself (see -O).'))
+                              'Default is `-1` which uses the optimization level '
+                              'of the Python interpreter itself (see `-O`).'))
     parser.add_argument('-e', metavar='DIR', dest='limit_sl_dest',
-                        help='Ignore symlinks pointing outsite of the DIR')
+                        help='Ignore symlinks pointing outsite of the `DIR`')
     parser.add_argument('--hardlink-dupes', action='store_true',
                         dest='hardlink_dupes',
                         help='Hardlink duplicated pyc files')

Lib/glob.py

@@ -149,21 +149,6 @@ def _glob0(dirname, basename, dir_fd, dironly, include_hidden=False):
             return [basename]
     return []
 
-_deprecated_function_message = (
-    "{name} is deprecated and will be removed in Python {remove}. Use "
-    "glob.glob and pass a directory to its root_dir argument instead."
-)
-
-def glob0(dirname, pattern):
-    import warnings
-    warnings._deprecated("glob.glob0", _deprecated_function_message, remove=(3, 15))
-    return _glob0(dirname, pattern, None, False)
-
-def glob1(dirname, pattern):
-    import warnings
-    warnings._deprecated("glob.glob1", _deprecated_function_message, remove=(3, 15))
-    return _glob1(dirname, pattern, None, False)
-
 # This helper function recursively yields relative pathnames inside a literal
 # directory.
 
@@ -385,6 +370,12 @@ class _GlobberBase:
         """
         raise NotImplementedError
 
+    @staticmethod
+    def stringify_path(path):
+        """Converts the path to a string object
+        """
+        raise NotImplementedError
+
     # High-level methods
 
     def compile(self, pat, altsep=None):
@@ -493,8 +484,9 @@ class _GlobberBase:
         select_next = self.selector(parts)
 
         def select_recursive(path, exists=False):
-            match_pos = len(str(path))
-            if match is None or match(str(path), match_pos):
+            path_str = self.stringify_path(path)
+            match_pos = len(path_str)
+            if match is None or match(path_str, match_pos):
                 yield from select_next(path, exists)
             stack = [path]
             while stack:
@@ -516,7 +508,7 @@ class _GlobberBase:
                         pass
 
                     if is_dir or not dir_only:
-                        entry_path_str = str(entry_path)
+                        entry_path_str = self.stringify_path(entry_path)
                         if dir_only:
                             entry_path = self.concat_path(entry_path, self.sep)
                         if match is None or match(entry_path_str, match_pos):
@@ -556,19 +548,6 @@ class _StringGlobber(_GlobberBase):
             entries = list(scandir_it)
         return ((entry, entry.name, entry.path) for entry in entries)
 
-
-class _PathGlobber(_GlobberBase):
-    """Provides shell-style pattern matching and globbing for pathlib paths.
-    """
-
     @staticmethod
-    def lexists(path):
-        return path.info.exists(follow_symlinks=False)
-
-    @staticmethod
-    def scandir(path):
-        return ((child.info, child.name, child) for child in path.iterdir())
-
-    @staticmethod
-    def concat_path(path, text):
-        return path.with_segments(str(path) + text)
+    def stringify_path(path):
+        return path  # Already a string.

Lib/pathlib/__init__.py

@@ -12,10 +12,13 @@ import os
 import posixpath
 import sys
 from errno import *
-from glob import _StringGlobber, _no_recurse_symlinks
 from itertools import chain
-from stat import S_ISDIR, S_ISREG, S_ISSOCK, S_ISBLK, S_ISCHR, S_ISFIFO
+from stat import (
+    S_IMODE, S_ISDIR, S_ISREG, S_ISLNK, S_ISSOCK, S_ISBLK, S_ISCHR, S_ISFIFO,
+)
 from _collections_abc import Sequence
+lazy import shutil
+lazy from glob import _StringGlobber, _no_recurse_symlinks
 
 try:
     import pwd
@@ -27,9 +30,9 @@ except ImportError:
     grp = None
 
 from pathlib._os import (
-    PathInfo, DirEntryInfo,
+    vfsopen, vfspath,
     ensure_different_files, ensure_distinct_paths,
-    copyfile2, copyfileobj, magic_open, copy_info,
+    copyfile2, copyfileobj,
 )
 
 
@@ -187,9 +190,6 @@ class PurePath:
     def __repr__(self):
         return "{}({!r})".format(self.__class__.__name__, self.as_posix())
 
-    def __fspath__(self):
-        return str(self)
-
     def __bytes__(self):
         """Return the bytes representation of the path.  This is only
         recommended to use under Unix."""
@@ -258,6 +258,9 @@ class PurePath:
                                                   self._tail) or '.'
             return self._str
 
+    __fspath__ = __str__
+    __vfspath__ = __str__
+
     @classmethod
     def _format_parsed_parts(cls, drv, root, tail):
         if drv or root:
@@ -483,16 +486,19 @@ class PurePath:
         """
         if not hasattr(other, 'with_segments'):
             other = self.with_segments(other)
-        for step, path in enumerate(chain([other], other.parents)):
+        parts = []
+        for path in chain([other], other.parents):
             if path == self or path in self.parents:
                 break
             elif not walk_up:
                 raise ValueError(f"{str(self)!r} is not in the subpath of {str(other)!r}")
             elif path.name == '..':
                 raise ValueError(f"'..' segment in {str(other)!r} cannot be walked")
+            else:
+                parts.append('..')
         else:
             raise ValueError(f"{str(self)!r} and {str(other)!r} have different anchors")
-        parts = ['..'] * step + self._tail[len(path._tail):]
+        parts.extend(self._tail[len(path._tail):])
         return self._from_parsed_parts('', '', parts)
 
     def is_relative_to(self, other):
@@ -513,18 +519,6 @@ class PurePath:
             return False
         return self.parser.isabs(self)
 
-    def is_reserved(self):
-        """Return True if the path contains one of the special names reserved
-        by the system, if any."""
-        import warnings
-        msg = ("pathlib.PurePath.is_reserved() is deprecated and scheduled "
-               "for removal in Python 3.15. Use os.path.isreserved() to "
-               "detect reserved paths on Windows.")
-        warnings._deprecated("pathlib.PurePath.is_reserved", msg, remove=(3, 15))
-        if self.parser is ntpath:
-            return self.parser.isreserved(self)
-        return False
-
     def as_uri(self):
         """Return the path as a URI."""
         import warnings
@@ -618,6 +612,211 @@ class PureWindowsPath(PurePath):
     __slots__ = ()
 
 
+_STAT_RESULT_ERROR = []  # falsy sentinel indicating stat() failed.
+
+
+class _Info:
+    """Implementation of pathlib.types.PathInfo that provides status
+    information by querying a wrapped os.stat_result object. Don't try to
+    construct it yourself."""
+    __slots__ = ('_path', '_entry', '_stat_result', '_lstat_result')
+
+    def __init__(self, path, entry=None):
+        self._path = path
+        self._entry = entry
+        self._stat_result = None
+        self._lstat_result = None
+
+    def __repr__(self):
+        path_type = "WindowsPath" if os.name == "nt" else "PosixPath"
+        return f"<{path_type}.info>"
+
+    def _stat(self, *, follow_symlinks=True):
+        """Return the status as an os.stat_result."""
+        if self._entry:
+            return self._entry.stat(follow_symlinks=follow_symlinks)
+        if follow_symlinks:
+            if not self._stat_result:
+                try:
+                    self._stat_result = os.stat(self._path)
+                except (OSError, ValueError):
+                    self._stat_result = _STAT_RESULT_ERROR
+                    raise
+            return self._stat_result
+        else:
+            if not self._lstat_result:
+                try:
+                    self._lstat_result = os.lstat(self._path)
+                except (OSError, ValueError):
+                    self._lstat_result = _STAT_RESULT_ERROR
+                    raise
+            return self._lstat_result
+
+    def exists(self, *, follow_symlinks=True):
+        """Whether this path exists."""
+        if self._entry:
+            if not follow_symlinks:
+                return True
+        if follow_symlinks:
+            if self._stat_result is _STAT_RESULT_ERROR:
+                return False
+        else:
+            if self._lstat_result is _STAT_RESULT_ERROR:
+                return False
+        try:
+            self._stat(follow_symlinks=follow_symlinks)
+        except (OSError, ValueError):
+            return False
+        return True
+
+    def is_dir(self, *, follow_symlinks=True):
+        """Whether this path is a directory."""
+        if self._entry:
+            try:
+                return self._entry.is_dir(follow_symlinks=follow_symlinks)
+            except OSError:
+                return False
+        if follow_symlinks:
+            if self._stat_result is _STAT_RESULT_ERROR:
+                return False
+        else:
+            if self._lstat_result is _STAT_RESULT_ERROR:
+                return False
+        try:
+            st = self._stat(follow_symlinks=follow_symlinks)
+        except (OSError, ValueError):
+            return False
+        return S_ISDIR(st.st_mode)
+
+    def is_file(self, *, follow_symlinks=True):
+        """Whether this path is a regular file."""
+        if self._entry:
+            try:
+                return self._entry.is_file(follow_symlinks=follow_symlinks)
+            except OSError:
+                return False
+        if follow_symlinks:
+            if self._stat_result is _STAT_RESULT_ERROR:
+                return False
+        else:
+            if self._lstat_result is _STAT_RESULT_ERROR:
+                return False
+        try:
+            st = self._stat(follow_symlinks=follow_symlinks)
+        except (OSError, ValueError):
+            return False
+        return S_ISREG(st.st_mode)
+
+    def is_symlink(self):
+        """Whether this path is a symbolic link."""
+        if self._entry:
+            try:
+                return self._entry.is_symlink()
+            except OSError:
+                return False
+        if self._lstat_result is _STAT_RESULT_ERROR:
+            return False
+        try:
+            st = self._stat(follow_symlinks=False)
+        except (OSError, ValueError):
+            return False
+        return S_ISLNK(st.st_mode)
+
+    def _posix_permissions(self, *, follow_symlinks=True):
+        """Return the POSIX file permissions."""
+        return S_IMODE(self._stat(follow_symlinks=follow_symlinks).st_mode)
+
+    def _file_id(self, *, follow_symlinks=True):
+        """Returns the identifier of the file."""
+        st = self._stat(follow_symlinks=follow_symlinks)
+        return st.st_dev, st.st_ino
+
+    def _access_time_ns(self, *, follow_symlinks=True):
+        """Return the access time in nanoseconds."""
+        return self._stat(follow_symlinks=follow_symlinks).st_atime_ns
+
+    def _mod_time_ns(self, *, follow_symlinks=True):
+        """Return the modify time in nanoseconds."""
+        return self._stat(follow_symlinks=follow_symlinks).st_mtime_ns
+
+    if hasattr(os.stat_result, 'st_flags'):
+        def _bsd_flags(self, *, follow_symlinks=True):
+            """Return the flags."""
+            return self._stat(follow_symlinks=follow_symlinks).st_flags
+
+    if hasattr(os, 'listxattr'):
+        def _xattrs(self, *, follow_symlinks=True):
+            """Return the xattrs as a list of (attr, value) pairs, or an empty
+            list if extended attributes aren't supported."""
+            try:
+                return [
+                    (attr, os.getxattr(self._path, attr, follow_symlinks=follow_symlinks))
+                    for attr in os.listxattr(self._path, follow_symlinks=follow_symlinks)]
+            except OSError as err:
+                if err.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):
+                    raise
+                return []
+
+
+def _copy_info(info, target, follow_symlinks=True):
+    """Copy metadata from the given PathInfo to the given local path."""
+    copy_times_ns = (
+        hasattr(info, '_access_time_ns') and
+        hasattr(info, '_mod_time_ns') and
+        (follow_symlinks or os.utime in os.supports_follow_symlinks))
+    if copy_times_ns:
+        t0 = info._access_time_ns(follow_symlinks=follow_symlinks)
+        t1 = info._mod_time_ns(follow_symlinks=follow_symlinks)
+        os.utime(target, ns=(t0, t1), follow_symlinks=follow_symlinks)
+
+    # We must copy extended attributes before the file is (potentially)
+    # chmod()'ed read-only, otherwise setxattr() will error with -EACCES.
+    copy_xattrs = (
+        hasattr(info, '_xattrs') and
+        hasattr(os, 'setxattr') and
+        (follow_symlinks or os.setxattr in os.supports_follow_symlinks))
+    if copy_xattrs:
+        xattrs = info._xattrs(follow_symlinks=follow_symlinks)
+        for attr, value in xattrs:
+            try:
+                os.setxattr(target, attr, value, follow_symlinks=follow_symlinks)
+            except OSError as e:
+                if e.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):
+                    raise
+
+    copy_posix_permissions = (
+        hasattr(info, '_posix_permissions') and
+        (follow_symlinks or os.chmod in os.supports_follow_symlinks))
+    if copy_posix_permissions:
+        posix_permissions = info._posix_permissions(follow_symlinks=follow_symlinks)
+        try:
+            os.chmod(target, posix_permissions, follow_symlinks=follow_symlinks)
+        except NotImplementedError:
+            # if we got a NotImplementedError, it's because
+            #   * follow_symlinks=False,
+            #   * lchown() is unavailable, and
+            #   * either
+            #       * fchownat() is unavailable or
+            #       * fchownat() doesn't implement AT_SYMLINK_NOFOLLOW.
+            #         (it returned ENOSUP.)
+            # therefore we're out of options--we simply cannot chown the
+            # symlink.  give up, suppress the error.
+            # (which is what shutil always did in this circumstance.)
+            pass
+
+    copy_bsd_flags = (
+        hasattr(info, '_bsd_flags') and
+        hasattr(os, 'chflags') and
+        (follow_symlinks or os.chflags in os.supports_follow_symlinks))
+    if copy_bsd_flags:
+        bsd_flags = info._bsd_flags(follow_symlinks=follow_symlinks)
+        try:
+            os.chflags(target, bsd_flags, follow_symlinks=follow_symlinks)
+        except OSError as why:
+            if why.errno not in (EOPNOTSUPP, ENOTSUP):
+                raise
+
+
 class Path(PurePath):
     """PurePath subclass that can make system calls.
 
@@ -643,7 +842,7 @@ class Path(PurePath):
         try:
             return self._info
         except AttributeError:
-            self._info = PathInfo(self)
+            self._info = _Info(str(self))
             return self._info
 
     def stat(self, *, follow_symlinks=True):
@@ -823,7 +1022,7 @@ class Path(PurePath):
     def _from_dir_entry(self, dir_entry, path_str):
         path = self.with_segments(path_str)
         path._str = path_str
-        path._info = DirEntryInfo(dir_entry)
+        path._info = _Info(dir_entry.path, dir_entry)
         return path
 
     def iterdir(self):
@@ -1003,7 +1202,7 @@ class Path(PurePath):
         fd = os.open(self, flags, mode)
         os.close(fd)
 
-    def mkdir(self, mode=0o777, parents=False, exist_ok=False):
+    def mkdir(self, mode=0o777, parents=False, exist_ok=False, *, parent_mode=None):
         """
         Create a new directory at this given path.
         """
@@ -1012,7 +1211,11 @@ class Path(PurePath):
         except FileNotFoundError:
             if not parents or self.parent == self:
                 raise
-            self.parent.mkdir(parents=True, exist_ok=True)
+            if parent_mode is not None:
+                self.parent.mkdir(mode=parent_mode, parents=True, exist_ok=True,
+                                  parent_mode=parent_mode)
+            else:
+                self.parent.mkdir(parents=True, exist_ok=True)
             self.mkdir(mode, parents=False, exist_ok=exist_ok)
         except OSError:
             # Cannot rely on checking for EEXIST, since the operating system
@@ -1057,8 +1260,6 @@ class Path(PurePath):
         if self.is_symlink() or self.is_junction():
             self.unlink()
         elif self.is_dir():
-            # Lazy import to improve module import time
-            import shutil
             shutil.rmtree(self)
         else:
             self.unlink()
@@ -1129,17 +1330,17 @@ class Path(PurePath):
                 self.joinpath(child.name)._copy_from(
                     child, follow_symlinks, preserve_metadata)
             if preserve_metadata:
-                copy_info(source.info, self)
+                _copy_info(source.info, self)
         else:
             self._copy_from_file(source, preserve_metadata)
 
     def _copy_from_file(self, source, preserve_metadata=False):
         ensure_different_files(source, self)
-        with magic_open(source, 'rb') as source_f:
+        with vfsopen(source, 'rb') as source_f:
             with open(self, 'wb') as target_f:
                 copyfileobj(source_f, target_f)
         if preserve_metadata:
-            copy_info(source.info, self)
+            _copy_info(source.info, self)
 
     if copyfile2:
         # Use fast OS routine for local file copying where available.
@@ -1159,14 +1360,14 @@ class Path(PurePath):
         # os.symlink() incorrectly creates a file-symlink on Windows. Avoid
         # this by passing *target_is_dir* to os.symlink() on Windows.
         def _copy_from_symlink(self, source, preserve_metadata=False):
-            os.symlink(str(source.readlink()), self, source.info.is_dir())
+            os.symlink(vfspath(source.readlink()), self, source.info.is_dir())
             if preserve_metadata:
-                copy_info(source.info, self, follow_symlinks=False)
+                _copy_info(source.info, self, follow_symlinks=False)
     else:
         def _copy_from_symlink(self, source, preserve_metadata=False):
-            os.symlink(str(source.readlink()), self)
+            os.symlink(vfspath(source.readlink()), self)
             if preserve_metadata:
-                copy_info(source.info, self, follow_symlinks=False)
+                _copy_info(source.info, self, follow_symlinks=False)
 
     def move(self, target):
         """

Lib/pathlib/_os.py

@@ -4,7 +4,6 @@ Low-level OS functionality wrappers used by pathlib.
 
 from errno import *
 from io import TextIOWrapper, text_encoding
-from stat import S_ISDIR, S_ISREG, S_ISLNK, S_IMODE
 import os
 import sys
 try:
@@ -166,48 +165,100 @@ def copyfileobj(source_f, target_f):
         write_target(buf)
 
 
-def magic_open(path, mode='r', buffering=-1, encoding=None, errors=None,
-               newline=None):
+def _open_reader(obj):
+    cls = type(obj)
+    try:
+        open_reader = cls.__open_reader__
+    except AttributeError:
+        cls_name = cls.__name__
+        raise TypeError(f"{cls_name} can't be opened for reading") from None
+    else:
+        return open_reader(obj)
+
+
+def _open_writer(obj, mode):
+    cls = type(obj)
+    try:
+        open_writer = cls.__open_writer__
+    except AttributeError:
+        cls_name = cls.__name__
+        raise TypeError(f"{cls_name} can't be opened for writing") from None
+    else:
+        return open_writer(obj, mode)
+
+
+def _open_updater(obj, mode):
+    cls = type(obj)
+    try:
+        open_updater = cls.__open_updater__
+    except AttributeError:
+        cls_name = cls.__name__
+        raise TypeError(f"{cls_name} can't be opened for updating") from None
+    else:
+        return open_updater(obj, mode)
+
+
+def vfsopen(obj, mode='r', buffering=-1, encoding=None, errors=None,
+            newline=None):
     """
     Open the file pointed to by this path and return a file object, as
     the built-in open() function does.
+
+    Unlike the built-in open() function, this function additionally accepts
+    'openable' objects, which are objects with any of these special methods:
+
+        __open_reader__()
+        __open_writer__(mode)
+        __open_updater__(mode)
+
+    '__open_reader__' is called for 'r' mode; '__open_writer__' for 'a', 'w'
+    and 'x' modes; and '__open_updater__' for 'r+' and 'w+' modes. If text
+    mode is requested, the result is wrapped in an io.TextIOWrapper object.
     """
+    if buffering != -1:
+        raise ValueError("buffer size can't be customized")
     text = 'b' not in mode
     if text:
         # Call io.text_encoding() here to ensure any warning is raised at an
         # appropriate stack level.
         encoding = text_encoding(encoding)
     try:
-        return open(path, mode, buffering, encoding, errors, newline)
+        return open(obj, mode, buffering, encoding, errors, newline)
     except TypeError:
         pass
-    cls = type(path)
+    if not text:
+        if encoding is not None:
+            raise ValueError("binary mode doesn't take an encoding argument")
+        if errors is not None:
+            raise ValueError("binary mode doesn't take an errors argument")
+        if newline is not None:
+            raise ValueError("binary mode doesn't take a newline argument")
     mode = ''.join(sorted(c for c in mode if c not in 'bt'))
-    if text:
-        try:
-            attr = getattr(cls, f'__open_{mode}__')
-        except AttributeError:
-            pass
-        else:
-            return attr(path, buffering, encoding, errors, newline)
-    elif encoding is not None:
-        raise ValueError("binary mode doesn't take an encoding argument")
-    elif errors is not None:
-        raise ValueError("binary mode doesn't take an errors argument")
-    elif newline is not None:
-        raise ValueError("binary mode doesn't take a newline argument")
-
-    try:
-        attr = getattr(cls, f'__open_{mode}b__')
-    except AttributeError:
-        pass
+    if mode == 'r':
+        stream = _open_reader(obj)
+    elif mode in ('a', 'w', 'x'):
+        stream = _open_writer(obj, mode)
+    elif mode in ('+r', '+w'):
+        stream = _open_updater(obj, mode[1])
     else:
-        stream = attr(path, buffering)
-        if text:
-            stream = TextIOWrapper(stream, encoding, errors, newline)
-        return stream
+        raise ValueError(f'invalid mode: {mode}')
+    if text:
+        stream = TextIOWrapper(stream, encoding, errors, newline)
+    return stream
 
-    raise TypeError(f"{cls.__name__} can't be opened with mode {mode!r}")
+
+def vfspath(obj):
+    """
+    Return the string representation of a virtual path object.
+    """
+    cls = type(obj)
+    try:
+        vfspath_method = cls.__vfspath__
+    except AttributeError:
+        cls_name = cls.__name__
+        raise TypeError(f"expected JoinablePath object, not {cls_name}") from None
+    else:
+        return vfspath_method(obj)
 
 
 def ensure_distinct_paths(source, target):
@@ -225,8 +276,8 @@ def ensure_distinct_paths(source, target):
         err = OSError(EINVAL, "Source path is a parent of target path")
     else:
         return
-    err.filename = str(source)
-    err.filename2 = str(target)
+    err.filename = vfspath(source)
+    err.filename2 = vfspath(target)
     raise err
 
 
@@ -247,284 +298,6 @@ def ensure_different_files(source, target):
         except (OSError, ValueError):
             return
     err = OSError(EINVAL, "Source and target are the same file")
-    err.filename = str(source)
-    err.filename2 = str(target)
+    err.filename = vfspath(source)
+    err.filename2 = vfspath(target)
     raise err
-
-
-def copy_info(info, target, follow_symlinks=True):
-    """Copy metadata from the given PathInfo to the given local path."""
-    copy_times_ns = (
-        hasattr(info, '_access_time_ns') and
-        hasattr(info, '_mod_time_ns') and
-        (follow_symlinks or os.utime in os.supports_follow_symlinks))
-    if copy_times_ns:
-        t0 = info._access_time_ns(follow_symlinks=follow_symlinks)
-        t1 = info._mod_time_ns(follow_symlinks=follow_symlinks)
-        os.utime(target, ns=(t0, t1), follow_symlinks=follow_symlinks)
-
-    # We must copy extended attributes before the file is (potentially)
-    # chmod()'ed read-only, otherwise setxattr() will error with -EACCES.
-    copy_xattrs = (
-        hasattr(info, '_xattrs') and
-        hasattr(os, 'setxattr') and
-        (follow_symlinks or os.setxattr in os.supports_follow_symlinks))
-    if copy_xattrs:
-        xattrs = info._xattrs(follow_symlinks=follow_symlinks)
-        for attr, value in xattrs:
-            try:
-                os.setxattr(target, attr, value, follow_symlinks=follow_symlinks)
-            except OSError as e:
-                if e.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):
-                    raise
-
-    copy_posix_permissions = (
-        hasattr(info, '_posix_permissions') and
-        (follow_symlinks or os.chmod in os.supports_follow_symlinks))
-    if copy_posix_permissions:
-        posix_permissions = info._posix_permissions(follow_symlinks=follow_symlinks)
-        try:
-            os.chmod(target, posix_permissions, follow_symlinks=follow_symlinks)
-        except NotImplementedError:
-            # if we got a NotImplementedError, it's because
-            #   * follow_symlinks=False,
-            #   * lchown() is unavailable, and
-            #   * either
-            #       * fchownat() is unavailable or
-            #       * fchownat() doesn't implement AT_SYMLINK_NOFOLLOW.
-            #         (it returned ENOSUP.)
-            # therefore we're out of options--we simply cannot chown the
-            # symlink.  give up, suppress the error.
-            # (which is what shutil always did in this circumstance.)
-            pass
-
-    copy_bsd_flags = (
-        hasattr(info, '_bsd_flags') and
-        hasattr(os, 'chflags') and
-        (follow_symlinks or os.chflags in os.supports_follow_symlinks))
-    if copy_bsd_flags:
-        bsd_flags = info._bsd_flags(follow_symlinks=follow_symlinks)
-        try:
-            os.chflags(target, bsd_flags, follow_symlinks=follow_symlinks)
-        except OSError as why:
-            if why.errno not in (EOPNOTSUPP, ENOTSUP):
-                raise
-
-
-class _PathInfoBase:
-    __slots__ = ('_path', '_stat_result', '_lstat_result')
-
-    def __init__(self, path):
-        self._path = str(path)
-
-    def __repr__(self):
-        path_type = "WindowsPath" if os.name == "nt" else "PosixPath"
-        return f"<{path_type}.info>"
-
-    def _stat(self, *, follow_symlinks=True, ignore_errors=False):
-        """Return the status as an os.stat_result, or None if stat() fails and
-        ignore_errors is true."""
-        if follow_symlinks:
-            try:
-                result = self._stat_result
-            except AttributeError:
-                pass
-            else:
-                if ignore_errors or result is not None:
-                    return result
-            try:
-                self._stat_result = os.stat(self._path)
-            except (OSError, ValueError):
-                self._stat_result = None
-                if not ignore_errors:
-                    raise
-            return self._stat_result
-        else:
-            try:
-                result = self._lstat_result
-            except AttributeError:
-                pass
-            else:
-                if ignore_errors or result is not None:
-                    return result
-            try:
-                self._lstat_result = os.lstat(self._path)
-            except (OSError, ValueError):
-                self._lstat_result = None
-                if not ignore_errors:
-                    raise
-            return self._lstat_result
-
-    def _posix_permissions(self, *, follow_symlinks=True):
-        """Return the POSIX file permissions."""
-        return S_IMODE(self._stat(follow_symlinks=follow_symlinks).st_mode)
-
-    def _file_id(self, *, follow_symlinks=True):
-        """Returns the identifier of the file."""
-        st = self._stat(follow_symlinks=follow_symlinks)
-        return st.st_dev, st.st_ino
-
-    def _access_time_ns(self, *, follow_symlinks=True):
-        """Return the access time in nanoseconds."""
-        return self._stat(follow_symlinks=follow_symlinks).st_atime_ns
-
-    def _mod_time_ns(self, *, follow_symlinks=True):
-        """Return the modify time in nanoseconds."""
-        return self._stat(follow_symlinks=follow_symlinks).st_mtime_ns
-
-    if hasattr(os.stat_result, 'st_flags'):
-        def _bsd_flags(self, *, follow_symlinks=True):
-            """Return the flags."""
-            return self._stat(follow_symlinks=follow_symlinks).st_flags
-
-    if hasattr(os, 'listxattr'):
-        def _xattrs(self, *, follow_symlinks=True):
-            """Return the xattrs as a list of (attr, value) pairs, or an empty
-            list if extended attributes aren't supported."""
-            try:
-                return [
-                    (attr, os.getxattr(self._path, attr, follow_symlinks=follow_symlinks))
-                    for attr in os.listxattr(self._path, follow_symlinks=follow_symlinks)]
-            except OSError as err:
-                if err.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):
-                    raise
-                return []
-
-
-class _WindowsPathInfo(_PathInfoBase):
-    """Implementation of pathlib.types.PathInfo that provides status
-    information for Windows paths. Don't try to construct it yourself."""
-    __slots__ = ('_exists', '_is_dir', '_is_file', '_is_symlink')
-
-    def exists(self, *, follow_symlinks=True):
-        """Whether this path exists."""
-        if not follow_symlinks and self.is_symlink():
-            return True
-        try:
-            return self._exists
-        except AttributeError:
-            if os.path.exists(self._path):
-                self._exists = True
-                return True
-            else:
-                self._exists = self._is_dir = self._is_file = False
-                return False
-
-    def is_dir(self, *, follow_symlinks=True):
-        """Whether this path is a directory."""
-        if not follow_symlinks and self.is_symlink():
-            return False
-        try:
-            return self._is_dir
-        except AttributeError:
-            if os.path.isdir(self._path):
-                self._is_dir = self._exists = True
-                return True
-            else:
-                self._is_dir = False
-                return False
-
-    def is_file(self, *, follow_symlinks=True):
-        """Whether this path is a regular file."""
-        if not follow_symlinks and self.is_symlink():
-            return False
-        try:
-            return self._is_file
-        except AttributeError:
-            if os.path.isfile(self._path):
-                self._is_file = self._exists = True
-                return True
-            else:
-                self._is_file = False
-                return False
-
-    def is_symlink(self):
-        """Whether this path is a symbolic link."""
-        try:
-            return self._is_symlink
-        except AttributeError:
-            self._is_symlink = os.path.islink(self._path)
-            return self._is_symlink
-
-
-class _PosixPathInfo(_PathInfoBase):
-    """Implementation of pathlib.types.PathInfo that provides status
-    information for POSIX paths. Don't try to construct it yourself."""
-    __slots__ = ()
-
-    def exists(self, *, follow_symlinks=True):
-        """Whether this path exists."""
-        st = self._stat(follow_symlinks=follow_symlinks, ignore_errors=True)
-        if st is None:
-            return False
-        return True
-
-    def is_dir(self, *, follow_symlinks=True):
-        """Whether this path is a directory."""
-        st = self._stat(follow_symlinks=follow_symlinks, ignore_errors=True)
-        if st is None:
-            return False
-        return S_ISDIR(st.st_mode)
-
-    def is_file(self, *, follow_symlinks=True):
-        """Whether this path is a regular file."""
-        st = self._stat(follow_symlinks=follow_symlinks, ignore_errors=True)
-        if st is None:
-            return False
-        return S_ISREG(st.st_mode)
-
-    def is_symlink(self):
-        """Whether this path is a symbolic link."""
-        st = self._stat(follow_symlinks=False, ignore_errors=True)
-        if st is None:
-            return False
-        return S_ISLNK(st.st_mode)
-
-
-PathInfo = _WindowsPathInfo if os.name == 'nt' else _PosixPathInfo
-
-
-class DirEntryInfo(_PathInfoBase):
-    """Implementation of pathlib.types.PathInfo that provides status
-    information by querying a wrapped os.DirEntry object. Don't try to
-    construct it yourself."""
-    __slots__ = ('_entry',)
-
-    def __init__(self, entry):
-        super().__init__(entry.path)
-        self._entry = entry
-
-    def _stat(self, *, follow_symlinks=True, ignore_errors=False):
-        try:
-            return self._entry.stat(follow_symlinks=follow_symlinks)
-        except OSError:
-            if not ignore_errors:
-                raise
-            return None
-
-    def exists(self, *, follow_symlinks=True):
-        """Whether this path exists."""
-        if not follow_symlinks:
-            return True
-        return self._stat(ignore_errors=True) is not None
-
-    def is_dir(self, *, follow_symlinks=True):
-        """Whether this path is a directory."""
-        try:
-            return self._entry.is_dir(follow_symlinks=follow_symlinks)
-        except OSError:
-            return False
-
-    def is_file(self, *, follow_symlinks=True):
-        """Whether this path is a regular file."""
-        try:
-            return self._entry.is_file(follow_symlinks=follow_symlinks)
-        except OSError:
-            return False
-
-    def is_symlink(self):
-        """Whether this path is a symbolic link."""
-        try:
-            return self._entry.is_symlink()
-        except OSError:
-            return False

Lib/pathlib/types.py

@@ -11,9 +11,10 @@ Protocols for supporting classes in pathlib.
 
 
 from abc import ABC, abstractmethod
-from glob import _PathGlobber
+from glob import _GlobberBase
 from io import text_encoding
-from pathlib._os import magic_open, ensure_distinct_paths, ensure_different_files, copyfileobj
+from pathlib._os import (vfsopen, vfspath, ensure_distinct_paths,
+                         ensure_different_files, copyfileobj)
 from pathlib import PurePath, Path
 from typing import Optional, Protocol, runtime_checkable
 
@@ -60,6 +61,25 @@ class PathInfo(Protocol):
     def is_symlink(self) -> bool: ...
 
 
+class _PathGlobber(_GlobberBase):
+    """Provides shell-style pattern matching and globbing for ReadablePath.
+    """
+
+    @staticmethod
+    def lexists(path):
+        return path.info.exists(follow_symlinks=False)
+
+    @staticmethod
+    def scandir(path):
+        return ((child.info, child.name, child) for child in path.iterdir())
+
+    @staticmethod
+    def concat_path(path, text):
+        return path.with_segments(vfspath(path) + text)
+
+    stringify_path = staticmethod(vfspath)
+
+
 class _JoinablePath(ABC):
     """Abstract base class for pure path objects.
 
@@ -86,20 +106,19 @@ class _JoinablePath(ABC):
         raise NotImplementedError
 
     @abstractmethod
-    def __str__(self):
-        """Return the string representation of the path, suitable for
-        passing to system calls."""
+    def __vfspath__(self):
+        """Return the string representation of the path."""
         raise NotImplementedError
 
     @property
     def anchor(self):
         """The concatenation of the drive and root, or ''."""
-        return _explode_path(str(self), self.parser.split)[0]
+        return _explode_path(vfspath(self), self.parser.split)[0]
 
     @property
     def name(self):
         """The final path component, if any."""
-        return self.parser.split(str(self))[1]
+        return self.parser.split(vfspath(self))[1]
 
     @property
     def suffix(self):
@@ -135,7 +154,7 @@ class _JoinablePath(ABC):
         split = self.parser.split
         if split(name)[0]:
             raise ValueError(f"Invalid name {name!r}")
-        path = str(self)
+        path = vfspath(self)
         path = path.removesuffix(split(path)[1]) + name
         return self.with_segments(path)
 
@@ -168,7 +187,7 @@ class _JoinablePath(ABC):
     def parts(self):
         """An object providing sequence-like access to the
         components in the filesystem path."""
-        anchor, parts = _explode_path(str(self), self.parser.split)
+        anchor, parts = _explode_path(vfspath(self), self.parser.split)
         if anchor:
             parts.append(anchor)
         return tuple(reversed(parts))
@@ -179,24 +198,24 @@ class _JoinablePath(ABC):
         paths) or a totally different path (if one of the arguments is
         anchored).
         """
-        return self.with_segments(str(self), *pathsegments)
+        return self.with_segments(vfspath(self), *pathsegments)
 
     def __truediv__(self, key):
         try:
-            return self.with_segments(str(self), key)
+            return self.with_segments(vfspath(self), key)
         except TypeError:
             return NotImplemented
 
     def __rtruediv__(self, key):
         try:
-            return self.with_segments(key, str(self))
+            return self.with_segments(key, vfspath(self))
         except TypeError:
             return NotImplemented
 
     @property
     def parent(self):
         """The logical parent of the path."""
-        path = str(self)
+        path = vfspath(self)
         parent = self.parser.split(path)[0]
         if path != parent:
             return self.with_segments(parent)
@@ -206,7 +225,7 @@ class _JoinablePath(ABC):
     def parents(self):
         """A sequence of this path's logical parents."""
         split = self.parser.split
-        path = str(self)
+        path = vfspath(self)
         parent = split(path)[0]
         parents = []
         while path != parent:
@@ -215,6 +234,33 @@ class _JoinablePath(ABC):
             parent = split(path)[0]
         return tuple(parents)
 
+    def relative_to(self, other, *, walk_up=False):
+        """Return the relative path to another path identified by the passed
+        arguments.  If the operation is not possible (because this is not
+        related to the other path), raise ValueError.
+
+        The *walk_up* parameter controls whether `..` may be used to resolve
+        the path.
+        """
+        parts = []
+        for path in (other,) + other.parents:
+            if self.is_relative_to(path):
+                break
+            elif not walk_up:
+                raise ValueError(f"{self!r} is not in the subpath of {other!r}")
+            elif path.name == '..':
+                raise ValueError(f"'..' segment in {other!r} cannot be walked")
+            else:
+                parts.append('..')
+        else:
+            raise ValueError(f"{self!r} and {other!r} have different anchors")
+        return self.with_segments(*parts, *self.parts[len(path.parts):])
+
+    def is_relative_to(self, other):
+        """Return True if the path is relative to another path or False.
+        """
+        return other == self or other in self.parents
+
     def full_match(self, pattern):
         """
         Return True if this path matches the given glob-style pattern. The
@@ -223,7 +269,7 @@ class _JoinablePath(ABC):
         case_sensitive = self.parser.normcase('Aa') == 'Aa'
         globber = _PathGlobber(self.parser.sep, case_sensitive, recursive=True)
         match = globber.compile(pattern, altsep=self.parser.altsep)
-        return match(str(self)) is not None
+        return match(vfspath(self)) is not None
 
 
 class _ReadablePath(_JoinablePath):
@@ -245,10 +291,10 @@ class _ReadablePath(_JoinablePath):
         raise NotImplementedError
 
     @abstractmethod
-    def __open_rb__(self, buffering=-1):
+    def __open_reader__(self):
         """
         Open the file pointed to by this path for reading in binary mode and
-        return a file object, like open(mode='rb').
+        return a file object.
         """
         raise NotImplementedError
 
@@ -256,7 +302,7 @@ class _ReadablePath(_JoinablePath):
         """
         Open the file in bytes mode, read it, and close the file.
         """
-        with magic_open(self, mode='rb', buffering=0) as f:
+        with vfsopen(self, mode='rb') as f:
             return f.read()
 
     def read_text(self, encoding=None, errors=None, newline=None):
@@ -266,7 +312,7 @@ class _ReadablePath(_JoinablePath):
         # Call io.text_encoding() here to ensure any warning is raised at an
         # appropriate stack level.
         encoding = text_encoding(encoding)
-        with magic_open(self, mode='r', encoding=encoding, errors=errors, newline=newline) as f:
+        with vfsopen(self, mode='r', encoding=encoding, errors=errors, newline=newline) as f:
             return f.read()
 
     @abstractmethod
@@ -375,10 +421,10 @@ class _WritablePath(_JoinablePath):
         raise NotImplementedError
 
     @abstractmethod
-    def __open_wb__(self, buffering=-1):
+    def __open_writer__(self, mode):
         """
         Open the file pointed to by this path for writing in binary mode and
-        return a file object, like open(mode='wb').
+        return a file object.
         """
         raise NotImplementedError
 
@@ -388,7 +434,7 @@ class _WritablePath(_JoinablePath):
         """
         # type-check for the buffer interface before truncating the file
         view = memoryview(data)
-        with magic_open(self, mode='wb') as f:
+        with vfsopen(self, mode='wb') as f:
             return f.write(view)
 
     def write_text(self, data, encoding=None, errors=None, newline=None):
@@ -401,7 +447,7 @@ class _WritablePath(_JoinablePath):
         if not isinstance(data, str):
             raise TypeError('data must be str, not %s' %
                             data.__class__.__name__)
-        with magic_open(self, mode='w', encoding=encoding, errors=errors, newline=newline) as f:
+        with vfsopen(self, mode='w', encoding=encoding, errors=errors, newline=newline) as f:
             return f.write(data)
 
     def _copy_from(self, source, follow_symlinks=True):
@@ -412,7 +458,7 @@ class _WritablePath(_JoinablePath):
         while stack:
             src, dst = stack.pop()
             if not follow_symlinks and src.info.is_symlink():
-                dst.symlink_to(str(src.readlink()), src.info.is_dir())
+                dst.symlink_to(vfspath(src.readlink()), src.info.is_dir())
             elif src.info.is_dir():
                 children = src.iterdir()
                 dst.mkdir()
@@ -420,8 +466,8 @@ class _WritablePath(_JoinablePath):
                     stack.append((child, dst.joinpath(child.name)))
             else:
                 ensure_different_files(src, dst)
-                with magic_open(src, 'rb') as source_f:
-                    with magic_open(dst, 'wb') as target_f:
+                with vfsopen(src, 'rb') as source_f:
+                    with vfsopen(dst, 'wb') as target_f:
                         copyfileobj(source_f, target_f)
 
 

Lib/py_compile.py

@@ -177,7 +177,7 @@ def main():
     import argparse
 
     description = 'A simple command-line interface for py_compile module.'
-    parser = argparse.ArgumentParser(description=description, color=True)
+    parser = argparse.ArgumentParser(description=description)
     parser.add_argument(
         '-q', '--quiet',
         action='store_true',
@@ -194,8 +194,10 @@ def main():
     else:
         filenames = args.filenames
     for filename in filenames:
+        cfilename = (None if sys.implementation.cache_tag
+                     else f"{filename.rpartition('.')[0]}.pyc")
         try:
-            compile(filename, doraise=True)
+            compile(filename, cfilename, doraise=True)
         except PyCompileError as error:
             if args.quiet:
                 parser.exit(1)

Lib/tempfile.py

@@ -43,6 +43,7 @@ import os as _os
 import shutil as _shutil
 import errno as _errno
 from random import Random as _Random
+import stat as _stat
 import sys as _sys
 import types as _types
 import weakref as _weakref
@@ -273,15 +274,68 @@ def _dont_follow_symlinks(func, path, *args):
     elif not _os.path.islink(path):
         func(path, *args)
 
-def _resetperms(path):
+def _resetflags(path):
     try:
         chflags = _os.chflags
     except AttributeError:
         pass
     else:
         _dont_follow_symlinks(chflags, path, 0)
+
+def _resetperms(path):
+    _resetflags(path)
     _dont_follow_symlinks(_os.chmod, path, 0o700)
 
+# True if TemporaryDirectory._rmtree() can work relative to open directories
+# instead of resolving paths again.
+_rmtree_use_dir_fd = (
+    {_os.chmod, _os.unlink, _os.lstat} <= _os.supports_dir_fd
+    and _os.chmod in _os.supports_fd
+)
+
+def _resetperms_fd(dir_fd, path):
+    # Same as _resetperms(), but for the directory referred to by dir_fd.
+    if dir_fd is None:
+        _resetperms(path)
+        return
+    _resetflags(path)
+    _os.chmod(dir_fd, 0o700)
+
+try:
+    _nofollow_mode = _os.O_RDONLY | _os.O_NONBLOCK | _os.O_NOFOLLOW
+except AttributeError:
+    _nofollow_mode = None
+
+def _resetperms_at(name, dir_fd, path):
+    # Same as _resetperms(), but name is resolved relative to the directory
+    # file descriptor dir_fd. path is only used for os.chflags(), which
+    # doesn't support dir_fd or file descriptors.
+    if dir_fd is None:
+        _resetperms(path)
+        return
+    _resetflags(path)
+    if _os.chmod in _os.supports_follow_symlinks:
+        _os.chmod(name, 0o700, dir_fd=dir_fd, follow_symlinks=False)
+    else:
+        # dir_fd & follow_symlinks is not supported on this platform.
+        # Try chmod opening the file with O_NOFOLLOW.
+        if _nofollow_mode is not None:
+            try:
+                fd = _os.open(name, _nofollow_mode, dir_fd=dir_fd)
+            except OSError:
+                pass
+            else:
+                try:
+                    _os.chmod(fd, 0o700)
+                finally:
+                    _os.close(fd)
+                return
+        # If that did not work, we change by name, which is subject to a race
+        # condition.
+        stat = _os.lstat(name, dir_fd=dir_fd)
+        if not _stat.S_ISLNK(stat.st_mode):
+            _os.chmod(name, 0o700, dir_fd=dir_fd)
+
 
 # User visible interfaces.
 
@@ -656,7 +710,7 @@ else:
             fd = None
             def opener(*args):
                 nonlocal fd
-                flags2 = (flags | _os.O_TMPFILE) & ~_os.O_CREAT
+                flags2 = (flags | _os.O_TMPFILE) & ~_os.O_CREAT & ~_os.O_EXCL
                 fd = _os.open(dir, flags2, 0o600)
                 return fd
             try:
@@ -691,7 +745,7 @@ else:
             fd, name = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
             try:
                 _os.unlink(name)
-            except BaseException as e:
+            except BaseException:
                 _os.close(fd)
                 raise
             return fd
@@ -913,8 +967,12 @@ class TemporaryDirectory:
             ignore_errors=self._ignore_cleanup_errors, delete=self._delete)
 
     @classmethod
-    def _rmtree(cls, name, ignore_errors=False, repeated=False):
-        def onexc(func, path, exc):
+    def _rmtree(cls, name, ignore_errors=False, repeated=False, dir_fd=None,
+                fullname=None):
+        if fullname is None:
+            fullname = name
+
+        def onexc(func, path, exc, direntry=None, dir_fd=None):
             # On DragonFly BSD, UF_NOUNLINK removal fails with EISDIR, not EPERM.
             if isinstance(exc, (PermissionError, IsADirectoryError)):
                 if repeated and path == name:
@@ -922,15 +980,30 @@ class TemporaryDirectory:
                         return
                     raise
 
+                # fullpath is path as seen from the working directory
+                fullpath = fullname + path[len(name):]
+                # base is path relative to dir_fd, the directory rmtree()
+                # reached it through, or the whole path when there is none
+                if dir_fd is None or not _rmtree_use_dir_fd:
+                    base, dir_fd = path, None
+                elif direntry is None:
+                    base = path
+                else:
+                    base = direntry.name
+
                 try:
                     if path != name:
-                        _resetperms(_os.path.dirname(path))
-                    _resetperms(path)
+                        # The parent directory of path is the one referred to
+                        # by dir_fd.
+                        _resetperms_fd(dir_fd, _os.path.dirname(fullpath))
+                    _resetperms_at(base, dir_fd, fullpath)
 
                     try:
-                        _os.unlink(path)
+                        _os.unlink(base, dir_fd=dir_fd)
                     except IsADirectoryError:
-                        cls._rmtree(path, ignore_errors=ignore_errors)
+                        cls._rmtree(base, ignore_errors=ignore_errors,
+                                    repeated=(path == name),
+                                    dir_fd=dir_fd, fullname=fullpath)
                     except PermissionError:
                         # The PermissionError handler was originally added for
                         # FreeBSD in directories, but it seems that it is raised
@@ -939,21 +1012,27 @@ class TemporaryDirectory:
                         # raise NotADirectoryError and mask the PermissionError.
                         # So we must re-raise the current PermissionError if
                         # path is not a directory.
-                        if not _os.path.isdir(path) or _os.path.isjunction(path):
+                        if (not _os.path.isdir(fullpath)
+                                or _os.path.isjunction(fullpath)):
                             if ignore_errors:
                                 return
                             raise
-                        cls._rmtree(path, ignore_errors=ignore_errors,
-                                    repeated=(path == name))
+                        cls._rmtree(base, ignore_errors=ignore_errors,
+                                    repeated=(path == name),
+                                    dir_fd=dir_fd, fullname=fullpath)
                 except FileNotFoundError:
                     pass
+                except OSError:
+                    if ignore_errors:
+                        return
+                    raise
             elif isinstance(exc, FileNotFoundError):
                 pass
             else:
                 if not ignore_errors:
                     raise
 
-        _shutil.rmtree(name, onexc=onexc)
+        _shutil.rmtree(name, onexc=onexc, dir_fd=dir_fd, _onexc_kwargs=True)
 
     @classmethod
     def _cleanup(cls, name, warn_message, ignore_errors=False, delete=True):

Lib/test/test_compileall.py

@@ -28,11 +28,15 @@ except (NotImplementedError, ModuleNotFoundError):
 from test import support
 from test.support import os_helper
 from test.support import script_helper
-from test.test_py_compile import without_source_date_epoch
-from test.test_py_compile import SourceDateEpochTestMeta
+from test.support.os_helper import without_source_date_epoch
+from test.support.os_helper import SourceDateEpochTestMeta
 from test.support.os_helper import FakePath
 
 
+if sys.implementation.cache_tag is None:
+    raise unittest.SkipTest('requires sys.implementation.cache_tag is not None')
+
+
 def get_pyc(script, opt):
     if not opt:
         # Replace None and 0 with ''

Lib/test/test_glob.py

@@ -4,7 +4,6 @@ import re
 import shutil
 import sys
 import unittest
-import warnings
 
 from test.support import is_wasi, Py_DEBUG
 from test.support.os_helper import (TESTFN, skip_unless_symlink,
@@ -393,36 +392,6 @@ class GlobTests(unittest.TestCase):
             for it in iters:
                 self.assertEqual(next(it), p)
 
-    def test_glob0(self):
-        with self.assertWarns(DeprecationWarning):
-            glob.glob0(self.tempdir, 'a')
-
-        with warnings.catch_warnings():
-            warnings.simplefilter('ignore')
-            eq = self.assertSequencesEqual_noorder
-            eq(glob.glob0(self.tempdir, 'a'), ['a'])
-            eq(glob.glob0(self.tempdir, '.bb'), ['.bb'])
-            eq(glob.glob0(self.tempdir, '.b*'), [])
-            eq(glob.glob0(self.tempdir, 'b'), [])
-            eq(glob.glob0(self.tempdir, '?'), [])
-            eq(glob.glob0(self.tempdir, '*a'), [])
-            eq(glob.glob0(self.tempdir, 'a*'), [])
-
-    def test_glob1(self):
-        with self.assertWarns(DeprecationWarning):
-            glob.glob1(self.tempdir, 'a')
-
-        with warnings.catch_warnings():
-            warnings.simplefilter('ignore')
-            eq = self.assertSequencesEqual_noorder
-            eq(glob.glob1(self.tempdir, 'a'), ['a'])
-            eq(glob.glob1(self.tempdir, '.bb'), ['.bb'])
-            eq(glob.glob1(self.tempdir, '.b*'), ['.bb'])
-            eq(glob.glob1(self.tempdir, 'b'), [])
-            eq(glob.glob1(self.tempdir, '?'), ['a'])
-            eq(glob.glob1(self.tempdir, '*a'), ['a', 'aaa'])
-            eq(glob.glob1(self.tempdir, 'a*'), ['a', 'aaa', 'aab'])
-
     def test_translate_matching(self):
         match = re.compile(glob.translate('*')).match
         self.assertIsNotNone(match('foo'))

Lib/test/test_pathlib/support/lexical_path.py

@@ -9,9 +9,10 @@ import posixpath
 from . import is_pypi
 
 if is_pypi:
-    from pathlib_abc import _JoinablePath
+    from pathlib_abc import vfspath, _JoinablePath
 else:
     from pathlib.types import _JoinablePath
+    from pathlib._os import vfspath
 
 
 class LexicalPath(_JoinablePath):
@@ -22,20 +23,20 @@ class LexicalPath(_JoinablePath):
         self._segments = pathsegments
 
     def __hash__(self):
-        return hash(str(self))
+        return hash(vfspath(self))
 
     def __eq__(self, other):
         if not isinstance(other, LexicalPath):
             return NotImplemented
-        return str(self) == str(other)
+        return vfspath(self) == vfspath(other)
 
-    def __str__(self):
+    def __vfspath__(self):
         if not self._segments:
             return ''
         return self.parser.join(*self._segments)
 
     def __repr__(self):
-        return f'{type(self).__name__}({str(self)!r})'
+        return f'{type(self).__name__}({vfspath(self)!r})'
 
     def with_segments(self, *pathsegments):
         return type(self)(*pathsegments)

Lib/test/test_pathlib/support/local_path.py

@@ -97,7 +97,7 @@ class LocalPathInfo(PathInfo):
     __slots__ = ('_path', '_exists', '_is_dir', '_is_file', '_is_symlink')
 
     def __init__(self, path):
-        self._path = str(path)
+        self._path = os.fspath(path)
         self._exists = None
         self._is_dir = None
         self._is_file = None
@@ -139,15 +139,13 @@ class ReadableLocalPath(_ReadablePath, LexicalPath):
     Simple implementation of a ReadablePath class for local filesystem paths.
     """
     __slots__ = ('info',)
+    __fspath__ = LexicalPath.__vfspath__
 
     def __init__(self, *pathsegments):
         super().__init__(*pathsegments)
         self.info = LocalPathInfo(self)
 
-    def __fspath__(self):
-        return str(self)
-
-    def __open_rb__(self, buffering=-1):
+    def __open_reader__(self):
         return open(self, 'rb')
 
     def iterdir(self):
@@ -163,12 +161,10 @@ class WritableLocalPath(_WritablePath, LexicalPath):
     """
 
     __slots__ = ()
+    __fspath__ = LexicalPath.__vfspath__
 
-    def __fspath__(self):
-        return str(self)
-
-    def __open_wb__(self, buffering=-1):
-        return open(self, 'wb')
+    def __open_writer__(self, mode):
+        return open(self, f'{mode}b')
 
     def mkdir(self, mode=0o777):
         os.mkdir(self, mode)

Lib/test/test_pathlib/support/zip_path.py

@@ -16,9 +16,10 @@ from stat import S_IFMT, S_ISDIR, S_ISREG, S_ISLNK
 from . import is_pypi
 
 if is_pypi:
-    from pathlib_abc import PathInfo, _ReadablePath, _WritablePath
+    from pathlib_abc import vfspath, PathInfo, _ReadablePath, _WritablePath
 else:
     from pathlib.types import PathInfo, _ReadablePath, _WritablePath
+    from pathlib._os import vfspath
 
 
 class ZipPathGround:
@@ -34,16 +35,16 @@ class ZipPathGround:
         root.zip_file.close()
 
     def create_file(self, path, data=b''):
-        path.zip_file.writestr(str(path), data)
+        path.zip_file.writestr(vfspath(path), data)
 
     def create_dir(self, path):
-        zip_info = zipfile.ZipInfo(str(path) + '/')
+        zip_info = zipfile.ZipInfo(vfspath(path) + '/')
         zip_info.external_attr |= stat.S_IFDIR << 16
         zip_info.external_attr |= stat.FILE_ATTRIBUTE_DIRECTORY
         path.zip_file.writestr(zip_info, '')
 
     def create_symlink(self, path, target):
-        zip_info = zipfile.ZipInfo(str(path))
+        zip_info = zipfile.ZipInfo(vfspath(path))
         zip_info.external_attr = stat.S_IFLNK << 16
         path.zip_file.writestr(zip_info, target.encode())
 
@@ -62,28 +63,28 @@ class ZipPathGround:
         self.create_symlink(p.joinpath('brokenLinkLoop'), 'brokenLinkLoop')
 
     def readtext(self, p):
-        with p.zip_file.open(str(p), 'r') as f:
+        with p.zip_file.open(vfspath(p), 'r') as f:
             f = io.TextIOWrapper(f, encoding='utf-8')
             return f.read()
 
     def readbytes(self, p):
-        with p.zip_file.open(str(p), 'r') as f:
+        with p.zip_file.open(vfspath(p), 'r') as f:
             return f.read()
 
     readlink = readtext
 
     def isdir(self, p):
-        path_str = str(p) + "/"
+        path_str = vfspath(p) + "/"
         return path_str in p.zip_file.NameToInfo
 
     def isfile(self, p):
-        info = p.zip_file.NameToInfo.get(str(p))
+        info = p.zip_file.NameToInfo.get(vfspath(p))
         if info is None:
             return False
         return not stat.S_ISLNK(info.external_attr >> 16)
 
     def islink(self, p):
-        info = p.zip_file.NameToInfo.get(str(p))
+        info = p.zip_file.NameToInfo.get(vfspath(p))
         if info is None:
             return False
         return stat.S_ISLNK(info.external_attr >> 16)
@@ -240,20 +241,20 @@ class ReadableZipPath(_ReadablePath):
             zip_file.filelist = ZipFileList(zip_file)
 
     def __hash__(self):
-        return hash((str(self), self.zip_file))
+        return hash((vfspath(self), self.zip_file))
 
     def __eq__(self, other):
         if not isinstance(other, ReadableZipPath):
             return NotImplemented
-        return str(self) == str(other) and self.zip_file is other.zip_file
+        return vfspath(self) == vfspath(other) and self.zip_file is other.zip_file
 
-    def __str__(self):
+    def __vfspath__(self):
         if not self._segments:
             return ''
         return self.parser.join(*self._segments)
 
     def __repr__(self):
-        return f'{type(self).__name__}({str(self)!r}, zip_file={self.zip_file!r})'
+        return f'{type(self).__name__}({vfspath(self)!r}, zip_file={self.zip_file!r})'
 
     def with_segments(self, *pathsegments):
         return type(self)(*pathsegments, zip_file=self.zip_file)
@@ -261,15 +262,15 @@ class ReadableZipPath(_ReadablePath):
     @property
     def info(self):
         tree = self.zip_file.filelist.tree
-        return tree.resolve(str(self), follow_symlinks=False)
+        return tree.resolve(vfspath(self), follow_symlinks=False)
 
-    def __open_rb__(self, buffering=-1):
+    def __open_reader__(self):
         info = self.info.resolve()
         if not info.exists():
             raise FileNotFoundError(errno.ENOENT, "File not found", self)
         elif info.is_dir():
             raise IsADirectoryError(errno.EISDIR, "Is a directory", self)
-        return self.zip_file.open(info.zip_info, 'r')
+        return self.zip_file.open(info.zip_info)
 
     def iterdir(self):
         info = self.info.resolve()
@@ -301,36 +302,36 @@ class WritableZipPath(_WritablePath):
         self.zip_file = zip_file
 
     def __hash__(self):
-        return hash((str(self), self.zip_file))
+        return hash((vfspath(self), self.zip_file))
 
     def __eq__(self, other):
         if not isinstance(other, WritableZipPath):
             return NotImplemented
-        return str(self) == str(other) and self.zip_file is other.zip_file
+        return vfspath(self) == vfspath(other) and self.zip_file is other.zip_file
 
-    def __str__(self):
+    def __vfspath__(self):
         if not self._segments:
             return ''
         return self.parser.join(*self._segments)
 
     def __repr__(self):
-        return f'{type(self).__name__}({str(self)!r}, zip_file={self.zip_file!r})'
+        return f'{type(self).__name__}({vfspath(self)!r}, zip_file={self.zip_file!r})'
 
     def with_segments(self, *pathsegments):
         return type(self)(*pathsegments, zip_file=self.zip_file)
 
-    def __open_wb__(self, buffering=-1):
-        return self.zip_file.open(str(self), 'w')
+    def __open_writer__(self, mode):
+        return self.zip_file.open(vfspath(self), mode)
 
     def mkdir(self, mode=0o777):
-        zinfo = zipfile.ZipInfo(str(self) + '/')
+        zinfo = zipfile.ZipInfo(vfspath(self) + '/')
         zinfo.external_attr |= stat.S_IFDIR << 16
         zinfo.external_attr |= stat.FILE_ATTRIBUTE_DIRECTORY
         self.zip_file.writestr(zinfo, '')
 
     def symlink_to(self, target, target_is_directory=False):
-        zinfo = zipfile.ZipInfo(str(self))
+        zinfo = zipfile.ZipInfo(vfspath(self))
         zinfo.external_attr = stat.S_IFLNK << 16
         if target_is_directory:
             zinfo.external_attr |= 0x10
-        self.zip_file.writestr(zinfo, str(target))
+        self.zip_file.writestr(zinfo, target)

Lib/test/test_pathlib/test_join.py

@@ -376,6 +376,61 @@ class JoinTestBase:
         self.assertRaises(ValueError, P('a/b').with_suffix, '.d/.')
         self.assertRaises(TypeError, P('a/b').with_suffix, None)
 
+    def test_relative_to(self):
+        P = self.cls
+        p = P('a/b')
+        self.assertEqual(p.relative_to(P('')), P('a', 'b'))
+        self.assertEqual(p.relative_to(P('a')), P('b'))
+        self.assertEqual(p.relative_to(P('a/b')), P(''))
+        self.assertEqual(p.relative_to(P(''), walk_up=True), P('a', 'b'))
+        self.assertEqual(p.relative_to(P('a'), walk_up=True), P('b'))
+        self.assertEqual(p.relative_to(P('a/b'), walk_up=True), P(''))
+        self.assertEqual(p.relative_to(P('a/c'), walk_up=True), P('..', 'b'))
+        self.assertEqual(p.relative_to(P('a/b/c'), walk_up=True), P('..'))
+        self.assertEqual(p.relative_to(P('c'), walk_up=True), P('..', 'a', 'b'))
+        self.assertRaises(ValueError, p.relative_to, P('c'))
+        self.assertRaises(ValueError, p.relative_to, P('a/b/c'))
+        self.assertRaises(ValueError, p.relative_to, P('a/c'))
+        self.assertRaises(ValueError, p.relative_to, P('/a'))
+        self.assertRaises(ValueError, p.relative_to, P('../a'))
+        self.assertRaises(ValueError, p.relative_to, P('a/..'))
+        self.assertRaises(ValueError, p.relative_to, P('/a/..'))
+        self.assertRaises(ValueError, p.relative_to, P('/'), walk_up=True)
+        self.assertRaises(ValueError, p.relative_to, P('/a'), walk_up=True)
+        self.assertRaises(ValueError, p.relative_to, P('../a'), walk_up=True)
+        self.assertRaises(ValueError, p.relative_to, P('a/..'), walk_up=True)
+        self.assertRaises(ValueError, p.relative_to, P('/a/..'), walk_up=True)
+        class Q(self.cls):
+            __eq__ = object.__eq__
+            __hash__ = object.__hash__
+        q = Q('a/b')
+        self.assertTrue(q.relative_to(q))
+        self.assertRaises(ValueError, q.relative_to, Q(''))
+        self.assertRaises(ValueError, q.relative_to, Q('a'))
+        self.assertRaises(ValueError, q.relative_to, Q('a'), walk_up=True)
+        self.assertRaises(ValueError, q.relative_to, Q('a/b'))
+        self.assertRaises(ValueError, q.relative_to, Q('c'))
+
+    def test_is_relative_to(self):
+        P = self.cls
+        p = P('a/b')
+        self.assertTrue(p.is_relative_to(P('')))
+        self.assertTrue(p.is_relative_to(P('a')))
+        self.assertTrue(p.is_relative_to(P('a/b')))
+        self.assertFalse(p.is_relative_to(P('c')))
+        self.assertFalse(p.is_relative_to(P('a/b/c')))
+        self.assertFalse(p.is_relative_to(P('a/c')))
+        self.assertFalse(p.is_relative_to(P('/a')))
+        class Q(self.cls):
+            __eq__ = object.__eq__
+            __hash__ = object.__hash__
+        q = Q('a/b')
+        self.assertTrue(q.is_relative_to(q))
+        self.assertFalse(q.is_relative_to(Q('')))
+        self.assertFalse(q.is_relative_to(Q('a')))
+        self.assertFalse(q.is_relative_to(Q('a/b')))
+        self.assertFalse(q.is_relative_to(Q('c')))
+
 
 class LexicalPathJoinTest(JoinTestBase, unittest.TestCase):
     cls = LexicalPath

Lib/test/test_pathlib/test_join_windows.py

@@ -8,6 +8,11 @@ import unittest
 from .support import is_pypi
 from .support.lexical_path import LexicalWindowsPath
 
+if is_pypi:
+    from pathlib_abc import vfspath
+else:
+    from pathlib._os import vfspath
+
 
 class JoinTestBase:
     def test_join(self):
@@ -70,17 +75,22 @@ class JoinTestBase:
         self.assertEqual(p / './dd:s', P(r'C:/a/b\./dd:s'))
         self.assertEqual(p / 'E:d:s', P('E:d:s'))
 
-    def test_str(self):
+    def test_vfspath(self):
         p = self.cls(r'a\b\c')
-        self.assertEqual(str(p), 'a\\b\\c')
+        self.assertEqual(vfspath(p), 'a\\b\\c')
         p = self.cls(r'c:\a\b\c')
-        self.assertEqual(str(p), 'c:\\a\\b\\c')
+        self.assertEqual(vfspath(p), 'c:\\a\\b\\c')
         p = self.cls('\\\\a\\b\\')
-        self.assertEqual(str(p), '\\\\a\\b\\')
+        self.assertEqual(vfspath(p), '\\\\a\\b\\')
         p = self.cls(r'\\a\b\c')
-        self.assertEqual(str(p), '\\\\a\\b\\c')
+        self.assertEqual(vfspath(p), '\\\\a\\b\\c')
         p = self.cls(r'\\a\b\c\d')
-        self.assertEqual(str(p), '\\\\a\\b\\c\\d')
+        self.assertEqual(vfspath(p), '\\\\a\\b\\c\\d')
+
+    def test_invalid_vspath(self):
+        msg = "expected JoinablePath object, not NoneType"
+        with self.assertRaisesRegex(TypeError, msg):
+            vfspath(None)
 
     def test_parts(self):
         P = self.cls

Lib/test/test_pathlib/test_pathlib.py

@@ -20,6 +20,8 @@ from test.support import cpython_only
 from test.support import is_emscripten, is_wasi, is_wasm32
 from test.support import infinite_recursion
 from test.support import os_helper
+from test.support import requires_root_user
+from test.support import requires_non_root_user
 from test.support.os_helper import TESTFN, FS_NONASCII, FakePath
 try:
     import fcntl
@@ -35,11 +37,6 @@ except ImportError:
     posix = None
 
 
-root_in_posix = False
-if hasattr(os, 'geteuid'):
-    root_in_posix = (os.geteuid() == 0)
-
-
 def patch_replace(old_test):
     def new_replace(self, target):
         raise OSError(errno.EXDEV, "Cross-device link", self, target)
@@ -84,7 +81,7 @@ class UnsupportedOperationTest(unittest.TestCase):
 class LazyImportTest(unittest.TestCase):
     @cpython_only
     def test_lazy_import(self):
-        import_helper.ensure_lazy_imports("pathlib", {"shutil"})
+        import_helper.ensure_lazy_imports("pathlib", {"glob", "shutil"})
 
 
 #
@@ -545,12 +542,6 @@ class PurePathTest(unittest.TestCase):
         self.assertRaises(ValueError, P('a/b').with_stem, '')
         self.assertRaises(ValueError, P('a/b').with_stem, '.')
 
-    def test_is_reserved_deprecated(self):
-        P = self.cls
-        p = P('a/b')
-        with self.assertWarns(DeprecationWarning):
-            p.is_reserved()
-
     def test_full_match_case_sensitive(self):
         P = self.cls
         self.assertFalse(P('A.py').full_match('a.PY', case_sensitive=True))
@@ -1560,7 +1551,7 @@ class PathTest(PurePathTest):
             self.assertRaises(FileNotFoundError, source.copy, target)
 
     @unittest.skipIf(sys.platform == "win32" or sys.platform == "wasi", "directories are always readable on Windows and WASI")
-    @unittest.skipIf(root_in_posix, "test fails with root privilege")
+    @requires_non_root_user
     def test_copy_dir_no_read_permission(self):
         base = self.cls(self.base)
         source = base / 'dirE'
@@ -2033,7 +2024,7 @@ class PathTest(PurePathTest):
         self.assertEqual(expected_name, p.owner())
 
     @unittest.skipUnless(pwd, "the pwd module is needed for this test")
-    @unittest.skipUnless(root_in_posix, "test needs root privilege")
+    @requires_root_user
     def test_owner_no_follow_symlinks(self):
         all_users = [u.pw_uid for u in pwd.getpwall()]
         if len(all_users) < 2:
@@ -2068,7 +2059,7 @@ class PathTest(PurePathTest):
         self.assertEqual(expected_name, p.group())
 
     @unittest.skipUnless(grp, "the grp module is needed for this test")
-    @unittest.skipUnless(root_in_posix, "test needs root privilege")
+    @requires_root_user
     def test_group_no_follow_symlinks(self):
         all_groups = [g.gr_gid for g in grp.getgrall()]
         if len(all_groups) < 2:
@@ -2501,6 +2492,116 @@ class PathTest(PurePathTest):
                 self.assertNotIn(str(p12), concurrently_created)
             self.assertTrue(p.exists())
 
+    @unittest.skipIf(
+        is_emscripten or is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_mkdir_parents_umask(self):
+        # Test that parent directories respect umask when parent_mode is not set
+        p = self.cls(self.base, 'umasktest', 'child')
+        self.assertFalse(p.exists())
+        if os.name != 'nt':
+            with os_helper.temp_umask(0o002):
+                p.mkdir(0o755, parents=True)
+                self.assertTrue(p.exists())
+                # Leaf directory gets the specified mode
+                self.assertEqual(p.stat().st_mode & 0o777, 0o755)
+                # Parent directory respects umask (0o777 & ~0o002 = 0o775)
+                self.assertEqual(p.parent.stat().st_mode & 0o777, 0o775)
+
+    @unittest.skipIf(
+        is_emscripten or is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_mkdir_with_parent_mode(self):
+        # Test the parent_mode parameter
+        p = self.cls(self.base, 'newdirPM', 'subdirPM')
+        self.assertFalse(p.exists())
+        if os.name != 'nt':
+            with os_helper.temp_umask(0o022):
+                # Specify different modes for parent and leaf directories
+                p.mkdir(0o755, parents=True, parent_mode=0o750)
+                self.assertTrue(p.exists())
+                self.assertTrue(p.is_dir())
+                # Leaf directory gets the mode parameter
+                self.assertEqual(p.stat().st_mode & 0o777, 0o755)
+                # Parent directory gets the parent_mode parameter
+                self.assertEqual(p.parent.stat().st_mode & 0o777, 0o750)
+
+    @unittest.skipIf(
+        is_emscripten or is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_mkdir_parent_mode_deep_hierarchy(self):
+        # Test parent_mode with deep directory hierarchy
+        p = self.cls(self.base, 'level1PM', 'level2PM', 'level3PM')
+        self.assertFalse(p.exists())
+        if os.name != 'nt':
+            with os_helper.temp_umask(0o022):
+                p.mkdir(0o755, parents=True, parent_mode=0o700)
+                self.assertTrue(p.exists())
+                # Check that all parent directories have parent_mode
+                level1 = self.cls(self.base, 'level1PM')
+                level2 = level1 / 'level2PM'
+                self.assertEqual(level1.stat().st_mode & 0o777, 0o700)
+                self.assertEqual(level2.stat().st_mode & 0o777, 0o700)
+                # Leaf directory has the regular mode
+                self.assertEqual(p.stat().st_mode & 0o777, 0o755)
+
+    @unittest.skipIf(
+        is_emscripten or is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_mkdir_parent_mode_combined_with_umask(self):
+        # parent_mode, like mode, is combined with the process umask; it does
+        # not bypass it.
+        p = self.cls(self.base, 'umaskPM', 'child')
+        self.assertFalse(p.exists())
+        if os.name != 'nt':
+            with os_helper.temp_umask(0o022):
+                p.mkdir(0o777, parents=True, parent_mode=0o777)
+                self.assertTrue(p.exists())
+                # 0o777 is masked down to 0o755 by the 0o022 umask, for both
+                # the leaf (mode) and the parent (parent_mode).
+                self.assertEqual(p.stat().st_mode & 0o777, 0o755)
+                self.assertEqual(p.parent.stat().st_mode & 0o777, 0o755)
+
+    @unittest.skipIf(
+        is_emscripten or is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_mkdir_parent_mode_same_as_mode(self):
+        # Test setting parent_mode same as mode
+        p = self.cls(self.base, 'samedirPM', 'subdirPM')
+        self.assertFalse(p.exists())
+        if os.name != 'nt':
+            with os_helper.temp_umask(0o022):
+                p.mkdir(0o705, parents=True, parent_mode=0o705)
+                self.assertTrue(p.exists())
+                # Both directories should have the same mode
+                self.assertEqual(p.stat().st_mode & 0o777, 0o705)
+                self.assertEqual(p.parent.stat().st_mode & 0o777, 0o705)
+
     @needs_symlinks
     def test_symlink_to(self):
         P = self.cls(self.base)
@@ -2769,6 +2870,7 @@ class PathTest(PurePathTest):
             if (isinstance(e, PermissionError) or
                     "AF_UNIX path too long" in str(e)):
                 self.skipTest("cannot bind Unix socket: " + str(e))
+            raise
         self.assertTrue(P.is_socket())
         self.assertFalse(P.is_fifo())
         self.assertFalse(P.is_file())

Lib/test/test_pathlib/test_read.py

@@ -13,10 +13,10 @@ from .support.zip_path import ReadableZipPath, ZipPathGround
 
 if is_pypi:
     from pathlib_abc import PathInfo, _ReadablePath
-    from pathlib_abc._os import magic_open
+    from pathlib_abc._os import vfsopen
 else:
     from pathlib.types import PathInfo, _ReadablePath
-    from pathlib._os import magic_open
+    from pathlib._os import vfsopen
 
 
 class ReadTestBase:
@@ -32,10 +32,16 @@ class ReadTestBase:
 
     def test_open_r(self):
         p = self.root / 'fileA'
-        with magic_open(p, 'r', encoding='utf-8') as f:
+        with vfsopen(p, 'r', encoding='utf-8') as f:
             self.assertIsInstance(f, io.TextIOBase)
             self.assertEqual(f.read(), 'this is file A\n')
 
+    def test_open_r_buffering_error(self):
+        p = self.root / 'fileA'
+        self.assertRaises(ValueError, vfsopen, p, 'r', buffering=0)
+        self.assertRaises(ValueError, vfsopen, p, 'r', buffering=1)
+        self.assertRaises(ValueError, vfsopen, p, 'r', buffering=1024)
+
     @unittest.skipIf(
         not getattr(sys.flags, 'warn_default_encoding', 0),
         "Requires warn_default_encoding",
@@ -43,17 +49,17 @@ class ReadTestBase:
     def test_open_r_encoding_warning(self):
         p = self.root / 'fileA'
         with self.assertWarns(EncodingWarning) as wc:
-            with magic_open(p, 'r'):
+            with vfsopen(p, 'r'):
                 pass
         self.assertEqual(wc.filename, __file__)
 
     def test_open_rb(self):
         p = self.root / 'fileA'
-        with magic_open(p, 'rb') as f:
+        with vfsopen(p, 'rb') as f:
             self.assertEqual(f.read(), b'this is file A\n')
-        self.assertRaises(ValueError, magic_open, p, 'rb', encoding='utf8')
-        self.assertRaises(ValueError, magic_open, p, 'rb', errors='strict')
-        self.assertRaises(ValueError, magic_open, p, 'rb', newline='')
+        self.assertRaises(ValueError, vfsopen, p, 'rb', encoding='utf8')
+        self.assertRaises(ValueError, vfsopen, p, 'rb', errors='strict')
+        self.assertRaises(ValueError, vfsopen, p, 'rb', newline='')
 
     def test_read_bytes(self):
         p = self.root / 'fileA'
@@ -323,6 +329,11 @@ class ReadTestBase:
         self.assertFalse((p / 'fileA\udfff').info.is_symlink())
         self.assertFalse((p / 'fileA\x00').info.is_symlink())
 
+    def test_invalid_mode(self):
+        p = self.root / 'fileA'
+        with self.assertRaisesRegex(ValueError, 'invalid mode'):
+            vfsopen(p, 'q')
+
 
 class ZipPathReadTest(ReadTestBase, unittest.TestCase):
     ground = ZipPathGround(ReadableZipPath)

Lib/test/test_pathlib/test_write.py

@@ -13,10 +13,10 @@ from .support.zip_path import WritableZipPath, ZipPathGround
 
 if is_pypi:
     from pathlib_abc import _WritablePath
-    from pathlib_abc._os import magic_open
+    from pathlib_abc._os import vfsopen
 else:
     from pathlib.types import _WritablePath
-    from pathlib._os import magic_open
+    from pathlib._os import vfsopen
 
 
 class WriteTestBase:
@@ -31,11 +31,17 @@ class WriteTestBase:
 
     def test_open_w(self):
         p = self.root / 'fileA'
-        with magic_open(p, 'w', encoding='utf-8') as f:
+        with vfsopen(p, 'w', encoding='utf-8') as f:
             self.assertIsInstance(f, io.TextIOBase)
             f.write('this is file A\n')
         self.assertEqual(self.ground.readtext(p), 'this is file A\n')
 
+    def test_open_w_buffering_error(self):
+        p = self.root / 'fileA'
+        self.assertRaises(ValueError, vfsopen, p, 'w', buffering=0)
+        self.assertRaises(ValueError, vfsopen, p, 'w', buffering=1)
+        self.assertRaises(ValueError, vfsopen, p, 'w', buffering=1024)
+
     @unittest.skipIf(
         not getattr(sys.flags, 'warn_default_encoding', 0),
         "Requires warn_default_encoding",
@@ -43,19 +49,19 @@ class WriteTestBase:
     def test_open_w_encoding_warning(self):
         p = self.root / 'fileA'
         with self.assertWarns(EncodingWarning) as wc:
-            with magic_open(p, 'w'):
+            with vfsopen(p, 'w'):
                 pass
         self.assertEqual(wc.filename, __file__)
 
     def test_open_wb(self):
         p = self.root / 'fileA'
-        with magic_open(p, 'wb') as f:
+        with vfsopen(p, 'wb') as f:
             #self.assertIsInstance(f, io.BufferedWriter)
             f.write(b'this is file A\n')
         self.assertEqual(self.ground.readbytes(p), b'this is file A\n')
-        self.assertRaises(ValueError, magic_open, p, 'wb', encoding='utf8')
-        self.assertRaises(ValueError, magic_open, p, 'wb', errors='strict')
-        self.assertRaises(ValueError, magic_open, p, 'wb', newline='')
+        self.assertRaises(ValueError, vfsopen, p, 'wb', encoding='utf8')
+        self.assertRaises(ValueError, vfsopen, p, 'wb', errors='strict')
+        self.assertRaises(ValueError, vfsopen, p, 'wb', newline='')
 
     def test_write_bytes(self):
         p = self.root / 'fileA'

Lib/test/test_py_compile.py

@@ -1,4 +1,3 @@
-import functools
 import importlib.util
 import os
 import py_compile
@@ -11,43 +10,7 @@ import unittest
 
 from test import support
 from test.support import os_helper, script_helper
-
-
-def without_source_date_epoch(fxn):
-    """Runs function with SOURCE_DATE_EPOCH unset."""
-    @functools.wraps(fxn)
-    def wrapper(*args, **kwargs):
-        with os_helper.EnvironmentVarGuard() as env:
-            env.unset('SOURCE_DATE_EPOCH')
-            return fxn(*args, **kwargs)
-    return wrapper
-
-
-def with_source_date_epoch(fxn):
-    """Runs function with SOURCE_DATE_EPOCH set."""
-    @functools.wraps(fxn)
-    def wrapper(*args, **kwargs):
-        with os_helper.EnvironmentVarGuard() as env:
-            env['SOURCE_DATE_EPOCH'] = '123456789'
-            return fxn(*args, **kwargs)
-    return wrapper
-
-
-# Run tests with SOURCE_DATE_EPOCH set or unset explicitly.
-class SourceDateEpochTestMeta(type(unittest.TestCase)):
-    def __new__(mcls, name, bases, dct, *, source_date_epoch):
-        cls = super().__new__(mcls, name, bases, dct)
-
-        for attr in dir(cls):
-            if attr.startswith('test_'):
-                meth = getattr(cls, attr)
-                if source_date_epoch:
-                    wrapper = with_source_date_epoch(meth)
-                else:
-                    wrapper = without_source_date_epoch(meth)
-                setattr(cls, attr, wrapper)
-
-        return cls
+from test.support.os_helper import SourceDateEpochTestMeta
 
 
 class PyCompileTestsBase:
@@ -56,7 +19,10 @@ class PyCompileTestsBase:
         self.directory = tempfile.mkdtemp(dir=os.getcwd())
         self.source_path = os.path.join(self.directory, '_test.py')
         self.pyc_path = self.source_path + 'c'
-        self.cache_path = importlib.util.cache_from_source(self.source_path)
+        try:
+            self.cache_path = importlib.util.cache_from_source(self.source_path)
+        except NotImplementedError:
+            self.cache_path = None
         self.cwd_drive = os.path.splitdrive(os.getcwd())[0]
         # In these tests we compute relative paths.  When using Windows, the
         # current working directory path and the 'self.source_path' might be
@@ -73,10 +39,31 @@ class PyCompileTestsBase:
         if self.cwd_drive:
             os.chdir(self.cwd_drive)
 
+    def assert_cache_path_exists(self, should_exist=True):
+        if self.cache_path:
+            if should_exist:
+                self.assertTrue(os.path.exists(self.cache_path))
+            else:
+                self.assertFalse(os.path.exists(self.cache_path))
+            return
+        cache_dir = os.path.join(self.directory, '__pycache__')
+        if not os.path.isdir(cache_dir):
+            if should_exist:
+                self.fail('no __pycache__ directory exists')
+            return
+        for f in os.listdir(cache_dir):
+            if f.startswith('_test.') and f.endswith('.pyc'):
+                if should_exist:
+                    return
+                self.fail(f'__pycache__/{f} was created')
+        else:
+            if should_exist:
+                self.fail('no __pycache__/_test.*.pyc file exists')
+
     def test_absolute_path(self):
         py_compile.compile(self.source_path, self.pyc_path)
         self.assertTrue(os.path.exists(self.pyc_path))
-        self.assertFalse(os.path.exists(self.cache_path))
+        self.assert_cache_path_exists(False)
 
     def test_do_not_overwrite_symlinks(self):
         # In the face of a cfile argument being a symlink, bail out.
@@ -98,23 +85,25 @@ class PyCompileTestsBase:
         with self.assertRaises(FileExistsError):
             py_compile.compile(self.source_path, os.devnull)
 
+    @unittest.skipIf(sys.implementation.cache_tag is None,
+                     'requires sys.implementation.cache_tag is not None')
     def test_cache_path(self):
         py_compile.compile(self.source_path)
-        self.assertTrue(os.path.exists(self.cache_path))
+        self.assert_cache_path_exists(True)
 
     def test_cwd(self):
         with os_helper.change_cwd(self.directory):
             py_compile.compile(os.path.basename(self.source_path),
                                os.path.basename(self.pyc_path))
         self.assertTrue(os.path.exists(self.pyc_path))
-        self.assertFalse(os.path.exists(self.cache_path))
+        self.assert_cache_path_exists(False)
 
     @unittest.expectedFailureIf(sys.platform == "darwin" and int(__import__("platform").release().split(".")[0]) < 20, "TODO: RUSTPYTHON")
     def test_relative_path(self):
         py_compile.compile(os.path.relpath(self.source_path),
                            os.path.relpath(self.pyc_path))
         self.assertTrue(os.path.exists(self.pyc_path))
-        self.assertFalse(os.path.exists(self.cache_path))
+        self.assert_cache_path_exists(False)
 
     @os_helper.skip_if_dac_override
     @unittest.skipIf(os.name == 'nt',
@@ -137,14 +126,14 @@ class PyCompileTestsBase:
                                   'tokenizedata',
                                   'bad_coding2.py')
         with support.captured_stderr():
-            self.assertIsNone(py_compile.compile(bad_coding, doraise=False))
-        self.assertFalse(os.path.exists(
-            importlib.util.cache_from_source(bad_coding)))
+            self.assertIsNone(py_compile.compile(bad_coding, self.pyc_path,
+                                                 doraise=False))
+        self.assertFalse(os.path.exists(self.pyc_path))
 
     def test_source_date_epoch(self):
         py_compile.compile(self.source_path, self.pyc_path)
         self.assertTrue(os.path.exists(self.pyc_path))
-        self.assertFalse(os.path.exists(self.cache_path))
+        self.assert_cache_path_exists(False)
         with open(self.pyc_path, 'rb') as fp:
             flags = importlib._bootstrap_external._classify_pyc(
                 fp.read(), 'test', {})
@@ -156,6 +145,8 @@ class PyCompileTestsBase:
         self.assertEqual(flags, expected_flags)
 
     @unittest.skipIf(sys.flags.optimize > 0, 'test does not work with -O')
+    @unittest.skipIf(sys.implementation.cache_tag is None,
+                     'requires sys.implementation.cache_tag is not None')
     def test_double_dot_no_clobber(self):
         # http://bugs.python.org/issue22966
         # py_compile foo.bar.py -> __pycache__/foo.cpython-34.pyc
@@ -178,6 +169,8 @@ class PyCompileTestsBase:
             self.assertTrue(os.path.exists(cache_path))
             self.assertFalse(os.path.exists(pyc_path))
 
+    @unittest.skipIf(sys.implementation.cache_tag is None,
+                     'requires sys.implementation.cache_tag is not None')
     def test_optimization_path(self):
         # Specifying optimized bytecode should lead to a path reflecting that.
         self.assertIn('opt-2', py_compile.compile(self.source_path, optimize=2))
@@ -185,17 +178,19 @@ class PyCompileTestsBase:
     def test_invalidation_mode(self):
         py_compile.compile(
             self.source_path,
+            self.pyc_path,
             invalidation_mode=py_compile.PycInvalidationMode.CHECKED_HASH,
         )
-        with open(self.cache_path, 'rb') as fp:
+        with open(self.pyc_path, 'rb') as fp:
             flags = importlib._bootstrap_external._classify_pyc(
                 fp.read(), 'test', {})
         self.assertEqual(flags, 0b11)
         py_compile.compile(
             self.source_path,
+            self.pyc_path,
             invalidation_mode=py_compile.PycInvalidationMode.UNCHECKED_HASH,
         )
-        with open(self.cache_path, 'rb') as fp:
+        with open(self.pyc_path, 'rb') as fp:
             flags = importlib._bootstrap_external._classify_pyc(
                 fp.read(), 'test', {})
         self.assertEqual(flags, 0b1)
@@ -205,11 +200,11 @@ class PyCompileTestsBase:
                                   'tokenizedata',
                                   'bad_coding2.py')
         with support.captured_stderr() as stderr:
-            self.assertIsNone(py_compile.compile(bad_coding, doraise=False, quiet=2))
-            self.assertIsNone(py_compile.compile(bad_coding, doraise=True, quiet=2))
+            self.assertIsNone(py_compile.compile(bad_coding, self.pyc_path, doraise=False, quiet=2))
+            self.assertIsNone(py_compile.compile(bad_coding, self.pyc_path, doraise=True, quiet=2))
             self.assertEqual(stderr.getvalue(), '')
             with self.assertRaises(py_compile.PyCompileError):
-                py_compile.compile(bad_coding, doraise=True, quiet=1)
+                py_compile.compile(bad_coding, self.pyc_path, doraise=True, quiet=1)
 
     def test_utf7_decoded_cr_compiles(self):
         with open(self.source_path, 'wb') as file:
@@ -239,8 +234,12 @@ class PyCompileCLITestCase(unittest.TestCase):
     def setUp(self):
         self.directory = tempfile.mkdtemp()
         self.source_path = os.path.join(self.directory, '_test.py')
-        self.cache_path = importlib.util.cache_from_source(self.source_path,
-                                optimization='' if __debug__ else 1)
+        try:
+            self.cache_path = importlib.util.cache_from_source(self.source_path,
+                                    optimization='' if __debug__ else 1)
+        except NotImplementedError:
+            # py_compile.main() assumes legacy pyc path if there is no cache_tag
+            self.cache_path = self.source_path + 'c'
         with open(self.source_path, 'w') as file:
             file.write('x = 123\n')
 

Lib/test/test_tempfile.py

@@ -14,6 +14,7 @@ import weakref
 import gc
 import shutil
 import subprocess
+import sysconfig
 from unittest import mock
 
 import unittest
@@ -1848,6 +1849,54 @@ class TestTemporaryDirectory(BaseTestCase):
                 new_flags = os.stat(dir1).st_flags
                 self.assertEqual(new_flags, old_flags)
 
+    @os_helper.skip_unless_symlink
+    @os_helper.skip_unless_working_chmod
+    @support.requires_non_root_user
+    @unittest.skipIf(support.is_emscripten, 'Fails due to Emscripten bug:'
+                                            'emscripten-core/emscripten#27761')
+    @unittest.skipUnless(shutil.rmtree.avoids_symlink_attacks,
+                         'requires the fd based implementation of rmtree()')
+    def test_cleanup_with_symlink_race(self):
+        # cleanup() should not operate on files outside of the temporary
+        # directory when a directory is replaced with a symlink while it
+        # recovers from a PermissionError (CVE-2026-12345).
+        with self.do_create(recurse=0) as target:
+            target_file = os.path.join(target, 'file1')
+            open(target_file, 'wb').close()
+            target_mode = os.stat(target_file).st_mode
+
+            d1 = self.do_create(recurse=0)
+            dir1 = os.path.join(d1.name, 'dir1')
+            os.mkdir(dir1)
+            open(os.path.join(dir1, 'file1'), 'wb').close()
+            # Removing contents of dir1 fails with a PermissionError, and
+            # dir1 is replaced with a symlink to target at the very moment
+            # cleanup() starts to recover from that error.
+            os.chmod(dir1, 0o500)
+            unlink = os.unlink
+            def hook(path, *, dir_fd=None):
+                try:
+                    return unlink(path, dir_fd=dir_fd)
+                except PermissionError:
+                    if not os.path.islink(dir1):
+                        os.chmod(dir1, 0o700)
+                        os.rename(dir1, dir1 + '_moved')
+                        os.symlink(target, dir1)
+                    raise
+            try:
+                with mock.patch('os.unlink', hook):
+                    with contextlib.suppress(OSError):
+                        d1.cleanup()
+            finally:
+                if os.path.islink(dir1):
+                    os.unlink(dir1)
+                    os.rename(dir1 + '_moved', dir1)
+                os.chmod(dir1, 0o700)
+                d1.cleanup()
+
+            self.assertTrue(os.path.exists(target_file))
+            self.assertEqual(os.stat(target_file).st_mode, target_mode)
+
     @support.cpython_only
     def test_del_on_collection(self):
         # A TemporaryDirectory is deleted when garbage collected
@@ -2020,6 +2069,29 @@ class TestTemporaryDirectory(BaseTestCase):
                     d.cleanup()
                 self.assertFalse(os.path.exists(d.name))
 
+    @support.subTests('ignore_errors', (True, False))
+    def test_parent_mode_preserved(self, ignore_errors):
+        # Test that cleanup does not touch the parent directory,
+        # even if that prevents removal.
+        for mode in range(8):
+            mode <<= 6
+            with self.subTest(mode=format(mode, '03o')):
+                outer = self.do_create()
+                with outer:
+                    d = self.do_create(dir=outer.name, dirs=2, files=2,
+                                       ignore_cleanup_errors=ignore_errors)
+                    with d:
+                        os.chmod(outer.name, mode)
+                        orig_mode = os.stat(outer.name).st_mode
+                        try:
+                            d.cleanup()
+                        except PermissionError:
+                            if ignore_errors:
+                                raise
+                        self.assertEqual(os.stat(outer.name).st_mode, orig_mode)
+                        outer.cleanup()
+                self.assertFalse(os.path.exists(outer.name))
+
     def check_flags(self, flags):
         # skip the test if these flags are not supported (ex: FreeBSD 13)
         filename = os_helper.TESTFN
@@ -2057,5 +2129,17 @@ class TestTemporaryDirectory(BaseTestCase):
         self.assertTrue(os.path.exists(working_dir))
         shutil.rmtree(working_dir)
 
+    @unittest.skipUnless(
+        sysconfig.get_config_var('PY_SUPPORT_TIER')
+        and sysconfig.get_config_var('PY_SUPPORT_TIER') <= 3,
+        'regression test for supported platforms')
+    @unittest.skipIf(support.MS_WINDOWS, 'dirfd not used on Windows')
+    @unittest.skipIf(support.is_wasi, 'WASI has no chmod')
+    def test_cleanup_safe(self):
+        """Verify that cleanup uses the safer code path"""
+        # This is a regression test. Feel free to add exceptions for new
+        # platforms, but don't forget to update the docs.
+        self.assertTrue(tempfile._rmtree_use_dir_fd)
+
 if __name__ == "__main__":
     unittest.main()

Lib/test/test_zipfile/__main__.py

@@ -1,6 +1,6 @@
 import unittest
 
-from . import load_tests  # noqa: F401
+from . import load_tests
 
 
 if __name__ == "__main__":

Lib/test/test_zipfile/_path/__init__.py

@@ -1 +0,0 @@
-

Lib/test/test_zipfile/_path/test_path.py

@@ -317,7 +317,7 @@ class TestPath(unittest.TestCase):
     HUGE_ZIPFILE_NUM_ENTRIES = 2**13
 
     def huge_zipfile(self):
-        """Create a read-only zipfile with a huge number of entries entries."""
+        """Create a read-only zipfile with a huge number of entries."""
         strm = io.BytesIO()
         zf = zipfile.ZipFile(strm, "w")
         for entry in map(str, range(self.HUGE_ZIPFILE_NUM_ENTRIES)):

Lib/test/test_zipfile/test_core.py

@@ -1,6 +1,7 @@
 import _pyio
 import array
 import contextlib
+import errno
 import importlib.util
 import io
 import itertools
@@ -20,14 +21,15 @@ from tempfile import TemporaryFile
 from random import randint, random, randbytes
 
 from test import archiver_tests
-from test.support import script_helper, os_helper
+from test.support import script_helper
 from test.support import (
     findfile, requires_zlib, requires_bz2, requires_lzma,
     requires_zstd, captured_stdout, captured_stderr, requires_subprocess,
     cpython_only
 )
 from test.support.os_helper import (
-    TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count, FakePath
+    TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count, FakePath,
+    with_source_date_epoch, without_source_date_epoch,
 )
 from test.support.import_helper import ensure_lazy_imports
 
@@ -654,6 +656,12 @@ class StoredTestsWithSourceFile(AbstractTestsWithSourceFile,
             os.utime(TESTFN, (ts, ts))
         except OverflowError:
             self.skipTest('Host fs cannot set timestamp to required value.')
+        except OSError as exc:
+            # Some file systems (e.g. UFS and ZFS on illumos) do not
+            # support timestamps that do not fit in 32 bits.
+            if exc.errno != errno.EOVERFLOW:
+                raise
+            self.skipTest('Host fs cannot set timestamp to required value.')
 
         mtime_ns = os.stat(TESTFN).st_mtime_ns
         if mtime_ns != (4386268800 * 10**9):
@@ -1880,29 +1888,24 @@ class OtherTests(unittest.TestCase):
                 zinfo.flag_bits |= zipfile._MASK_USE_DATA_DESCRIPTOR  # Include an extended local header.
                 orig_zip.writestr(zinfo, data)
 
+    @with_source_date_epoch(epoch=1735715999)
     def test_write_with_source_date_epoch(self):
-        with os_helper.EnvironmentVarGuard() as env:
-            # Set the SOURCE_DATE_EPOCH environment variable to a specific timestamp
-            env['SOURCE_DATE_EPOCH'] = "1735715999"
+        with zipfile.ZipFile(TESTFN, "w") as zf:
+            zf.writestr("test_source_date_epoch.txt", "Testing SOURCE_DATE_EPOCH")
 
-            with zipfile.ZipFile(TESTFN, "w") as zf:
-                zf.writestr("test_source_date_epoch.txt", "Testing SOURCE_DATE_EPOCH")
-
-            with zipfile.ZipFile(TESTFN, "r") as zf:
-                zip_info = zf.getinfo("test_source_date_epoch.txt")
-                expected_utc = (2025, 1, 1, 7, 19, 58)
-                self.assertEqual(zip_info.date_time, expected_utc)
+        with zipfile.ZipFile(TESTFN, "r") as zf:
+            zip_info = zf.getinfo("test_source_date_epoch.txt")
+            expected_utc = (2025, 1, 1, 7, 19, 58)
+            self.assertEqual(zip_info.date_time, expected_utc)
 
+    @without_source_date_epoch
     def test_write_without_source_date_epoch(self):
-        with os_helper.EnvironmentVarGuard() as env:
-            del env['SOURCE_DATE_EPOCH']
+        with zipfile.ZipFile(TESTFN, "w") as zf:
+            zf.writestr("test_no_source_date_epoch.txt", "Testing without SOURCE_DATE_EPOCH")
 
-            with zipfile.ZipFile(TESTFN, "w") as zf:
-                zf.writestr("test_no_source_date_epoch.txt", "Testing without SOURCE_DATE_EPOCH")
-
-            with zipfile.ZipFile(TESTFN, "r") as zf:
-                zip_info = zf.getinfo("test_no_source_date_epoch.txt")
-                self.assertTimestampAlmostEqual(time.localtime(), zip_info.date_time, tolerance=2)
+        with zipfile.ZipFile(TESTFN, "r") as zf:
+            zip_info = zf.getinfo("test_no_source_date_epoch.txt")
+            self.assertTimestampAlmostEqual(time.localtime(), zip_info.date_time, tolerance=2)
 
     def assertTimestampAlmostEqual(self, time1, time2, tolerance):
         import datetime
@@ -2720,6 +2723,115 @@ class OtherTests(unittest.TestCase):
         unlink(TESTFN2)
 
 
+class AbstractBoundedDecompressTests:
+    # ZipExtFile._read1() bounds the output of each decompress() call so that a
+    # small member declaring a large uncompressed size cannot expand into one
+    # unbounded read.
+    def test_read1_output_is_bounded(self):
+        buf = io.BytesIO()
+        with zipfile.ZipFile(buf, "w", compression=self.compression) as zf:
+            zf.writestr("big", b"\0" * (4 * 1024 * 1024))
+        with zipfile.ZipFile(io.BytesIO(buf.getvalue())) as zf:
+            with zf.open("big") as f:
+                self.assertLessEqual(len(f._read1(100)), f.MIN_READ_SIZE)
+
+
+class StoredBoundedDecompressTests(AbstractBoundedDecompressTests,
+                                   unittest.TestCase):
+    compression = zipfile.ZIP_STORED
+
+
+@requires_zlib()
+class DeflateBoundedDecompressTests(AbstractBoundedDecompressTests,
+                                    unittest.TestCase):
+    compression = zipfile.ZIP_DEFLATED
+
+
+@requires_bz2()
+class Bzip2BoundedDecompressTests(AbstractBoundedDecompressTests,
+                                  unittest.TestCase):
+    compression = zipfile.ZIP_BZIP2
+
+
+@requires_lzma()
+class LzmaBoundedDecompressTests(AbstractBoundedDecompressTests,
+                                 unittest.TestCase):
+    compression = zipfile.ZIP_LZMA
+
+
+@requires_zstd()
+class ZstdBoundedDecompressTests(AbstractBoundedDecompressTests,
+                                 unittest.TestCase):
+    compression = zipfile.ZIP_ZSTANDARD
+
+
+class MonkeypatchedDecompressorTests(unittest.TestCase):
+    # Some third-party projects monkey-patch _get_decompressor() to add
+    # additional compression schemes. This can break at any time as the
+    # internal compressor objects change.
+    # To protect users, we try to keep this case working.
+    # See also: GH-156002 and GH-113767.
+    COMPRESSION = 99
+
+    class Compressor:
+        """Compressor with only the original BZ2Compressor API"""
+        def compress(self, data):
+            return data.swapcase()
+
+        def flush(self):
+            return b''
+
+    class Decompressor:
+        """Decompressor with only the 3.3+ BZ2Decompressor API"""
+        eof = False
+
+        def decompress(self, data):
+            return data.swapcase()
+
+    def setUp(self):
+        orig_check_compression = zipfile._check_compression
+        orig_get_compressor = zipfile._get_compressor
+        orig_get_decompressor = zipfile._get_decompressor
+
+        def check_compression(compression):
+            if compression != self.COMPRESSION:
+                orig_check_compression(compression)
+
+        def get_compressor(compress_type, compresslevel=None):
+            if compress_type == self.COMPRESSION:
+                return self.Compressor()
+            return orig_get_compressor(compress_type, compresslevel)
+
+        def get_decompressor(compress_type):
+            if compress_type == self.COMPRESSION:
+                return self.Decompressor()
+            return orig_get_decompressor(compress_type)
+
+        self.enterContext(mock.patch.object(
+            zipfile, '_check_compression', check_compression))
+        self.enterContext(mock.patch.object(
+            zipfile, '_get_compressor', get_compressor))
+        self.enterContext(mock.patch.object(
+            zipfile, '_get_decompressor', get_decompressor))
+
+    def test_roundtrip_monkeypatched_decompressor(self):
+        data = bytes(range(256)) * 8
+        buf = io.BytesIO()
+        with zipfile.ZipFile(buf, "w", compression=self.COMPRESSION) as zf:
+            zf.writestr("member", data)
+        self.assertIn(data.swapcase(), buf.getvalue())
+        with zipfile.ZipFile(io.BytesIO(buf.getvalue())) as zf:
+            self.assertEqual(zf.read("member"), data)
+            with zf.open("member") as f:
+                self.assertEqual(f.read(100), data[:100])
+                self.assertEqual(f.read1(100), data[100:200])
+                f.seek(-100, os.SEEK_END)
+                self.assertEqual(f.read(), data[-100:])
+                # Rewinding past the read buffer re-creates the decompressor.
+                f.seek(0)
+                self.assertEqual(f.read(), data)
+
+
 class AbstractBadCrcTests:
     def test_testzip_with_bad_crc(self):
         """Tests that files with bad CRCs return their name from testzip."""

Lib/zipfile/__init__.py

@@ -786,7 +786,16 @@ class LZMADecompressor:
         self._unconsumed = b''
         self.eof = False
 
-    def decompress(self, data):
+    @property
+    def needs_input(self):
+        # While the LZMA properties header is still being buffered, more input
+        # is required; afterwards defer to the wrapped decompressor so a bounded
+        # decompress() call can be drained across reads.
+        if self._decomp is None:
+            return True
+        return self._decomp.needs_input
+
+    def decompress(self, data, max_length=-1):
         if self._decomp is None:
             self._unconsumed += data
             if len(self._unconsumed) <= 4:
@@ -802,7 +811,7 @@ class LZMADecompressor:
             data = self._unconsumed[4 + psize:]
             del self._unconsumed
 
-        result = self._decomp.decompress(data)
+        result = self._decomp.decompress(data, max_length)
         self.eof = self._decomp.eof
         return result
 
@@ -1171,8 +1180,15 @@ class ZipExtFile(io.BufferedIOBase):
             data = self._decompressor.unconsumed_tail
             if n > len(data):
                 data += self._read2(n - len(data))
-        else:
+        elif self._compress_type == ZIP_STORED:
             data = self._read2(n)
+        else:
+            # bzip2/lzma/zstd: a bounded decompress() call may leave input
+            # buffered inside the decompressor; drain that before reading more.
+            if getattr(self._decompressor, "needs_input", True):
+                data = self._read2(n)
+            else:
+                data = b''
 
         if self._compress_type == ZIP_STORED:
             self._eof = self._compress_left <= 0
@@ -1185,8 +1201,17 @@ class ZipExtFile(io.BufferedIOBase):
             if self._eof:
                 data += self._decompressor.flush()
         else:
-            data = self._decompressor.decompress(data)
-            self._eof = self._decompressor.eof or self._compress_left <= 0
+            # Bound the output of a single decompress() call (mirroring the
+            # DEFLATE path above) so that a small compressed member cannot
+            # expand into one unbounded read.
+            try:
+                data = self._decompressor.decompress(data, max(n, self.MIN_READ_SIZE))
+            except TypeError:
+                # See MonkeypatchedDecompressorTests in test_core.py
+                data = self._decompressor.decompress(data)
+            self._eof = (self._decompressor.eof or
+                         self._compress_left <= 0 and
+                         getattr(self._decompressor, "needs_input", True))
 
         data = data[:self._left]
         self._left -= len(data)
@@ -2242,10 +2267,10 @@ class PyZipFile(ZipFile):
                     basename = name
                 if self.debug:
                     print("Adding package in", pathname, "as", basename)
-                fname, arcname = self._get_codename(initname[0:-3], basename)
+                arcname, bytecode = self._get_code(initname[0:-3], basename)
                 if self.debug:
                     print("Adding", arcname)
-                self.write(fname, arcname)
+                self.writestr(arcname, bytecode)
                 dirlist = sorted(os.listdir(pathname))
                 dirlist.remove("__init__.py")
                 # Add all *.py files and package subdirectories
@@ -2262,11 +2287,10 @@ class PyZipFile(ZipFile):
                             if self.debug:
                                 print('file %r skipped by filterfunc' % path)
                             continue
-                        fname, arcname = self._get_codename(path[0:-3],
-                                                            basename)
+                        arcname, bytecode = self._get_code(path[0:-3], basename)
                         if self.debug:
                             print("Adding", arcname)
-                        self.write(fname, arcname)
+                        self.writestr(arcname, bytecode)
             else:
                 # This is NOT a package directory, add its files at top level
                 if self.debug:
@@ -2279,108 +2303,65 @@ class PyZipFile(ZipFile):
                             if self.debug:
                                 print('file %r skipped by filterfunc' % path)
                             continue
-                        fname, arcname = self._get_codename(path[0:-3],
-                                                            basename)
+                        arcname, bytecode = self._get_code(path[0:-3], basename)
                         if self.debug:
                             print("Adding", arcname)
-                        self.write(fname, arcname)
+                        self.writestr(arcname, bytecode)
         else:
             if pathname[-3:] != ".py":
                 raise RuntimeError(
                     'Files added with writepy() must end with ".py"')
-            fname, arcname = self._get_codename(pathname[0:-3], basename)
+            arcname, bytecode = self._get_code(pathname[0:-3], basename)
             if self.debug:
                 print("Adding file", arcname)
-            self.write(fname, arcname)
+            self.writestr(arcname, bytecode)
 
-    def _get_codename(self, pathname, basename):
-        """Return (filename, archivename) for the path.
+    def _get_code(self, pathname, basename):
+        """Return (arcname, bytecode) for the path.
 
-        Given a module name path, return the correct file path and
-        archive name, compiling if necessary.  For example, given
-        /python/lib/string, return (/python/lib/string.pyc, string).
+        Given a module name path, return the bytecode and archive
+        name.  For example, given /python/lib/string, return
+        ('string', b'<bytecode of string>').
         """
-        def _compile(file, optimize=-1):
-            import py_compile
-            if self.debug:
-                print("Compiling", file)
-            try:
-                py_compile.compile(file, doraise=True, optimize=optimize)
-            except py_compile.PyCompileError as err:
-                print(err.msg)
-                return False
-            return True
+        import importlib._bootstrap_external
+        import importlib.machinery
 
         file_py  = pathname + ".py"
         file_pyc = pathname + ".pyc"
-        pycache_opt0 = importlib.util.cache_from_source(file_py, optimization='')
-        pycache_opt1 = importlib.util.cache_from_source(file_py, optimization=1)
-        pycache_opt2 = importlib.util.cache_from_source(file_py, optimization=2)
-        if self._optimize == -1:
-            # legacy mode: use whatever file is present
-            if (os.path.isfile(file_pyc) and
-                  os.stat(file_pyc).st_mtime >= os.stat(file_py).st_mtime):
-                # Use .pyc file.
-                arcname = fname = file_pyc
-            elif (os.path.isfile(pycache_opt0) and
-                  os.stat(pycache_opt0).st_mtime >= os.stat(file_py).st_mtime):
-                # Use the __pycache__/*.pyc file, but write it to the legacy pyc
-                # file name in the archive.
-                fname = pycache_opt0
-                arcname = file_pyc
-            elif (os.path.isfile(pycache_opt1) and
-                  os.stat(pycache_opt1).st_mtime >= os.stat(file_py).st_mtime):
-                # Use the __pycache__/*.pyc file, but write it to the legacy pyc
-                # file name in the archive.
-                fname = pycache_opt1
-                arcname = file_pyc
-            elif (os.path.isfile(pycache_opt2) and
-                  os.stat(pycache_opt2).st_mtime >= os.stat(file_py).st_mtime):
-                # Use the __pycache__/*.pyc file, but write it to the legacy pyc
-                # file name in the archive.
-                fname = pycache_opt2
-                arcname = file_pyc
-            else:
-                # Compile py into PEP 3147 pyc file.
-                if _compile(file_py):
-                    if sys.flags.optimize == 0:
-                        fname = pycache_opt0
-                    elif sys.flags.optimize == 1:
-                        fname = pycache_opt1
-                    else:
-                        fname = pycache_opt2
-                    arcname = file_pyc
-                else:
-                    fname = arcname = file_py
+        archivename = os.path.split(file_pyc)[1]
+
+        loader = importlib.machinery.SourceFileLoader('<py_compile>', file_py)
+        source_bytes = loader.get_data(file_py)
+        try:
+            if self.debug:
+                print("Compiling", file_py)
+            code = loader.source_to_code(source_bytes, archivename)
+        except Exception as err:
+            # Historically, this function prints messages here rather than raising
+            # (see test_zipfile.test_write_filtered_python_package)
+            from py_compile import PyCompileError
+            print(PyCompileError(type(err), err, file_py).msg)
+
+            archivename = os.path.split(file_py)[1]
+            bytecode = source_bytes
         else:
-            # new mode: use given optimization level
-            if self._optimize == 0:
-                fname = pycache_opt0
-                arcname = file_pyc
-            else:
-                arcname = file_pyc
-                if self._optimize == 1:
-                    fname = pycache_opt1
-                elif self._optimize == 2:
-                    fname = pycache_opt2
-                else:
-                    msg = "invalid value for 'optimize': {!r}".format(self._optimize)
-                    raise ValueError(msg)
-            if not (os.path.isfile(fname) and
-                    os.stat(fname).st_mtime >= os.stat(file_py).st_mtime):
-                if not _compile(file_py, optimize=self._optimize):
-                    fname = arcname = file_py
-        archivename = os.path.split(arcname)[1]
+            # Historically this function has used timestamp comparisons, so we
+            # keep using it until someone makes that specific improvement.
+            source_stats = loader.path_stats(file_py)
+            bytecode = importlib._bootstrap_external._code_to_timestamp_pyc(
+                code, source_stats['mtime'], source_stats['size'])
+
         if basename:
             archivename = "%s/%s" % (basename, archivename)
-        return (fname, archivename)
+
+        return archivename, bytecode
 
 
 def main(args=None):
     import argparse
 
     description = 'A simple command-line interface for zipfile module.'
-    parser = argparse.ArgumentParser(description=description, color=True)
+    parser = argparse.ArgumentParser(description=description)
     group = parser.add_mutually_exclusive_group(required=True)
     group.add_argument('-l', '--list', metavar='<zipfile>',
                        help='Show listing of a zipfile')
@@ -2393,7 +2374,7 @@ def main(args=None):
     group.add_argument('-t', '--test', metavar='<zipfile>',
                        help='Test if a zipfile is valid')
     parser.add_argument('--metadata-encoding', metavar='<encoding>',
-                        help='Specify encoding of member names for -l, -e and -t')
+                        help='Specify encoding of member names for `-l`, `-e` and `-t`')
     args = parser.parse_args(args)
 
     encoding = args.metadata_encoding

Lib/zipfile/_path/__init__.py

@@ -7,7 +7,7 @@ https://github.com/python/importlib_metadata/wiki/Development-Methodology
 for more detail.
 """
 
-import contextlib
+import functools
 import io
 import itertools
 import pathlib
@@ -17,6 +17,7 @@ import stat
 import sys
 import zipfile
 
+from ._functools import save_method_args
 from .glob import Translator
 
 __all__ = ['Path']
@@ -85,13 +86,12 @@ class InitializedState:
     Mix-in to save the initialization state for pickling.
     """
 
+    @save_method_args
     def __init__(self, *args, **kwargs):
-        self.__args = args
-        self.__kwargs = kwargs
         super().__init__(*args, **kwargs)
 
     def __getstate__(self):
-        return self.__args, self.__kwargs
+        return self._saved___init__.args, self._saved___init__.kwargs
 
     def __setstate__(self, state):
         args, kwargs = state
@@ -180,16 +180,19 @@ class FastLookup(CompleteDirs):
     """
 
     def namelist(self):
-        with contextlib.suppress(AttributeError):
-            return self.__names
-        self.__names = super().namelist()
-        return self.__names
+        return self._namelist
+
+    @functools.cached_property
+    def _namelist(self):
+        return super().namelist()
 
     def _name_set(self):
-        with contextlib.suppress(AttributeError):
-            return self.__lookup
-        self.__lookup = super()._name_set()
-        return self.__lookup
+        return self._name_set_prop
+
+    @functools.cached_property
+    def _name_set_prop(self):
+        return super()._name_set()
+
 
 def _extract_text_encoding(encoding=None, *args, **kwargs):
     # compute stack level so that the caller of the caller sees any warning.

Lib/zipfile/_path/_functools.py

@@ -0,0 +1,20 @@
+import collections
+import functools
+
+
+# from jaraco.functools 4.0.2
+def save_method_args(method):
+    """
+    Wrap a method such that when it is called, the args and kwargs are
+    saved on the method.
+    """
+    args_and_kwargs = collections.namedtuple('args_and_kwargs', 'args kwargs')  # noqa: PYI024
+
+    @functools.wraps(method)
+    def wrapper(self, /, *args, **kwargs):
+        attr_name = '_saved_' + method.__name__
+        attr = args_and_kwargs(args, kwargs)
+        setattr(self, attr_name, attr)
+        return method(self, *args, **kwargs)
+
+    return wrapper