diff --git a/Lib/compileall.py b/Lib/compileall.py index 67fe370451e..812a496611e 100644 --- a/Lib/compileall.py +++ b/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') diff --git a/Lib/glob.py b/Lib/glob.py index 173272ba38f..07b9d3eb952 100644 --- a/Lib/glob.py +++ b/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 @@ def concat_path(path, text): """ 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 @@ def recursive_selector(self, part, parts): 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 @@ def select_recursive_step(stack, match_pos): 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 @@ def scandir(path): 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. diff --git a/Lib/pathlib/__init__.py b/Lib/pathlib/__init__.py index 0d763d1f0dc..8dd16c6225b 100644 --- a/Lib/pathlib/__init__.py +++ b/Lib/pathlib/__init__.py @@ -12,10 +12,13 @@ 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 @@ 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 @@ def __reduce__(self): 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 @@ def __str__(self): 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 @@ def relative_to(self, other, *, walk_up=False): """ 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 @@ def is_absolute(self): 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 @@ def info(self): 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 @@ def _filter_trailing_slash(self, paths): 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 @@ def touch(self, mode=0o666, exist_ok=True): 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 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False): 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 @@ def _delete(self): 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 @@ def _copy_from(self, source, follow_symlinks=True, preserve_metadata=False): 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 @@ def _copy_from_file(self, source, preserve_metadata=False): # 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): """ diff --git a/Lib/pathlib/_os.py b/Lib/pathlib/_os.py index 039836941dd..79a1969d5f8 100644 --- a/Lib/pathlib/_os.py +++ b/Lib/pathlib/_os.py @@ -4,7 +4,6 @@ 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 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: + raise ValueError(f'invalid mode: {mode}') 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") + stream = TextIOWrapper(stream, encoding, errors, newline) + return stream + +def vfspath(obj): + """ + Return the string representation of a virtual path object. + """ + cls = type(obj) try: - attr = getattr(cls, f'__open_{mode}b__') + vfspath_method = cls.__vfspath__ except AttributeError: - pass + cls_name = cls.__name__ + raise TypeError(f"expected JoinablePath object, not {cls_name}") from None else: - stream = attr(path, buffering) - if text: - stream = TextIOWrapper(stream, encoding, errors, newline) - return stream - - raise TypeError(f"{cls.__name__} can't be opened with mode {mode!r}") + 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 diff --git a/Lib/pathlib/types.py b/Lib/pathlib/types.py index d8f5c34a1a7..f21ce077454 100644 --- a/Lib/pathlib/types.py +++ b/Lib/pathlib/types.py @@ -11,9 +11,10 @@ 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 @@ def is_file(self, *, follow_symlinks: bool = True) -> bool: ... 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 @@ def with_segments(self, *pathsegments): 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 @@ def with_name(self, name): 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 @@ def with_suffix(self, suffix): 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 @@ def joinpath(self, *pathsegments): 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 @@ def parent(self): 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 @@ def parents(self): 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 @@ def full_match(self, pattern): 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 @@ def info(self): 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 @@ def read_bytes(self): """ 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 @@ def read_text(self, encoding=None, errors=None, newline=None): # 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 @@ def mkdir(self): 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 @@ def write_bytes(self, data): """ # 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 @@ def write_text(self, data, encoding=None, errors=None, newline=None): 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 @@ def _copy_from(self, source, follow_symlinks=True): 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 @@ def _copy_from(self, source, follow_symlinks=True): 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) diff --git a/Lib/py_compile.py b/Lib/py_compile.py index 43d8ec90ffb..7ca479141e0 100644 --- a/Lib/py_compile.py +++ b/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) diff --git a/Lib/tempfile.py b/Lib/tempfile.py index 6c036d2fd51..d14c187a275 100644 --- a/Lib/tempfile.py +++ b/Lib/tempfile.py @@ -43,6 +43,7 @@ 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 @@ def TemporaryFile(mode='w+b', buffering=-1, encoding=None, 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 @@ def opener(*args): 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 @@ def __init__(self, suffix=None, prefix=None, dir=None, 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 @@ def onexc(func, path, exc): 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 @@ def onexc(func, path, exc): # 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): diff --git a/Lib/test/test_compileall.py b/Lib/test/test_compileall.py index be5e7d7feee..a612a729014 100644 --- a/Lib/test/test_compileall.py +++ b/Lib/test/test_compileall.py @@ -28,11 +28,15 @@ 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 '' diff --git a/Lib/test/test_glob.py b/Lib/test/test_glob.py index d0ed5129253..9e4e82b2542 100644 --- a/Lib/test/test_glob.py +++ b/Lib/test/test_glob.py @@ -4,7 +4,6 @@ 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 @@ def test_glob_many_open_files(self): 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')) diff --git a/Lib/test/test_pathlib/support/lexical_path.py b/Lib/test/test_pathlib/support/lexical_path.py index f29a521af9b..fd7fbf283a6 100644 --- a/Lib/test/test_pathlib/support/lexical_path.py +++ b/Lib/test/test_pathlib/support/lexical_path.py @@ -9,9 +9,10 @@ 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 @@ def __init__(self, *pathsegments): 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) diff --git a/Lib/test/test_pathlib/support/local_path.py b/Lib/test/test_pathlib/support/local_path.py index d481fd45ead..ddfd6fd4195 100644 --- a/Lib/test/test_pathlib/support/local_path.py +++ b/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) diff --git a/Lib/test/test_pathlib/support/zip_path.py b/Lib/test/test_pathlib/support/zip_path.py index 2905260c9df..90b939b6a59 100644 --- a/Lib/test/test_pathlib/support/zip_path.py +++ b/Lib/test/test_pathlib/support/zip_path.py @@ -16,9 +16,10 @@ 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 @@ def teardown(self, root): 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 @@ def create_hierarchy(self, p): 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 @@ def __init__(self, *pathsegments, zip_file): 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 @@ def with_segments(self, *pathsegments): @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 @@ def __init__(self, *pathsegments, zip_file): 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) diff --git a/Lib/test/test_pathlib/test_join.py b/Lib/test/test_pathlib/test_join.py index 6b51a09e5ac..4630210e492 100644 --- a/Lib/test/test_pathlib/test_join.py +++ b/Lib/test/test_pathlib/test_join.py @@ -376,6 +376,61 @@ def test_with_suffix(self): 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 diff --git a/Lib/test/test_pathlib/test_join_windows.py b/Lib/test/test_pathlib/test_join_windows.py index 2cc634f25ef..5afefa61b58 100644 --- a/Lib/test/test_pathlib/test_join_windows.py +++ b/Lib/test/test_pathlib/test_join_windows.py @@ -8,6 +8,11 @@ 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 @@ def test_div(self): 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 diff --git a/Lib/test/test_pathlib/test_pathlib.py b/Lib/test/test_pathlib/test_pathlib.py index a1ea69a6b90..aff66c8efed 100644 --- a/Lib/test/test_pathlib/test_pathlib.py +++ b/Lib/test/test_pathlib/test_pathlib.py @@ -20,6 +20,8 @@ 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 @@ 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 @@ def test_is_notimplemented(self): 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 @@ def test_with_stem_empty(self): 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 @@ def raiser(*args, **kwargs): 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 @@ def test_owner(self): 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 @@ def test_group(self): 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 @@ def my_mkdir(path, mode=0o777): 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 @@ def test_is_socket_true(self): 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()) diff --git a/Lib/test/test_pathlib/test_read.py b/Lib/test/test_pathlib/test_read.py index 482203c290a..51eede9d9eb 100644 --- a/Lib/test/test_pathlib/test_read.py +++ b/Lib/test/test_pathlib/test_read.py @@ -13,10 +13,10 @@ 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 @@ def test_is_readable(self): 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 @@ def test_open_r(self): 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 @@ def test_info_is_symlink(self): 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) diff --git a/Lib/test/test_pathlib/test_write.py b/Lib/test/test_pathlib/test_write.py index 15054e804ec..fa2a81a47b3 100644 --- a/Lib/test/test_pathlib/test_write.py +++ b/Lib/test/test_pathlib/test_write.py @@ -13,10 +13,10 @@ 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 @@ def test_is_writable(self): 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 @@ def test_open_w(self): 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' diff --git a/Lib/test/test_py_compile.py b/Lib/test/test_py_compile.py index acd2dde8867..6eb01ce7bbc 100644 --- a/Lib/test/test_py_compile.py +++ b/Lib/test/test_py_compile.py @@ -1,4 +1,3 @@ -import functools import importlib.util import os import py_compile @@ -11,43 +10,7 @@ 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 @@ def setUp(self): 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 @@ def tearDown(self): 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 @@ def test_do_not_overwrite_nonregular_files(self): 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 @@ def test_bad_coding(self): '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 @@ def test_source_date_epoch(self): 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 @@ def test_double_dot_no_clobber(self): 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 @@ def test_optimization_path(self): 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 @@ def test_quiet(self): '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') diff --git a/Lib/test/test_tempfile.py b/Lib/test/test_tempfile.py index cf8b1037194..4bb7d95a903 100644 --- a/Lib/test/test_tempfile.py +++ b/Lib/test/test_tempfile.py @@ -14,6 +14,7 @@ import gc import shutil import subprocess +import sysconfig from unittest import mock import unittest @@ -1848,6 +1849,54 @@ def test(target, target_is_directory): 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 @@ def test_modes(self): 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 @@ def test_delete_false(self): 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() diff --git a/Lib/test/test_zipfile/__main__.py b/Lib/test/test_zipfile/__main__.py index e25ac946edf..90da74ade38 100644 --- a/Lib/test/test_zipfile/__main__.py +++ b/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__": diff --git a/Lib/test/test_zipfile/_path/__init__.py b/Lib/test/test_zipfile/_path/__init__.py index 8b137891791..e69de29bb2d 100644 --- a/Lib/test/test_zipfile/_path/__init__.py +++ b/Lib/test/test_zipfile/_path/__init__.py @@ -1 +0,0 @@ - diff --git a/Lib/test/test_zipfile/_path/test_path.py b/Lib/test/test_zipfile/_path/test_path.py index 351d9eefeb0..e7931b6f394 100644 --- a/Lib/test/test_zipfile/_path/test_path.py +++ b/Lib/test/test_zipfile/_path/test_path.py @@ -317,7 +317,7 @@ def test_mutability(self, alpharep): 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)): diff --git a/Lib/test/test_zipfile/test_core.py b/Lib/test/test_zipfile/test_core.py index f9ad28ea04d..8f226826524 100644 --- a/Lib/test/test_zipfile/test_core.py +++ b/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 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 @@ def test_add_file_after_2107(self): 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 @@ def test_writestr_extended_local_header_issue1202(self): 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 @@ def tearDown(self): 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.""" diff --git a/Lib/zipfile/__init__.py b/Lib/zipfile/__init__.py index 2b6e6163b96..552ab17bd67 100644 --- a/Lib/zipfile/__init__.py +++ b/Lib/zipfile/__init__.py @@ -786,7 +786,16 @@ def __init__(self): 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 @@ def decompress(self, data): 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 @@ def _read1(self, n): 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 @@ def _read1(self, n): 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 @@ def writepy(self, pathname, basename="", filterfunc=None): 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 @@ def writepy(self, pathname, basename="", filterfunc=None): 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 @@ def writepy(self, pathname, basename="", filterfunc=None): 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''). """ - 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('', 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='', help='Show listing of a zipfile') @@ -2393,7 +2374,7 @@ def main(args=None): group.add_argument('-t', '--test', metavar='', help='Test if a zipfile is valid') parser.add_argument('--metadata-encoding', metavar='', - 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 diff --git a/Lib/zipfile/_path/__init__.py b/Lib/zipfile/_path/__init__.py index 80f5d607731..faae4c84cae 100644 --- a/Lib/zipfile/_path/__init__.py +++ b/Lib/zipfile/_path/__init__.py @@ -7,7 +7,7 @@ for more detail. """ -import contextlib +import functools import io import itertools import pathlib @@ -17,6 +17,7 @@ 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. diff --git a/Lib/zipfile/_path/_functools.py b/Lib/zipfile/_path/_functools.py new file mode 100644 index 00000000000..7390be21873 --- /dev/null +++ b/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