Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-27 02:38
📄 __init__.py from pyfakefs._version import __version__ # noqa: F401	56 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:56
📄 _version.py __version__ = "5.10.2"	23 B	2026-01-20 12:58
📄 fake_file.py # Copyright 2009 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Fake implementations for different file objects.""" import errno import io import os import sys import traceback from stat import ( S_IFREG, S_IFDIR, ) from types import TracebackType from typing import ( List, Optional, Callable, Union, Any, Dict, cast, AnyStr, NoReturn, Iterator, TextIO, Type, TYPE_CHECKING, ) from pyfakefs import helpers from pyfakefs.helpers import ( FakeStatResult, BinaryBufferIO, TextBufferIO, is_int_type, is_unicode_string, to_string, matching_string, real_encoding, AnyPath, AnyString, get_locale_encoding, _OpenModes, is_root, ) if TYPE_CHECKING: from pyfakefs.fake_filesystem import FakeFilesystem # Work around pyupgrade auto-rewriting `io.open()` to `open()`. io_open = io.open AnyFileWrapper = Union[ "FakeFileWrapper", "FakeDirWrapper", "StandardStreamWrapper", "FakePipeWrapper", ] AnyFile = Union["FakeFile", "FakeDirectory"] class FakeLargeFileIoException(Exception): """Exception thrown on unsupported operations for fake large files. Fake large files have a size with no real content. """ def __init__(self, file_path: str) -> None: super().__init__( "Read and write operations not supported for " "fake large file: %s" % file_path ) class FakeFile: """Provides the appearance of a real file. Attributes currently faked out: * `st_mode`: user-specified, otherwise S_IFREG * `st_ctime`: the time.time() timestamp of the file change time (updated each time a file's attributes is modified). * `st_atime`: the time.time() timestamp when the file was last accessed. * `st_mtime`: the time.time() timestamp when the file was last modified. * `st_size`: the size of the file * `st_nlink`: the number of hard links to the file * `st_ino`: the inode number - a unique number identifying the file * `st_dev`: a unique number identifying the (fake) file system device the file belongs to * `st_uid`: always set to USER_ID, which can be changed globally using `set_uid` * `st_gid`: always set to GROUP_ID, which can be changed globally using `set_gid` .. note:: The resolution for `st_ctime`, `st_mtime` and `st_atime` in the real file system depends on the used file system (for example it is only 1s for HFS+ and older Linux file systems, but much higher for ext4 and NTFS). This is currently ignored by pyfakefs, which uses the resolution of `time.time()`. Under Windows, `st_atime` is not updated for performance reasons by default. pyfakefs never updates `st_atime` under Windows, assuming the default setting. """ stat_types = ( "st_mode", "st_ino", "st_dev", "st_nlink", "st_uid", "st_gid", "st_size", "st_atime", "st_mtime", "st_ctime", "st_atime_ns", "st_mtime_ns", "st_ctime_ns", ) def __init__( self, name: AnyStr, st_mode: int = S_IFREG \| helpers.PERM_DEF_FILE, contents: Optional[AnyStr] = None, filesystem: Optional["FakeFilesystem"] = None, encoding: Optional[str] = None, errors: Optional[str] = None, side_effect: Optional[Callable[["FakeFile"], None]] = None, open_modes: Optional[_OpenModes] = None, ): """ Args: name: Name of the file/directory, without parent path information st_mode: The stat.S_IF* constant representing the file type (i.e. stat.S_IFREG, stat.S_IFDIR), and the file permissions. If no file type is set (e.g. permission flags only), a regular file type is assumed. contents: The contents of the filesystem object; should be a string or byte object for regular files, and a dict of other FakeFile or FakeDirectory objects with the file names as keys for FakeDirectory objects filesystem: The fake filesystem where the file is created. encoding: If contents is a unicode string, the encoding used for serialization. errors: The error mode used for encoding/decoding errors. side_effect: function handle that is executed when file is written, must accept the file object as an argument. open_modes: The modes the file was opened with (e.g. can read, write etc.) """ # to be backwards compatible regarding argument order, we raise on None if filesystem is None: raise ValueError("filesystem shall not be None") self.filesystem: "FakeFilesystem" = filesystem self._side_effect: Optional[Callable] = side_effect self.name: AnyStr = name # type: ignore[assignment] self.stat_result = FakeStatResult( filesystem.is_windows_fs, helpers.get_uid(), helpers.get_gid(), helpers.now(), ) if st_mode >> 12 == 0: st_mode \|= S_IFREG self.stat_result.st_mode = st_mode self.st_size: int = 0 self.encoding: Optional[str] = real_encoding(encoding) self.errors: str = errors or "strict" self._byte_contents: Optional[bytes] = self._encode_contents(contents) self.stat_result.st_size = ( len(self._byte_contents) if self._byte_contents is not None else 0 ) self.epoch: int = 0 self.parent_dir: Optional[FakeDirectory] = None # Linux specific: extended file system attributes self.xattr: Dict = {} self.opened_as: AnyString = "" self.open_modes = open_modes @property def byte_contents(self) -> Optional[bytes]: """Return the contents as raw byte array.""" return self._byte_contents @property def contents(self) -> Optional[str]: """Return the contents as string with the original encoding.""" if isinstance(self.byte_contents, bytes): return self.byte_contents.decode( self.encoding or get_locale_encoding(), errors=self.errors, ) return None @property def st_ctime(self) -> float: """Return the creation time of the fake file.""" return self.stat_result.st_ctime @st_ctime.setter def st_ctime(self, val: float) -> None: """Set the creation time of the fake file.""" self.stat_result.st_ctime = val @property def st_atime(self) -> float: """Return the access time of the fake file.""" return self.stat_result.st_atime @st_atime.setter def st_atime(self, val: float) -> None: """Set the access time of the fake file.""" self.stat_result.st_atime = val @property def st_mtime(self) -> float: """Return the modification time of the fake file.""" return self.stat_result.st_mtime @st_mtime.setter def st_mtime(self, val: float) -> None: """Set the modification time of the fake file.""" self.stat_result.st_mtime = val def set_large_file_size(self, st_size: int) -> None: """Sets the self.st_size attribute and replaces self.content with `None`. Provided specifically to simulate very large files without regards to their content (which wouldn't fit in memory). Note that read/write operations with such a file raise :py:class:`FakeLargeFileIoException`. Args: st_size: (int) The desired file size Raises: OSError: if ``st_size`` is not a non-negative integer, or if ``st_size`` exceeds the available file system space """ self._check_positive_int(st_size) if self.st_size: self.size = 0 if self.filesystem: self.filesystem.change_disk_usage(st_size, self.name, self.st_dev) self.st_size = st_size self._byte_contents = None def _check_positive_int(self, size: int) -> None: # the size should be an positive integer value if not is_int_type(size) or size < 0: self.filesystem.raise_os_error(errno.ENOSPC, self.name) def is_large_file(self) -> bool: """Return `True` if this file was initialized with size but no contents. """ return self._byte_contents is None def _encode_contents(self, contents: Union[str, bytes, None]) -> Optional[bytes]: if is_unicode_string(contents): contents = bytes( cast(str, contents), self.encoding or get_locale_encoding(), self.errors, ) return cast(bytes, contents) def set_initial_contents(self, contents: AnyStr) -> bool: """Sets the file contents and size. Called internally after initial file creation. Args: contents: string, new content of file. Returns: `True` if the contents have been changed. Raises: OSError: if the `st_size` is not a non-negative integer, or if `st_size` exceeds the available file system space """ byte_contents = self._encode_contents(contents) changed = self._byte_contents != byte_contents st_size = len(byte_contents) if byte_contents else 0 current_size = self.st_size or 0 self.filesystem.change_disk_usage( st_size - current_size, self.name, self.st_dev ) self._byte_contents = byte_contents self.st_size = st_size self.epoch += 1 return changed def set_contents(self, contents: AnyStr, encoding: Optional[str] = None) -> bool: """Sets the file contents and size and increases the modification time. Also executes the side_effects if available. Args: contents: (str, bytes) new content of file. encoding: (str) the encoding to be used for writing the contents if they are a Unicode string. If not given, the locale preferred encoding is used. Returns: `True` if the contents have been changed. Raises: OSError: if `st_size` is not a non-negative integer, or if it exceeds the available file system space. """ self.encoding = real_encoding(encoding) changed = self.set_initial_contents(contents) if self._side_effect is not None: self._side_effect(self) return changed @property def size(self) -> int: """Return the size in bytes of the file contents.""" return self.st_size @size.setter def size(self, st_size: int) -> None: """Resizes file content, padding with nulls if new size exceeds the old size. Args: st_size: The desired size for the file. Raises: OSError: if the st_size arg is not a non-negative integer or if st_size exceeds the available file system space """ self._check_positive_int(st_size) current_size = self.st_size or 0 self.filesystem.change_disk_usage( st_size - current_size, self.name, self.st_dev ) if self._byte_contents: if st_size < current_size: self._byte_contents = self._byte_contents[:st_size] else: self._byte_contents += b"\0" * (st_size - current_size) self.st_size = st_size self.epoch += 1 @property def path(self) -> AnyStr: # type: ignore[type-var] """Return the full path of the current object.""" names: List[AnyStr] = [] # pytype: disable=invalid-annotation obj: Optional[FakeFile] = self while obj: names.insert(0, matching_string(self.name, obj.name)) # type: ignore obj = obj.parent_dir sep = self.filesystem.get_path_separator(names[0]) if names[0] == sep: names.pop(0) dir_path = sep.join(names) drive = self.filesystem.splitdrive(dir_path)[0] # if a Windows path already starts with a drive or UNC path, # no extra separator is needed if not drive: dir_path = sep + dir_path else: dir_path = sep.join(names) return self.filesystem.absnormpath(dir_path) if sys.version_info >= (3, 12): @property def is_junction(self) -> bool: return self.filesystem.isjunction(self.path) def __getattr__(self, item: str) -> Any: """Forward some properties to stat_result.""" if item in self.stat_types: return getattr(self.stat_result, item) return super().__getattribute__(item) def __setattr__(self, key: str, value: Any) -> None: """Forward some properties to stat_result.""" if key in self.stat_types: return setattr(self.stat_result, key, value) return super().__setattr__(key, value) def __str__(self) -> str: return f"{self.name!r}({self.st_mode:o})" def has_permission(self, permission_bits: int) -> bool: """Checks if the given permissions are set in the fake file. Args: permission_bits: The permission bits as set for the user. Returns: `True` if the permissions are set in the correct class (user/group/other). """ if helpers.get_uid() == self.stat_result.st_uid: return self.st_mode & permission_bits == permission_bits if helpers.get_gid() == self.stat_result.st_gid: return self.st_mode & (permission_bits >> 3) == permission_bits >> 3 return self.st_mode & (permission_bits >> 6) == permission_bits >> 6 class FakeNullFile(FakeFile): def __init__(self, filesystem: "FakeFilesystem") -> None: super().__init__(filesystem.devnull, filesystem=filesystem, contents="") @property def byte_contents(self) -> bytes: return b"" def set_initial_contents(self, contents: AnyStr) -> bool: return False class FakeFileFromRealFile(FakeFile): """Represents a fake file copied from the real file system. The contents of the file are read on demand only. """ def __init__( self, file_path: str, filesystem: "FakeFilesystem", side_effect: Optional[Callable] = None, ) -> None: """ Args: file_path: Path to the existing file. filesystem: The fake filesystem where the file is created. Raises: OSError: if the file does not exist in the real file system. OSError: if the file already exists in the fake file system. """ super().__init__( name=os.path.basename(file_path), filesystem=filesystem, side_effect=side_effect, ) self.contents_read = False @property def byte_contents(self) -> Optional[bytes]: if not self.contents_read: self.contents_read = True with io_open(self.file_path, "rb") as f: self._byte_contents = f.read() # On MacOS and BSD, the above io.open() updates atime on the real file self.st_atime = os.stat(self.file_path).st_atime return self._byte_contents def set_contents(self, contents, encoding=None): self.contents_read = True super().set_contents(contents, encoding) def is_large_file(self): """The contents are never faked.""" return False class FakeDirectory(FakeFile): """Provides the appearance of a real directory.""" def __init__( self, name: str, perm_bits: int = helpers.PERM_DEF, filesystem: Optional["FakeFilesystem"] = None, ): """ Args: name: name of the file/directory, without parent path information perm_bits: permission bits. defaults to 0o777. filesystem: if set, the fake filesystem where the directory is created """ FakeFile.__init__(self, name, S_IFDIR \| perm_bits, "", filesystem=filesystem) # directories have the link count of contained entries, # including '.' and '..' self.st_nlink += 1 self._entries: Dict[str, AnyFile] = {} def set_contents(self, contents: AnyStr, encoding: Optional[str] = None) -> bool: raise self.filesystem.raise_os_error(errno.EISDIR, self.path) @property def entries(self) -> Dict[str, FakeFile]: """Return the list of contained directory entries.""" return self._entries @property def ordered_dirs(self) -> List[str]: """Return the list of contained directory entry names ordered by creation order. """ return [ item[0] for item in sorted(self._entries.items(), key=lambda entry: entry[1].st_ino) ] def add_entry(self, path_object: FakeFile) -> None: """Adds a child FakeFile to this directory. Args: path_object: FakeFile instance to add as a child of this directory. Raises: OSError: if the directory has no write permission (Posix only) OSError: if the file or directory to be added already exists """ if ( not helpers.is_root() and not self.filesystem.is_windows_fs and not self.has_permission(helpers.PERM_WRITE) ): self.filesystem.raise_os_error(errno.EACCES, self.path) path_object_name: str = to_string(path_object.name) if path_object_name in self.entries: self.filesystem.raise_os_error(errno.EEXIST, self.path) self._entries[path_object_name] = path_object path_object.parent_dir = self if path_object.st_ino is None: self.filesystem.last_ino += 1 path_object.st_ino = self.filesystem.last_ino self.st_nlink += 1 path_object.st_nlink += 1 path_object.st_dev = self.st_dev if path_object.st_nlink == 1: self.filesystem.change_disk_usage( path_object.size, path_object.name, self.st_dev ) def get_entry(self, pathname_name: str) -> AnyFile: """Retrieves the specified child file or directory entry. Args: pathname_name: The basename of the child object to retrieve. Returns: The fake file or directory object. Raises: KeyError: if no child exists by the specified name. """ pathname_name = self._normalized_entryname(pathname_name) return self.entries[to_string(pathname_name)] def _normalized_entryname(self, pathname_name: str) -> str: if not self.filesystem.is_case_sensitive: matching_names = [ name for name in self.entries if name.lower() == pathname_name.lower() ] if matching_names: pathname_name = matching_names[0] return pathname_name def remove_entry(self, pathname_name: str, recursive: bool = True) -> None: """Removes the specified child file or directory. Args: pathname_name: Basename of the child object to remove. recursive: If `True` (default), the entries in contained directories are deleted first. Used to propagate removal errors (e.g. permission problems) from contained entries. Raises: KeyError: if no child exists by the specified name. OSError: if user lacks permission to delete the file, or (Windows only) the file is open. """ pathname_name = self._normalized_entryname(pathname_name) entry = self.get_entry(pathname_name) if self.filesystem.is_windows_fs: if not is_root() and entry.st_mode & helpers.PERM_WRITE == 0: self.filesystem.raise_os_error(errno.EACCES, pathname_name) if self.filesystem.has_open_file(entry): raise_error = True if os.name == "posix" and not hasattr(os, "O_TMPFILE"): # special handling for emulating Windows under macOS and PyPi # tempfile uses unlink based on the real OS while deleting # a temporary file, so we ignore that error in this specific case st = traceback.extract_stack(limit=6) if sys.version_info < (3, 10): if ( st[0].name == "TemporaryFile" and st[0].line == "_os.unlink(name)" ): raise_error = False else: # TemporaryFile implementation has changed in Python 3.10 if st[0].name == "opener" and st[0].line == "_os.unlink(name)": raise_error = False if raise_error: self.filesystem.raise_os_error(errno.EACCES, pathname_name) else: if not helpers.is_root() and not self.has_permission( helpers.PERM_WRITE \| helpers.PERM_EXE ): self.filesystem.raise_os_error(errno.EACCES, pathname_name) if recursive and isinstance(entry, FakeDirectory): while entry.entries: entry.remove_entry(list(entry.entries)[0]) elif entry.st_nlink == 1: self.filesystem.change_disk_usage(-entry.size, pathname_name, entry.st_dev) self.st_nlink -= 1 entry.st_nlink -= 1 assert entry.st_nlink >= 0 del self.entries[to_string(pathname_name)] @property def size(self) -> int: """Return the total size of all files contained in this directory tree. """ return sum([item[1].size for item in self.entries.items()]) @size.setter def size(self, st_size: int) -> None: """Setting the size is an error for a directory.""" raise self.filesystem.raise_os_error(errno.EISDIR, self.path) def has_parent_object(self, dir_object: "FakeDirectory") -> bool: """Return `True` if dir_object is a direct or indirect parent directory, or if both are the same object.""" obj: Optional[FakeDirectory] = self while obj: if obj == dir_object: return True obj = obj.parent_dir return False def __str__(self) -> str: description = super().__str__() + ":\n" for item in self.entries: item_desc = self.entries[item].__str__() for line in item_desc.split("\n"): if line: description = description + " " + line + "\n" return description class FakeDirectoryFromRealDirectory(FakeDirectory): """Represents a fake directory copied from the real file system. The contents of the directory are read on demand only. """ def __init__( self, source_path: AnyPath, filesystem: "FakeFilesystem", read_only: bool, target_path: Optional[AnyPath] = None, ): """ Args: source_path: Full directory path. filesystem: The fake filesystem where the directory is created. read_only: If set, all files under the directory are treated as read-only, e.g. a write access raises an exception; otherwise, writing to the files changes the fake files only as usually. target_path: If given, the target path of the directory, otherwise the target is the same as `source_path`. Raises: OSError: if the directory does not exist in the real file system """ target_path = target_path or source_path real_stat = os.stat(source_path) super().__init__( name=to_string(os.path.split(target_path)[1]), perm_bits=real_stat.st_mode, filesystem=filesystem, ) self.st_ctime = real_stat.st_ctime self.st_atime = real_stat.st_atime self.st_mtime = real_stat.st_mtime self.st_gid = real_stat.st_gid self.st_uid = real_stat.st_uid self.source_path = source_path # type: ignore self.read_only = read_only self.contents_read = False @property def entries(self) -> Dict[str, FakeFile]: """Return the list of contained directory entries, loading them if not already loaded.""" if not self.contents_read: self.contents_read = True base = self.path for entry in os.listdir(self.source_path): source_path = os.path.join(self.source_path, entry) target_path = os.path.join(base, entry) # type: ignore if os.path.islink(source_path): self.filesystem.add_real_symlink(source_path, target_path) elif os.path.isdir(source_path): self.filesystem.add_real_directory( source_path, self.read_only, target_path=target_path ) else: self.filesystem.add_real_file( source_path, self.read_only, target_path=target_path ) return self._entries @property def size(self) -> int: # we cannot get the size until the contents are loaded if not self.contents_read: return 0 return super().size @size.setter def size(self, st_size: int) -> None: raise self.filesystem.raise_os_error(errno.EISDIR, self.path) class FakeFileWrapper: """Wrapper for a stream object for use by a FakeFile object. If the wrapper has any data written to it, it will propagate to the FakeFile object on close() or flush(). """ def __init__( self, file_object: FakeFile, file_path: AnyStr, open_modes: _OpenModes, allow_update: bool, delete_on_close: bool, filesystem: "FakeFilesystem", newline: Optional[str], binary: bool, closefd: bool, encoding: Optional[str], errors: Optional[str], buffering: int, raw_io: bool, opened_as_fd: bool, is_stream: bool = False, ): self.file_object = file_object self.file_path = file_path # type: ignore[var-annotated] self.open_modes = open_modes self.allow_update = allow_update self._closefd = closefd self._file_epoch = file_object.epoch self.raw_io = raw_io self._binary = binary self.opened_as_fd = opened_as_fd self.is_stream = is_stream self._changed = False self._buffer_size = buffering if self._buffer_size == 0 and not binary: raise ValueError("can't have unbuffered text I/O") # buffer_size is ignored in text mode elif self._buffer_size == -1 or not binary: self._buffer_size = io.DEFAULT_BUFFER_SIZE self._use_line_buffer = not binary and buffering == 1 contents = file_object.byte_contents self._encoding = encoding or get_locale_encoding() errors = errors or "strict" self._io: Union[BinaryBufferIO, TextBufferIO] = ( BinaryBufferIO(contents) if binary else TextBufferIO( contents, encoding=encoding, newline=newline, errors=errors ) ) self._read_whence = 0 self._read_seek = 0 self._flush_pos = 0 if contents: self._flush_pos = len(contents) if self.allow_update: if not self.open_modes.append: self._io.seek(0) else: self._io.seek(self._flush_pos) self._read_seek = self._io.tell() if delete_on_close: assert filesystem, "delete_on_close=True requires filesystem" self._filesystem = filesystem self.delete_on_close = delete_on_close # override, don't modify FakeFile.name, as FakeFilesystem expects # it to be the file name only, no directories. self.name = file_object.opened_as self.filedes: Optional[int] = None def __enter__(self) -> "FakeFileWrapper": """To support usage of this fake file with the 'with' statement.""" return self def __exit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: """To support usage of this fake file with the 'with' statement.""" self.close() def _raise(self, message: str) -> NoReturn: if self.raw_io: self._filesystem.raise_os_error(errno.EBADF, self.file_path) raise io.UnsupportedOperation(message) def get_object(self) -> FakeFile: """Return the FakeFile object that is wrapped by the current instance. """ return self.file_object def fileno(self) -> int: """Return the file descriptor of the file object.""" if self.filedes is not None: return self.filedes self._filesystem.raise_os_error(errno.EBADF) def close(self) -> None: """Close the file.""" self.close_fd(self.filedes) def close_fd(self, fd: Optional[int]) -> None: """Close the file for the given file descriptor.""" # ignore closing a closed file if not self._is_open(): return # for raw io, all writes are flushed immediately if not self.raw_io: try: self.flush() except OSError as e: if e.errno == errno.EBADF: # if we get here, we have an open file descriptor # without write permission, which has to be closed assert self.filedes self._filesystem.close_open_file(self.filedes) raise if self._filesystem.is_windows_fs and self._changed: self.file_object.st_mtime = helpers.now() assert fd is not None if self._closefd: self._filesystem.close_open_file(fd) else: open_files = self._filesystem.open_files[fd] assert open_files is not None open_files.remove(self) if self.delete_on_close: self._filesystem.remove_object( self.get_object().path # type: ignore[arg-type] ) @property def closed(self) -> bool: """Simulate the `closed` attribute on file.""" return not self._is_open() @property def mode(self) -> str: if self.open_modes.append: m = "ab" if self._binary else "a" return m + "+" if self.open_modes.can_read else m if self.open_modes.truncate: if self._binary: return "rb+" if self.open_modes.can_read else "wb" return "w+" if self.open_modes.can_read else "w" if self.open_modes.must_not_exist: m = "xb" if self._binary else "x" return m + "+" if self.open_modes.can_read else m m = "rb" if self._binary else "r" return m + "+" if self.open_modes.can_write else m def _try_flush(self, old_pos: int) -> None: """Try to flush and reset the position if it fails.""" flush_pos = self._flush_pos try: self.flush() except OSError: # write failed - reset to previous position self._io.seek(old_pos) self._io.truncate() self._flush_pos = flush_pos raise def flush(self) -> None: """Flush file contents to 'disk'.""" if self.is_stream: return self._check_open_file() if self.allow_update: if self.open_modes.append: contents = self._io.getvalue() self._sync_io() old_contents = self.file_object.byte_contents assert old_contents is not None contents = old_contents + contents[self._flush_pos :] self._set_stream_contents(contents) else: self._io.flush() contents = self._io.getvalue() changed = self.file_object.set_contents(contents, self._encoding) self.update_flush_pos() if changed: if self._filesystem.is_windows_fs: self._changed = True else: current_time = helpers.now() self.file_object.st_ctime = current_time self.file_object.st_mtime = current_time self._file_epoch = self.file_object.epoch self._flush_related_files() else: buf_length = len(self._io.getvalue()) content_length = 0 if self.file_object.byte_contents is not None: content_length = len(self.file_object.byte_contents) # an error is only raised if there is something to flush if content_length != buf_length: self._filesystem.raise_os_error(errno.EBADF) def update_flush_pos(self) -> None: self._flush_pos = self._io.tell() def _flush_related_files(self) -> None: for open_files in self._filesystem.open_files[3:]: if open_files is not None: for open_file in open_files: if ( open_file is not self and isinstance(open_file, FakeFileWrapper) and self.file_object == open_file.file_object and not open_file.open_modes.append ): open_file._sync_io() def seek(self, offset: int, whence: int = 0) -> None: """Move read/write pointer in 'file'.""" self._check_open_file() if not self.open_modes.append: self._io.seek(offset, whence) else: self._read_seek = offset self._read_whence = whence if not self.is_stream: self.flush() def tell(self) -> int: """Return the file's current position. Returns: int, file's current position in bytes. """ self._check_open_file() if not self.is_stream: self.flush() if not self.open_modes.append: return self._io.tell() if self._read_whence: write_seek = self._io.tell() self._io.seek(self._read_seek, self._read_whence) self._read_seek = self._io.tell() self._read_whence = 0 self._io.seek(write_seek) return self._read_seek def _sync_io(self) -> None: """Update the stream with changes to the file object contents.""" if self._file_epoch == self.file_object.epoch: return contents = self.file_object.byte_contents assert contents is not None self._set_stream_contents(contents) self._file_epoch = self.file_object.epoch def _set_stream_contents(self, contents: bytes) -> None: whence = self._io.tell() self._io.seek(0) self._io.truncate() self._io.putvalue(contents) if not self.open_modes.append: self._io.seek(whence) def _read_wrappers(self, name: str) -> Callable: """Wrap a stream attribute in a read wrapper. Returns a read_wrapper which tracks our own read pointer since the stream object has no concept of a different read and write pointer. Args: name: The name of the attribute to wrap. Should be a read call. Returns: The read_wrapper function. """ io_attr = getattr(self._io, name) def read_wrapper(args, kwargs): """Wrap all read calls to the stream object. We do this to track the read pointer separate from the write pointer. Anything that wants to read from the stream object while we're in append mode goes through this. Args: args: pass through args *kwargs: pass through kwargs Returns: Wrapped stream object method """ self._io.seek(self._read_seek, self._read_whence) ret_value = io_attr(args, *kwargs) self._read_seek = self._io.tell() self._read_whence = 0 self._io.seek(0, 2) return ret_value return read_wrapper def _other_wrapper(self, name: str) -> Callable: """Wrap a stream attribute in an other_wrapper. Args: name: the name of the stream attribute to wrap. Returns: other_wrapper which is described below. """ io_attr = getattr(self._io, name) def other_wrapper(args, *kwargs): """Wrap all other calls to the stream Object. We do this to track changes to the write pointer. Anything that moves the write pointer in a file open for appending should move the read pointer as well. Args: args: Pass through args. *kwargs: Pass through kwargs. Returns: Wrapped stream object method. """ write_seek = self._io.tell() ret_value = io_attr(args, *kwargs) if write_seek != self._io.tell(): self._read_seek = self._io.tell() self._read_whence = 0 return ret_value return other_wrapper def _write_wrapper(self, name: str) -> Callable: """Wrap a stream attribute in a write_wrapper. Args: name: the name of the stream attribute to wrap. Returns: write_wrapper which is described below. """ io_attr = getattr(self._io, name) def write_wrapper(args, *kwargs): """Wrap all other calls to the stream Object. We do this to track changes to the write pointer. Anything that moves the write pointer in a file open for appending should move the read pointer as well. Args: args: Pass through args. *kwargs: Pass through kwargs. Returns: Wrapped stream object method. """ old_pos = self._io.tell() ret_value = io_attr(args, *kwargs) new_pos = self._io.tell() # if the buffer size is exceeded, we flush use_line_buf = self._use_line_buffer and "\n" in args[0] if new_pos - self._flush_pos > self._buffer_size or use_line_buf: flush_all = new_pos - old_pos > self._buffer_size or use_line_buf # if the current write does not exceed the buffer size, # we revert to the previous position and flush that, # otherwise we flush all if not flush_all: self._io.seek(old_pos) self._io.truncate() self._try_flush(old_pos) if not flush_all: ret_value = io_attr(args, *kwargs) if self.open_modes.append: self._read_seek = self._io.tell() self._read_whence = 0 return ret_value return write_wrapper def _adapt_size_for_related_files(self, size: int) -> None: for open_files in self._filesystem.open_files[3:]: if open_files is not None: for open_file in open_files: if ( open_file is not self and isinstance(open_file, FakeFileWrapper) and self.file_object == open_file.file_object and cast(FakeFileWrapper, open_file).open_modes.append ): open_file._read_seek += size def _truncate_wrapper(self) -> Callable: """Wrap truncate() to allow flush after truncate. Returns: Wrapper which is described below. """ io_attr = self._io.truncate def truncate_wrapper(args, *kwargs): """Wrap truncate call to call flush after truncate.""" if self.open_modes.append: self._io.seek(self._read_seek, self._read_whence) size = io_attr(args, *kwargs) self.flush() if not self.is_stream: self.file_object.size = size buffer_size = len(self._io.getvalue()) if buffer_size < size: self._io.seek(buffer_size) self._io.putvalue(b"\0" (size - buffer_size)) self.file_object.set_contents(self._io.getvalue(), self._encoding) self._flush_pos = size self._adapt_size_for_related_files(size - buffer_size) self.flush() return size return truncate_wrapper def size(self) -> int: """Return the content size in bytes of the wrapped file.""" return self.file_object.st_size def __getattr__(self, name: str) -> Any: if self.file_object.is_large_file(): raise FakeLargeFileIoException(self.file_path) reading = name.startswith("read") or name == "next" truncate = name == "truncate" writing = name.startswith("write") or truncate if reading or writing: self._check_open_file() if not self.open_modes.can_read and reading: return self._read_error() if not self.opened_as_fd and not self.allow_update and writing: return self._write_error() if reading: self._sync_io() if not self.is_stream: self.flush() if not self._filesystem.is_windows_fs: self.file_object.st_atime = helpers.now() if truncate: return self._truncate_wrapper() if self.open_modes.append: if reading: return self._read_wrappers(name) elif not writing: return self._other_wrapper(name) if writing: return self._write_wrapper(name) return getattr(self._io, name) def _read_error(self) -> Callable: def read_error(args, kwargs): """Throw an error unless the argument is zero.""" if args and args[0] == 0: if self._filesystem.is_windows_fs and self.raw_io: return b"" if self._binary else "" self._raise("File is not open for reading.") return read_error def _write_error(self) -> Callable: def write_error(args, **kwargs): """Throw an error.""" if self.raw_io: if self._filesystem.is_windows_fs and args and len(args[0]) == 0: return 0 self._raise("File is not open for writing.") return write_error def _is_open(self) -> bool: if self.filedes is not None and self.filedes < len(self._filesystem.open_files): open_files = self._filesystem.open_files[self.filedes] if open_files is not None and self in open_files: return True return False def _check_open_file(self) -> None: if not self.is_stream and not self._is_open(): raise ValueError("I/O operation on closed file") def __iter__(self) -> Union[Iterator[str], Iterator[bytes]]: if not self.open_modes.can_read: self._raise("File is not open for reading") return self._io.__iter__() def __next__(self): if not self.open_modes.can_read: self._raise("File is not open for reading") return next(self._io) class StandardStreamWrapper: """Wrapper for a system standard stream to be used in open files list.""" def __init__(self, stream_object: TextIO): self._stream_object = stream_object self.filedes: Optional[int] = None def get_object(self) -> TextIO: return self._stream_object def fileno(self) -> int: """Return the file descriptor of the wrapped standard stream.""" if self.filedes is not None: return self.filedes raise OSError(errno.EBADF, os.strerror(errno.EBADF)) def read(self, n: int = -1) -> bytes: return cast(bytes, self._stream_object.read()) def write(self, contents: bytes) -> int: self._stream_object.write(cast(str, contents)) return len(contents) def close(self) -> None: """We do not support closing standard streams.""" def close_fd(self, fd: Optional[int]) -> None: """We do not support closing standard streams.""" def is_stream(self) -> bool: return True def __enter__(self) -> "StandardStreamWrapper": """To support usage of this standard stream with the 'with' statement.""" return self def __exit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: """To support usage of this standard stream with the 'with' statement.""" self.close() class FakeDirWrapper: """Wrapper for a FakeDirectory object to be used in open files list.""" def __init__( self, file_object: FakeDirectory, file_path: AnyString, filesystem: "FakeFilesystem", ): self.file_object = file_object self.file_path = file_path self._filesystem = filesystem self.filedes: Optional[int] = None def get_object(self) -> FakeDirectory: """Return the FakeFile object that is wrapped by the current instance.""" return self.file_object def fileno(self) -> int: """Return the file descriptor of the file object.""" if self.filedes is not None: return self.filedes raise OSError(errno.EBADF, os.strerror(errno.EBADF)) def close(self) -> None: """Close the directory.""" self.close_fd(self.filedes) def close_fd(self, fd: Optional[int]) -> None: """Close the directory.""" assert fd is not None self._filesystem.close_open_file(fd) def read(self, numBytes: int = -1) -> bytes: """Read from the directory.""" return self.file_object.read(numBytes) def write(self, contents: bytes) -> int: """Write to the directory.""" self.file_object.write(contents) return len(contents) def __enter__(self) -> "FakeDirWrapper": """To support usage of this fake directory with the 'with' statement.""" return self def __exit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: """To support usage of this fake directory with the 'with' statement.""" self.close() class FakePipeWrapper: """Wrapper for a read or write descriptor of a real pipe object to be used in open files list. """ def __init__( self, filesystem: "FakeFilesystem", fd: int, can_write: bool, mode: str = "", ): self._filesystem = filesystem self.fd = fd # the real file descriptor self.can_write = can_write self.file_object = None self.filedes: Optional[int] = None self.real_file = None if mode: self.real_file = open(fd, mode) def __enter__(self) -> "FakePipeWrapper": """To support usage of this fake pipe with the 'with' statement.""" return self def __exit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: """To support usage of this fake pipe with the 'with' statement.""" self.close() def get_object(self) -> None: return self.file_object def fileno(self) -> int: """Return the fake file descriptor of the pipe object.""" if self.filedes is not None: return self.filedes raise OSError(errno.EBADF, os.strerror(errno.EBADF)) def read(self, numBytes: int = -1) -> bytes: """Read from the real pipe.""" if self.real_file: return self.real_file.read(numBytes) # pytype: disable=bad-return-type return os.read(self.fd, numBytes) def flush(self) -> None: """Flush the real pipe?""" def write(self, contents: bytes) -> int: """Write to the real pipe.""" if self.real_file: return self.real_file.write(contents) return os.write(self.fd, contents) def close(self) -> None: """Close the pipe descriptor.""" self.close_fd(self.filedes) def close_fd(self, fd: Optional[int]) -> None: """Close the pipe descriptor with the given file descriptor.""" assert fd is not None open_files = self._filesystem.open_files[fd] assert open_files is not None open_files.remove(self) if self.real_file: self.real_file.close() else: os.close(self.fd) def readable(self) -> bool: """The pipe end can either be readable or writable.""" return not self.can_write def writable(self) -> bool: """The pipe end can either be readable or writable.""" return self.can_write def seekable(self) -> bool: """A pipe is not seekable.""" return False	49.83 KB	2026-01-20 12:58
📄 fake_filesystem.py # Copyright 2009 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """A fake filesystem implementation for unit testing. :Usage: >>> from pyfakefs import fake_filesystem, fake_os >>> filesystem = fake_filesystem.FakeFilesystem() >>> os_module = fake_os.FakeOsModule(filesystem) >>> pathname = '/a/new/dir/new-file' Create a new file object, creating parent directory objects as needed: >>> os_module.path.exists(pathname) False >>> new_file = filesystem.create_file(pathname) File objects can't be overwritten: >>> os_module.path.exists(pathname) True >>> try: ... filesystem.create_file(pathname) ... except OSError as e: ... assert e.errno == errno.EEXIST, 'unexpected errno: %d' % e.errno ... assert e.strerror == 'File exists in the fake filesystem' Remove a file object: >>> filesystem.remove_object(pathname) >>> os_module.path.exists(pathname) False Create a new file object at the previous path: >>> beatles_file = filesystem.create_file(pathname, ... contents='Dear Prudence\\nWon\\'t you come out to play?\\n') >>> os_module.path.exists(pathname) True Use the FakeFileOpen class to read fake file objects: >>> file_module = fake_filesystem.FakeFileOpen(filesystem) >>> for line in file_module(pathname): ... print(line.rstrip()) ... Dear Prudence Won't you come out to play? File objects cannot be treated like directory objects: >>> try: ... os_module.listdir(pathname) ... except OSError as e: ... assert e.errno == errno.ENOTDIR, 'unexpected errno: %d' % e.errno ... assert e.strerror == 'Not a directory in the fake filesystem' The FakeOsModule can list fake directory objects: >>> os_module.listdir(os_module.path.dirname(pathname)) ['new-file'] The FakeOsModule also supports stat operations: >>> import stat >>> stat.S_ISREG(os_module.stat(pathname).st_mode) True >>> stat.S_ISDIR(os_module.stat(os_module.path.dirname(pathname)).st_mode) True """ import contextlib import dataclasses import errno import heapq import os import random import sys import tempfile from collections import namedtuple, OrderedDict from doctest import TestResults from enum import Enum from stat import ( S_IFREG, S_IFDIR, S_ISLNK, S_IFMT, S_ISDIR, S_IFLNK, S_ISREG, ) from typing import ( List, Callable, Union, Any, Dict, Tuple, cast, AnyStr, overload, NoReturn, Optional, ) from pyfakefs import fake_file, fake_path, fake_io, fake_os, helpers, fake_open from pyfakefs.fake_file import AnyFileWrapper, AnyFile from pyfakefs.helpers import ( is_int_type, make_string_path, to_string, matching_string, AnyPath, AnyString, WINDOWS_PROPERTIES, POSIX_PROPERTIES, FSType, ) if sys.platform.startswith("linux"): # on newer Linux system, the default maximum recursion depth is 40 # we ignore older systems here _MAX_LINK_DEPTH = 40 else: # on MacOS and Windows, the maximum recursion depth is 32 _MAX_LINK_DEPTH = 32 class OSType(Enum): """Defines the real or simulated OS of the underlying file system.""" LINUX = "linux" MACOS = "macos" WINDOWS = "windows" # definitions for backwards compatibility FakeFile = fake_file.FakeFile FakeNullFile = fake_file.FakeNullFile FakeFileFromRealFile = fake_file.FakeFileFromRealFile FakeDirectory = fake_file.FakeDirectory FakeDirectoryFromRealDirectory = fake_file.FakeDirectoryFromRealDirectory FakeFileWrapper = fake_file.FakeFileWrapper StandardStreamWrapper = fake_file.StandardStreamWrapper FakeDirWrapper = fake_file.FakeDirWrapper FakePipeWrapper = fake_file.FakePipeWrapper FakePathModule = fake_path.FakePathModule FakeOsModule = fake_os.FakeOsModule FakeFileOpen = fake_open.FakeFileOpen FakeIoModule = fake_io.FakeIoModule if sys.platform != "win32": FakeFcntlModule = fake_io.FakeFcntlModule PatchMode = fake_io.PatchMode is_root = helpers.is_root get_uid = helpers.get_uid set_uid = helpers.set_uid get_gid = helpers.get_gid set_gid = helpers.set_gid reset_ids = helpers.reset_ids PERM_READ = helpers.PERM_READ PERM_WRITE = helpers.PERM_WRITE PERM_EXE = helpers.PERM_EXE PERM_DEF = helpers.PERM_DEF PERM_DEF_FILE = helpers.PERM_DEF_FILE PERM_ALL = helpers.PERM_ALL class FakeFilesystem: """Provides the appearance of a real directory tree for unit testing. Attributes: is_case_sensitive: `True` if a case-sensitive file system is assumed. root: The root :py:class:`FakeDirectory<pyfakefs.fake_file.FakeDirectory>` entry of the file system. umask: The umask used for newly created files, see `os.umask`. patcher: Holds the Patcher object if created from it. Allows access to the patcher object if using the pytest fs fixture. patch_open_code: Defines how `io.open_code <https://docs.python.org/3/library/io.html#io.open_code>`__ will be patched; patching can be on, off, or in automatic mode. shuffle_listdir_results: If `True`, `os.listdir` will not sort the results to match the real file system behavior. """ def __init__( self, path_separator: str = os.path.sep, total_size: Optional[int] = None, patcher: Any = None, create_temp_dir: bool = False, ) -> None: """ Args: path_separator: optional substitute for os.path.sep total_size: if not `None`, the total size in bytes of the root filesystem. patcher: the Patcher instance if created from the Patcher create_temp_dir: If `True`, a temp directory is created on initialization. Under Posix, if the temp directory is not `/tmp`, a link to the temp path is additionally created at `/tmp`. Example usage to use the same path separator under all systems: >>> filesystem = FakeFilesystem(path_separator='/') """ self.patcher = patcher self.create_temp_dir = create_temp_dir # is_windows_fs can be used to test the behavior of pyfakefs under # Windows fs on non-Windows systems and vice verse; # is it used to support drive letters, UNC paths and some other # Windows-specific features self._is_windows_fs = sys.platform == "win32" # can be used to test some MacOS-specific behavior under other systems self._is_macos = sys.platform == "darwin" # is_case_sensitive can be used to test pyfakefs for case-sensitive # file systems on non-case-sensitive systems and vice verse self.is_case_sensitive: bool = not (self._is_windows_fs or self._is_macos) # by default, we use the configured filesystem self.fs_type = FSType.DEFAULT base_properties = ( WINDOWS_PROPERTIES if self._is_windows_fs else POSIX_PROPERTIES ) self.fs_properties = [ dataclasses.replace(base_properties), POSIX_PROPERTIES, WINDOWS_PROPERTIES, ] self.path_separator = path_separator self.root: FakeDirectory self._cwd = "" # We can't query the current value without changing it: self.umask: int = os.umask(0o22) os.umask(self.umask) # A list of open file objects. Their position in the list is their # file descriptor number self.open_files: List[Optional[List[AnyFileWrapper]]] = [] # A heap containing all free positions in self.open_files list self._free_fd_heap: List[int] = [] # last used numbers for inodes (st_ino) and devices (st_dev) self.last_ino: int = 0 self.last_dev: int = 0 self.mount_points: Dict[AnyString, Dict] = OrderedDict() self.dev_null: Any = None self.reset(total_size=total_size, init_pathlib=False) # set from outside if needed self.patch_open_code = PatchMode.OFF self.shuffle_listdir_results = False @property def is_linux(self) -> bool: """Returns `True` in a real or faked Linux file system.""" return not self.is_windows_fs and not self.is_macos @property def is_windows_fs(self) -> bool: """Returns `True` in a real or faked Windows file system.""" return self.fs_type == FSType.WINDOWS or ( self.fs_type == FSType.DEFAULT and self._is_windows_fs ) @is_windows_fs.setter def is_windows_fs(self, value: bool) -> None: if self._is_windows_fs != value: self._is_windows_fs = value if value: self._is_macos = False self.reset() FakePathModule.reset(self) @property def is_macos(self) -> bool: """Returns `True` in a real or faked macOS file system.""" return self._is_macos @is_macos.setter def is_macos(self, value: bool) -> None: if self._is_macos != value: self._is_macos = value if value: self._is_windows_fs = False self.reset() FakePathModule.reset(self) @property def path_separator(self) -> str: """Returns the path separator, corresponds to `os.path.sep`.""" return self.fs_properties[self.fs_type.value].sep @path_separator.setter def path_separator(self, value: str) -> None: self.fs_properties[0].sep = value if value != os.sep: self.alternative_path_separator = None @property def alternative_path_separator(self) -> Optional[str]: """Returns the alternative path separator, corresponds to `os.path.altsep`.""" return self.fs_properties[self.fs_type.value].altsep @alternative_path_separator.setter def alternative_path_separator(self, value: Optional[str]) -> None: self.fs_properties[0].altsep = value @property def devnull(self) -> str: return self.fs_properties[self.fs_type.value].devnull @property def pathsep(self) -> str: return self.fs_properties[self.fs_type.value].pathsep @property def line_separator(self) -> str: return self.fs_properties[self.fs_type.value].linesep @property def cwd(self) -> str: """Return the current working directory of the fake filesystem.""" return self._cwd @cwd.setter def cwd(self, value: str) -> None: """Set the current working directory of the fake filesystem. Make sure a new drive or share is auto-mounted under Windows. """ _cwd = make_string_path(value) self._cwd = _cwd.replace( matching_string(_cwd, os.sep), matching_string(_cwd, self.path_separator) ) self._auto_mount_drive_if_needed(value) @property def root_dir(self) -> FakeDirectory: """Return the root directory, which represents "/" under POSIX, and the current drive under Windows.""" if self.is_windows_fs: return self._mount_point_dir_for_cwd() return self.root @property def root_dir_name(self) -> str: """Return the root directory name, which is "/" under POSIX, and the root path of the current drive under Windows.""" root_dir = to_string(self.root_dir.name) if not root_dir.endswith(self.path_separator): return root_dir + self.path_separator return root_dir @property def os(self) -> OSType: """Return the real or simulated type of operating system.""" return ( OSType.WINDOWS if self.is_windows_fs else OSType.MACOS if self.is_macos else OSType.LINUX ) @os.setter def os(self, value: OSType) -> None: """Set the simulated type of operating system underlying the fake file system.""" self._is_windows_fs = value == OSType.WINDOWS self._is_macos = value == OSType.MACOS self.is_case_sensitive = value == OSType.LINUX self.fs_type = FSType.DEFAULT base_properties = ( WINDOWS_PROPERTIES if self._is_windows_fs else POSIX_PROPERTIES ) self.fs_properties[0] = base_properties self.reset() FakePathModule.reset(self) def reset(self, total_size: Optional[int] = None, init_pathlib: bool = True): """Remove all file system contents and reset the root.""" self.root = FakeDirectory(self.path_separator, filesystem=self) self.dev_null = FakeNullFile(self) self.open_files.clear() self._free_fd_heap.clear() self.last_ino = 0 self.last_dev = 0 self.mount_points.clear() self._add_root_mount_point(total_size) self._add_standard_streams() if self.create_temp_dir: self._create_temp_dir() if init_pathlib: from pyfakefs import fake_pathlib fake_pathlib.init_module(self) @contextlib.contextmanager def use_fs_type(self, fs_type: FSType): old_fs_type = self.fs_type try: self.fs_type = fs_type yield finally: self.fs_type = old_fs_type def _add_root_mount_point(self, total_size): mount_point = "C:" if self.is_windows_fs else self.path_separator self._cwd = mount_point if not self.cwd.endswith(self.path_separator): self._cwd += self.path_separator self.add_mount_point(mount_point, total_size) def pause(self) -> None: """Pause the patching of the file system modules until :py:meth:`resume` is called. After that call, all file system calls are executed in the real file system. Calling `pause()` twice is silently ignored. Only allowed if the file system object was created by a `Patcher` object. This is also the case for the pytest `fs` fixture. Raises: RuntimeError: if the file system was not created by a `Patcher`. """ if self.patcher is None: raise RuntimeError( "pause() can only be called from a fake file " "system object created by a Patcher object" ) self.patcher.pause() def resume(self) -> None: """Resume the patching of the file system modules if :py:meth:`pause` has been called before. After that call, all file system calls are executed in the fake file system. Does nothing if patching is not paused. Raises: RuntimeError: if the file system has not been created by `Patcher`. """ if self.patcher is None: raise RuntimeError( "resume() can only be called from a fake file " "system object created by a Patcher object" ) self.patcher.resume() def clear_cache(self) -> None: """Clear the cache of non-patched modules.""" if self.patcher: self.patcher.clear_cache() def raise_os_error( self, err_no: int, filename: Optional[AnyString] = None, winerror: Optional[int] = None, ) -> NoReturn: """Raises OSError. The error message is constructed from the given error code and shall start with the error string issued in the real system. Note: this is not true under Windows if winerror is given - in this case a localized message specific to winerror will be shown in the real file system. Args: err_no: A numeric error code from the C variable errno. filename: The name of the affected file, if any. winerror: Windows only - the specific Windows error code. """ message = os.strerror(err_no) if winerror is not None and sys.platform == "win32" and self.is_windows_fs: raise OSError(err_no, message, filename, winerror) raise OSError(err_no, message, filename) def get_path_separator(self, path: AnyStr) -> AnyStr: """Return the path separator as the same type as path""" return matching_string(path, self.path_separator) def _alternative_path_separator(self, path: AnyStr) -> Optional[AnyStr]: """Return the alternative path separator as the same type as path""" return matching_string(path, self.alternative_path_separator) def starts_with_sep(self, path: AnyStr) -> bool: """Return `True` if path starts with a path separator.""" sep = self.get_path_separator(path) altsep = self._alternative_path_separator(path) return path.startswith(sep) or altsep is not None and path.startswith(altsep) def add_mount_point( self, path: AnyStr, total_size: Optional[int] = None, can_exist: bool = False, ) -> Dict: """Add a new mount point for a filesystem device. The mount point gets a new unique device number. Args: path: The root path for the new mount path. total_size: The new total size of the added filesystem device in bytes. Defaults to infinite size. can_exist: If `True`, no error is raised if the mount point already exists. Returns: The newly created mount point dict. Raises: OSError: if trying to mount an existing mount point again, and ``can_exist`` is `False`. """ path = self.normpath(self.normcase(path)) for mount_point in self.mount_points: if ( self.is_case_sensitive and path == matching_string(path, mount_point) or not self.is_case_sensitive and path.lower() == matching_string(path, mount_point.lower()) ): if can_exist: return self.mount_points[mount_point] self.raise_os_error(errno.EEXIST, path) self.last_dev += 1 self.mount_points[path] = { "idev": self.last_dev, "total_size": total_size, "used_size": 0, } if path == matching_string(path, self.root.name): # special handling for root path: has been created before root_dir = self.root self.last_ino += 1 root_dir.st_ino = self.last_ino else: root_dir = self._create_mount_point_dir(path) root_dir.st_dev = self.last_dev return self.mount_points[path] def _create_mount_point_dir(self, directory_path: AnyPath) -> FakeDirectory: """A version of `create_dir` for the mount point directory creation, which avoids circular calls and unneeded checks. """ dir_path = self.make_string_path(directory_path) path_components = self._path_components(dir_path) current_dir = self.root new_dirs = [] for component in [to_string(p) for p in path_components]: directory = self._directory_content(current_dir, to_string(component))[1] if not directory: new_dir = FakeDirectory(component, filesystem=self) new_dirs.append(new_dir) current_dir.add_entry(new_dir) current_dir = new_dir else: current_dir = cast(FakeDirectory, directory) for new_dir in new_dirs: new_dir.st_mode = S_IFDIR \| helpers.PERM_DEF return current_dir def _auto_mount_drive_if_needed(self, path: AnyStr) -> Optional[Dict]: """Windows only: if `path` is located on an unmounted drive or UNC mount point, the drive/mount point is added to the mount points.""" if self.is_windows_fs: drive = self.splitdrive(path)[0] if drive: return self.add_mount_point(path=drive, can_exist=True) return None def _mount_point_for_path(self, path: AnyStr) -> Dict: path = self.absnormpath(self._original_path(path)) for mount_path in self.mount_points: if path == matching_string(path, mount_path): return self.mount_points[mount_path] mount_path = matching_string(path, "") drive = self.splitdrive(path)[0] for root_path in self.mount_points: root_path = matching_string(path, root_path) if drive and not root_path.startswith(drive): continue if path.startswith(root_path) and len(root_path) > len(mount_path): mount_path = root_path if mount_path: return self.mount_points[to_string(mount_path)] mount_point = self._auto_mount_drive_if_needed(path) assert mount_point return mount_point def _mount_point_dir_for_cwd(self) -> FakeDirectory: """Return the fake directory object of the mount point where the current working directory points to.""" def object_from_path(file_path) -> FakeDirectory: path_components = self._path_components(file_path) target = self.root for component in path_components: target = cast(FakeDirectory, target.get_entry(component)) return target path = to_string(self.cwd) for mount_path in self.mount_points: if path == to_string(mount_path): return object_from_path(mount_path) mount_path = "" drive = to_string(self.splitdrive(path)[0]) for root_path in self.mount_points: str_root_path = to_string(root_path) if drive and not str_root_path.startswith(drive): continue if path.startswith(str_root_path) and len(str_root_path) > len(mount_path): mount_path = root_path return object_from_path(mount_path) def _mount_point_for_device(self, idev: int) -> Optional[Dict]: for mount_point in self.mount_points.values(): if mount_point["idev"] == idev: return mount_point return None def get_disk_usage(self, path: Optional[AnyStr] = None) -> Tuple[int, int, int]: """Return the total, used and free disk space in bytes as named tuple, or placeholder values simulating unlimited space if not set. .. note:: This matches the return value of ``shutil.disk_usage()``. Args: path: The disk space is returned for the file system device where `path` resides. Defaults to the root path (e.g. '/' on Unix systems). """ DiskUsage = namedtuple("DiskUsage", "total, used, free") if path is None: mount_point = next(iter(self.mount_points.values())) else: file_path = make_string_path(path) mount_point = self._mount_point_for_path(file_path) if mount_point and mount_point["total_size"] is not None: return DiskUsage( mount_point["total_size"], mount_point["used_size"], mount_point["total_size"] - mount_point["used_size"], ) return DiskUsage(1024 * 1024 * 1024 * 1024, 0, 1024 * 1024 * 1024 * 1024) def set_disk_usage(self, total_size: int, path: Optional[AnyStr] = None) -> None: """Changes the total size of the file system, preserving the used space. Example usage: set the size of an auto-mounted Windows drive. Args: total_size: The new total size of the filesystem in bytes. path: The disk space is changed for the file system device where `path` resides. Defaults to the root path (e.g. '/' on Unix systems). Raises: OSError: if the new space is smaller than the used size. """ file_path: AnyStr = ( path if path is not None else self.root_dir_name # type: ignore ) mount_point = self._mount_point_for_path(file_path) if ( mount_point["total_size"] is not None and mount_point["used_size"] > total_size ): self.raise_os_error(errno.ENOSPC, path) mount_point["total_size"] = total_size def change_disk_usage( self, usage_change: int, file_path: AnyStr, st_dev: int ) -> None: """Change the used disk space by the given amount. Args: usage_change: Number of bytes added to the used space. If negative, the used space will be decreased. file_path: The path of the object needing the disk space. st_dev: The device ID for the respective file system. Raises: OSError: if `usage_change` exceeds the free file system space """ mount_point = self._mount_point_for_device(st_dev) if mount_point: total_size = mount_point["total_size"] if total_size is not None: if total_size - mount_point["used_size"] < usage_change: self.raise_os_error(errno.ENOSPC, file_path) mount_point["used_size"] += usage_change def stat(self, entry_path: AnyStr, follow_symlinks: bool = True): """Return the os.stat-like tuple for the FakeFile object of ``entry_path``. Args: entry_path: Path to filesystem object to retrieve. follow_symlinks: If `False` and ``entry_path`` points to a symlink, the link itself is inspected instead of the linked object. Returns: The `FakeStatResult` object corresponding to ``entry_path``. Raises: OSError: if the filesystem object doesn't exist. """ # stat should return the tuple representing return value of os.stat try: file_object = self.resolve( entry_path, follow_symlinks, allow_fd=True, check_read_perm=False, check_exe_perm=False, ) except TypeError: file_object = self.resolve(entry_path) if not is_root(): # make sure stat raises if a parent dir is not readable parent_dir = file_object.parent_dir if parent_dir: self._get_object(parent_dir.path, check_read_perm=False) # type: ignore[arg-type] self.raise_for_filepath_ending_with_separator( entry_path, file_object, follow_symlinks ) return file_object.stat_result.copy() def raise_for_filepath_ending_with_separator( self, entry_path: AnyStr, file_object: FakeFile, follow_symlinks: bool = True, macos_handling: bool = False, ) -> None: if self.ends_with_path_separator(entry_path): if S_ISLNK(file_object.st_mode): try: link_object = self.resolve(entry_path) except OSError as exc: if self.is_macos and exc.errno != errno.ENOENT: return if self.is_windows_fs: self.raise_os_error(errno.ENOTDIR, entry_path) raise if not follow_symlinks or self.is_windows_fs or self.is_macos: file_object = link_object if self.is_windows_fs: is_error = S_ISREG(file_object.st_mode) elif self.is_macos and macos_handling: is_error = not S_ISLNK(file_object.st_mode) else: is_error = not S_ISDIR(file_object.st_mode) if is_error: self.raise_os_error(errno.ENOTDIR, entry_path) def chmod( self, path: Union[AnyStr, int], mode: int, follow_symlinks: bool = True, force_unix_mode: bool = False, ) -> None: """Change the permissions of a file as encoded in integer mode. Args: path: (str \| int) Path to the file or file descriptor. mode: (int) Permissions. follow_symlinks: If `False` and `path` points to a symlink, the link itself is affected instead of the linked object. force_unix_mode: if `True` and run under Windows, the mode is not adapted for Windows to allow making dirs unreadable """ allow_fd = not self.is_windows_fs or sys.version_info >= (3, 13) file_object = self.resolve( path, follow_symlinks, allow_fd=allow_fd, check_owner=True ) if self.is_windows_fs and not force_unix_mode: if mode & helpers.PERM_WRITE: file_object.st_mode = file_object.st_mode \| 0o222 else: file_object.st_mode = file_object.st_mode & 0o777555 else: file_object.st_mode = (file_object.st_mode & ~helpers.PERM_ALL) \| ( mode & helpers.PERM_ALL ) file_object.st_ctime = helpers.now() def utime( self, path: AnyStr, times: Optional[Tuple[Union[int, float], Union[int, float]]] = None, , ns: Optional[Tuple[int, int]] = None, follow_symlinks: bool = True, ) -> None: """Change the access and modified times of a file. Args: path: (str) Path to the file. times: 2-tuple of int or float numbers, of the form (atime, mtime) which is used to set the access and modified times in seconds. If `None`, both times are set to the current time. ns: 2-tuple of int numbers, of the form (atime, mtime) which is used to set the access and modified times in nanoseconds. If `None`, both times are set to the current time. follow_symlinks: If `False` and ``path`` points to a symlink, the link itself is queried instead of the linked object. Raises: TypeError: If anything other than the expected types is specified in the passed ``times`` or ``ns`` tuple, or if the tuple length is not equal to 2. ValueError: If both ``times`` and ``ns`` are specified. """ self._handle_utime_arg_errors(ns, times) file_object = self.resolve(path, follow_symlinks, allow_fd=True) if times is not None: for file_time in times: if not isinstance(file_time, (int, float)): raise TypeError("atime and mtime must be numbers") file_object.st_atime = times[0] file_object.st_mtime = times[1] elif ns is not None: for file_time in ns: if not isinstance(file_time, int): raise TypeError("atime and mtime must be ints") file_object.st_atime_ns = ns[0] file_object.st_mtime_ns = ns[1] else: current_time = helpers.now() file_object.st_atime = current_time file_object.st_mtime = current_time @staticmethod def _handle_utime_arg_errors( ns: Optional[Tuple[int, int]], times: Optional[Tuple[Union[int, float], Union[int, float]]], ): if times is not None and ns is not None: raise ValueError( "utime: you may specify either 'times' or 'ns' but not both" ) if times is not None and len(times) != 2: raise TypeError("utime: 'times' must be either a tuple of two ints or None") if ns is not None and len(ns) != 2: raise TypeError("utime: 'ns' must be a tuple of two ints") def add_open_file(self, file_obj: AnyFileWrapper, new_fd: int = -1) -> int: """Add file_obj to the list of open files on the filesystem. Used internally to manage open files. The position in the open_files array is the file descriptor number. Args: file_obj: File object to be added to open files list. new_fd: The optional new file descriptor. Returns: File descriptor number for the file object. """ if new_fd >= 0: size = len(self.open_files) if new_fd < size: open_files = self.open_files[new_fd] if open_files: for f in open_files: try: f.close() except OSError: pass if new_fd in self._free_fd_heap: self._free_fd_heap.remove(new_fd) self.open_files[new_fd] = [file_obj] else: for fd in range(size, new_fd): self.open_files.append([]) heapq.heappush(self._free_fd_heap, fd) self.open_files.append([file_obj]) return new_fd if self._free_fd_heap: open_fd = heapq.heappop(self._free_fd_heap) self.open_files[open_fd] = [file_obj] return open_fd self.open_files.append([file_obj]) return len(self.open_files) - 1 def close_open_file(self, file_des: int) -> None: """Remove file object with given descriptor from the list of open files. Sets the entry in `open_files` to `None`. Args: file_des: Descriptor of file object to be removed from open files list. """ self.open_files[file_des] = None heapq.heappush(self._free_fd_heap, file_des) def get_open_file(self, file_des: int) -> AnyFileWrapper: """Return an open file. Args: file_des: File descriptor of the open file. Raises: OSError: an invalid file descriptor. TypeError: filedes is not an integer. Returns: Open file object. """ try: return self.get_open_files(file_des)[0] except IndexError: self.raise_os_error(errno.EBADF, str(file_des)) def get_open_files(self, file_des: int) -> List[AnyFileWrapper]: """Return the list of open files for a file descriptor. Args: file_des: File descriptor of the open files. Raises: OSError: an invalid file descriptor. TypeError: filedes is not an integer. Returns: List of open file objects. """ if not is_int_type(file_des): raise TypeError("an integer is required") valid = file_des < len(self.open_files) if valid: return self.open_files[file_des] or [] self.raise_os_error(errno.EBADF, str(file_des)) def has_open_file(self, file_object: FakeFile) -> bool: """Return `True` if the given file object is in the list of open files. Args: file_object: The FakeFile object to be checked. Returns: `True` if the file is open. """ return file_object in [ wrappers[0].get_object() for wrappers in self.open_files if wrappers ] def _normalize_path_sep(self, path: AnyStr) -> AnyStr: alt_sep = self._alternative_path_separator(path) if alt_sep is not None: return path.replace(alt_sep, self.get_path_separator(path)) return path def normcase(self, path: AnyStr) -> AnyStr: """Replace all appearances of alternative path separator with path separator. Do nothing if no alternative separator is set. Args: path: The path to be normalized. Returns: The normalized path that will be used internally. """ file_path = make_string_path(path) return self._normalize_path_sep(file_path) def normpath(self, path: AnyStr) -> AnyStr: """Mimic os.path.normpath using the specified path_separator. Mimics os.path.normpath using the path_separator that was specified for this FakeFilesystem. Normalizes the path, but unlike the method absnormpath, does not make it absolute. Eliminates dot components (. and ..) and combines repeated path separators (//). Initial .. components are left in place for relative paths. If the result is an empty path, '.' is returned instead. This also replaces alternative path separator with path separator. That is, it behaves like the real os.path.normpath on Windows if initialized with '\\' as path separator and '/' as alternative separator. Args: path: (str) The path to normalize. Returns: (str) A copy of path with empty components and dot components removed. """ path_str = self.normcase(path) drive, path_str = self.splitdrive(path_str) sep = self.get_path_separator(path_str) is_absolute_path = path_str.startswith(sep) path_components: List[AnyStr] = path_str.split( sep ) # pytype: disable=invalid-annotation collapsed_path_components: List[ AnyStr ] = [] # pytype: disable=invalid-annotation dot = matching_string(path_str, ".") dotdot = matching_string(path_str, "..") for component in path_components: if (not component) or (component == dot): continue if component == dotdot: if collapsed_path_components and ( collapsed_path_components[-1] != dotdot ): # Remove an up-reference: directory/.. collapsed_path_components.pop() continue elif is_absolute_path: # Ignore leading .. components if starting from the # root directory. continue collapsed_path_components.append(component) collapsed_path = sep.join(collapsed_path_components) if is_absolute_path: collapsed_path = sep + collapsed_path return drive + collapsed_path or dot def _original_path(self, path: AnyStr) -> AnyStr: """Return a normalized case version of the given path for case-insensitive file systems. For case-sensitive file systems, return path unchanged. Args: path: the file path to be transformed Returns: A version of path matching the case of existing path elements. """ def components_to_path(): if len(path_components) > len(normalized_components): normalized_components.extend( to_string(p) for p in path_components[len(normalized_components) :] ) sep = self.path_separator normalized_path = sep.join(normalized_components) if self.starts_with_sep(path) and not self.starts_with_sep(normalized_path): normalized_path = sep + normalized_path if len(normalized_path) == 2 and self.starts_with_drive_letter( normalized_path ): normalized_path += sep return normalized_path if self.is_case_sensitive or not path: return path path = self.replace_windows_root(path) path_components = self._path_components(path) normalized_components = [] current_dir = self.root for component in path_components: if not isinstance(current_dir, FakeDirectory): return components_to_path() dir_name, directory = self._directory_content( current_dir, to_string(component) ) if directory is None or ( isinstance(directory, FakeDirectory) and directory._byte_contents is None and directory.st_size == 0 ): return components_to_path() current_dir = cast(FakeDirectory, directory) normalized_components.append(dir_name) return components_to_path() def absnormpath(self, path: AnyStr) -> AnyStr: """Absolutize and minimalize the given path. Forces all relative paths to be absolute, and normalizes the path to eliminate dot and empty components. Args: path: Path to normalize. Returns: The normalized path relative to the current working directory, or the root directory if path is empty. """ path = self.normcase(path) cwd = matching_string(path, self.cwd) if not path: path = self.get_path_separator(path) if path == matching_string(path, "."): path = cwd elif not self._starts_with_root_path(path): # Prefix relative paths with cwd, if cwd is not root. root_name = matching_string(path, self.root.name) empty = matching_string(path, "") path = self.get_path_separator(path).join( (cwd != root_name and cwd or empty, path) ) else: path = self.replace_windows_root(path) return self.normpath(path) def splitpath(self, path: AnyStr) -> Tuple[AnyStr, AnyStr]: """Mimic os.path.split using the specified path_separator. Mimics os.path.split using the path_separator that was specified for this FakeFilesystem. Args: path: (str) The path to split. Returns: (str) A duple (pathname, basename) for which pathname does not end with a slash, and basename does not contain a slash. """ path = make_string_path(path) sep = self.get_path_separator(path) alt_sep = self._alternative_path_separator(path) seps = sep if alt_sep is None else sep + alt_sep drive, path = self.splitdrive(path) i = len(path) while i and path[i - 1] not in seps: i -= 1 head, tail = path[:i], path[i:] # now tail has no slashes # remove trailing slashes from head, unless it's all slashes head = head.rstrip(seps) or head return drive + head, tail def splitdrive(self, path: AnyStr) -> Tuple[AnyStr, AnyStr]: """Splits the path into the drive part and the rest of the path. Taken from Windows specific implementation in Python 3.5 and slightly adapted. Args: path: the full path to be splitpath. Returns: A tuple of the drive part and the rest of the path, or of an empty string and the full path if drive letters are not supported or no drive is present. """ path_str = make_string_path(path) if self.is_windows_fs: if len(path_str) >= 2: norm_str = self.normcase(path_str) sep = self.get_path_separator(path_str) # UNC path_str handling if (norm_str[0:2] == sep 2) and (norm_str[2:3] != sep): # UNC path_str handling - splits off the mount point # instead of the drive sep_index = norm_str.find(sep, 2) if sep_index == -1: return path_str[:0], path_str sep_index2 = norm_str.find(sep, sep_index + 1) if sep_index2 == sep_index + 1: return path_str[:0], path_str if sep_index2 == -1: sep_index2 = len(path_str) return path_str[:sep_index2], path_str[sep_index2:] if path_str[1:2] == matching_string(path_str, ":"): return path_str[:2], path_str[2:] return path_str[:0], path_str def splitroot(self, path: AnyStr): """Split a pathname into drive, root and tail. Implementation taken from ntpath and posixpath. """ p = os.fspath(path) if isinstance(p, bytes): sep = self.path_separator.encode() altsep = None alternative_path_separator = self.alternative_path_separator if alternative_path_separator is not None: altsep = alternative_path_separator.encode() colon = b":" unc_prefix = b"\\\\?\\UNC\\" empty = b"" else: sep = self.path_separator altsep = self.alternative_path_separator colon = ":" unc_prefix = "\\\\?\\UNC\\" empty = "" if self.is_windows_fs: normp = p.replace(altsep, sep) if altsep else p if normp[:1] == sep: if normp[1:2] == sep: # UNC drives, e.g. \\server\share or \\?\UNC\server\share # Device drives, e.g. \\.\device or \\?\device start = 8 if normp[:8].upper() == unc_prefix else 2 index = normp.find(sep, start) if index == -1: return p, empty, empty index2 = normp.find(sep, index + 1) if index2 == -1: return p, empty, empty return p[:index2], p[index2 : index2 + 1], p[index2 + 1 :] else: # Relative path with root, e.g. \Windows return empty, p[:1], p[1:] elif normp[1:2] == colon: if normp[2:3] == sep: # Absolute drive-letter path, e.g. X:\Windows return p[:2], p[2:3], p[3:] else: # Relative path with drive, e.g. X:Windows return p[:2], empty, p[2:] else: # Relative path, e.g. Windows return empty, empty, p else: if p[:1] != sep: # Relative path, e.g.: 'foo' return empty, empty, p elif p[1:2] != sep or p[2:3] == sep: # Absolute path, e.g.: '/foo', '///foo', '////foo', etc. return empty, sep, p[1:] else: return empty, p[:2], p[2:] def _join_paths_with_drive_support(self, all_paths: AnyStr) -> AnyStr: """Taken from Python 3.5 os.path.join() code in ntpath.py and slightly adapted""" base_path = all_paths[0] paths_to_add = all_paths[1:] sep = self.get_path_separator(base_path) seps = [sep, self._alternative_path_separator(base_path)] result_drive, result_path = self.splitdrive(base_path) for path in paths_to_add: drive_part, path_part = self.splitdrive(path) if path_part and path_part[:1] in seps: # Second path is absolute if drive_part or not result_drive: result_drive = drive_part result_path = path_part continue elif drive_part and drive_part != result_drive: if self.is_case_sensitive or drive_part.lower() != result_drive.lower(): # Different drives => ignore the first path entirely result_drive = drive_part result_path = path_part continue # Same drive in different case result_drive = drive_part # Second path is relative to the first if result_path and result_path[-1:] not in seps: result_path = result_path + sep result_path = result_path + path_part # add separator between UNC and non-absolute path colon = matching_string(base_path, ":") if ( result_path and result_path[:1] not in seps and result_drive and result_drive[-1:] != colon ): return result_drive + sep + result_path return result_drive + result_path def joinpaths(self, paths: AnyStr) -> AnyStr: """Mimic os.path.join using the specified path_separator. Args: paths: (str) Zero or more paths to join. Returns: (str) The paths joined by the path separator, starting with the last absolute path in paths. """ file_paths = [os.fspath(path) for path in paths] if len(file_paths) == 1: return paths[0] if self.is_windows_fs: return self._join_paths_with_drive_support(file_paths) path = file_paths[0] sep = self.get_path_separator(file_paths[0]) for path_segment in file_paths[1:]: if path_segment.startswith(sep) or not path: # An absolute path path = path_segment elif path.endswith(sep): path += path_segment else: path += sep + path_segment return path @overload def _path_components(self, path: str) -> List[str]: ... @overload def _path_components(self, path: bytes) -> List[bytes]: ... def _path_components(self, path: AnyStr) -> List[AnyStr]: """Breaks the path into a list of component names. Does not include the root directory as a component, as all paths are considered relative to the root directory for the FakeFilesystem. Callers should basically follow this pattern: .. code:: python file_path = self.absnormpath(file_path) path_components = self._path_components(file_path) current_dir = self.root for component in path_components: if component not in current_dir.entries: raise OSError _do_stuff_with_component(current_dir, component) current_dir = current_dir.get_entry(component) Args: path: Path to tokenize. Returns: The list of names split from path. """ if not path or path == self.get_path_separator(path): return [] drive, path = self.splitdrive(path) sep = self.get_path_separator(path) # handle special case of Windows emulated under POSIX if self.is_windows_fs and sys.platform != "win32": path = path.replace(matching_string(sep, "\\"), sep) path_components = path.split(sep) assert drive or path_components if not path_components[0]: if len(path_components) > 1 and not path_components[1]: path_components = [] else: # This is an absolute path. path_components = path_components[1:] if drive: path_components.insert(0, drive) return path_components def starts_with_drive_letter(self, file_path: AnyStr) -> bool: """Return `True` if file_path starts with a drive letter. Args: file_path: the full path to be examined. Returns: `True` if drive letter support is enabled in the filesystem and the path starts with a drive letter. """ colon = matching_string(file_path, ":") if len(file_path) >= 2 and file_path[0:1].isalpha() and file_path[1:2] == colon: if self.is_windows_fs: return True if os.name == "nt": # special case if we are emulating Posix under Windows # check if the path exists because it has been mapped in # this is not foolproof, but handles most cases try: if len(file_path) == 2: # avoid recursion, check directly in the entries return any( [ entry.upper() == file_path.upper() for entry in self.root_dir.entries ] ) self.get_object_from_normpath(file_path) return True except OSError: return False return False def _starts_with_root_path(self, file_path: AnyStr) -> bool: root_name = matching_string(file_path, self.root.name) file_path = self._normalize_path_sep(file_path) return ( file_path.startswith(root_name) or not self.is_case_sensitive and file_path.lower().startswith(root_name.lower()) or self.starts_with_drive_letter(file_path) ) def replace_windows_root(self, path: AnyStr) -> AnyStr: """In windows, if a path starts with a single separator, it points to the root dir of the current mount point, usually a drive - replace it with that mount point path to get the real path. """ if path and self.is_windows_fs and self.root_dir: sep = self.get_path_separator(path) # ignore UNC paths if path[0:1] == sep and (len(path) == 1 or path[1:2] != sep): # check if we already have a mount point for that path for root_path in self.mount_points: root_path = matching_string(path, root_path) if path.startswith(root_path): return path # must be a pointer to the current drive - replace it mount_point = matching_string(path, self.root_dir_name) path = mount_point + path[1:] return path def _is_root_path(self, file_path: AnyStr) -> bool: root_name = matching_string(file_path, self.root.name) return file_path == root_name or self.is_mount_point(file_path) def is_mount_point(self, file_path: AnyStr) -> bool: """Return `True` if `file_path` points to a mount point.""" for mount_point in self.mount_points: mount_point = matching_string(file_path, mount_point) if ( file_path == mount_point or not self.is_case_sensitive and file_path.lower() == mount_point.lower() ): return True if ( self.is_windows_fs and len(file_path) == 3 and len(mount_point) == 2 and self.starts_with_drive_letter(file_path) and file_path[:2].lower() == mount_point.lower() ): return True return False def ends_with_path_separator(self, path: Union[int, AnyPath]) -> bool: """Return `True` if ``file_path`` ends with a valid path separator.""" if isinstance(path, int): return False file_path = make_string_path(path) if not file_path: return False sep = self.get_path_separator(file_path) altsep = self._alternative_path_separator(file_path) return file_path not in (sep, altsep) and ( file_path.endswith(sep) or altsep is not None and file_path.endswith(altsep) ) def is_filepath_ending_with_separator(self, path: AnyStr) -> bool: if not self.ends_with_path_separator(path): return False return self.isfile(self._path_without_trailing_separators(path)) def _directory_content( self, directory: FakeDirectory, component: str ) -> Tuple[Optional[str], Optional[AnyFile]]: if not isinstance(directory, FakeDirectory): return None, None if component in directory.entries: return component, directory.entries[component] if not self.is_case_sensitive: matching_content = [ (subdir, directory.entries[subdir]) for subdir in directory.entries if subdir.lower() == component.lower() ] if matching_content: return matching_content[0] return None, None def exists(self, file_path: AnyPath, check_link: bool = False) -> bool: """Return true if a path points to an existing file system object. Args: file_path: The path to examine. check_link: If `True`, links are not followed Returns: (bool) `True` if the corresponding object exists. Raises: TypeError: if file_path is `None`. """ if check_link and self.islink(file_path): return True path = to_string(self.make_string_path(file_path)) if path is None: raise TypeError if not path: return False if path == self.devnull: return not self.is_windows_fs or sys.version_info >= (3, 8) try: if self.is_filepath_ending_with_separator(path): return False path = self.resolve_path(path) except OSError: return False if self._is_root_path(path): return True path_components: List[str] = self._path_components(path) current_dir = self.root for component in path_components: directory = self._directory_content(current_dir, to_string(component))[1] if directory is None: return False current_dir = cast(FakeDirectory, directory) return True def resolve_path(self, file_path: AnyStr, allow_fd: bool = False) -> AnyStr: """Follow a path, resolving symlinks. ResolvePath traverses the filesystem along the specified file path, resolving file names and symbolic links until all elements of the path are exhausted, or we reach a file which does not exist. If all the elements are not consumed, they just get appended to the path resolved so far. This gives us the path which is as resolved as it can be, even if the file does not exist. This behavior mimics Unix semantics, and is best shown by example. Given a file system that looks like this: /a/b/ /a/b/c -> /a/b2 c is a symlink to /a/b2 /a/b2/x /a/c -> ../d /a/x -> y Then: /a/b/x => /a/b/x /a/c => /a/d /a/x => /a/y /a/b/c/d/e => /a/b2/d/e Args: file_path: The path to examine. allow_fd: If `True`, `file_path` may be open file descriptor. Returns: The resolved_path (str or byte). Raises: TypeError: if `file_path` is `None`. OSError: if `file_path` is '' or a part of the path doesn't exist. """ if allow_fd and isinstance(file_path, int): return self.get_open_file(file_path).get_object().path path = make_string_path(file_path) if path is None: # file.open(None) raises TypeError, so mimic that. raise TypeError("Expected file system path string, received None") if sys.platform == "win32" and self.os != OSType.WINDOWS: path = path.replace( matching_string(path, os.sep), matching_string(path, self.path_separator), ) if not path or not self._valid_relative_path(path): # file.open('') raises OSError, so mimic that, and validate that # all parts of a relative path exist. self.raise_os_error(errno.ENOENT, path) path = self.absnormpath(self._original_path(path)) path = self.replace_windows_root(path) if self._is_root_path(path): return path if path == matching_string(path, self.devnull): return path path_components = self._path_components(path) resolved_components = self._resolve_components(path_components) path = self._components_to_path(resolved_components) # after resolving links, we have to check again for Windows root return self.replace_windows_root(path) # pytype: disable=bad-return-type def _components_to_path(self, component_folders): sep = ( self.get_path_separator(component_folders[0]) if component_folders else self.path_separator ) path = sep.join(component_folders) if not self._starts_with_root_path(path): path = sep + path return path def _resolve_components(self, components: List[AnyStr]) -> List[str]: current_dir = self.root link_depth = 0 path_components = [to_string(comp) for comp in components] resolved_components: List[str] = [] while path_components: component = path_components.pop(0) resolved_components.append(component) directory = self._directory_content(current_dir, component)[1] if directory is None: # The component of the path at this point does not actually # exist in the folder. We can't resolve the path any more. # It is legal to link to a file that does not yet exist, so # rather than raise an error, we just append the remaining # components to what return path we have built so far and # return that. resolved_components.extend(path_components) break # Resolve any possible symlinks in the current path component. elif S_ISLNK(directory.st_mode): # This link_depth check is not really meant to be an accurate # check. It is just a quick hack to prevent us from looping # forever on cycles. if link_depth > _MAX_LINK_DEPTH: self.raise_os_error( errno.ELOOP, self._components_to_path(resolved_components), ) link_path = self._follow_link(resolved_components, directory) # Following the link might result in the complete replacement # of the current_dir, so we evaluate the entire resulting path. target_components = self._path_components(link_path) path_components = target_components + path_components resolved_components = [] current_dir = self.root link_depth += 1 else: current_dir = cast(FakeDirectory, directory) return resolved_components def _valid_relative_path(self, file_path: AnyStr) -> bool: if self.is_windows_fs: return True slash_dotdot = matching_string(file_path, self.path_separator + "..") while file_path and slash_dotdot in file_path: file_path = file_path[: file_path.rfind(slash_dotdot)] if not self.exists(self.absnormpath(file_path)): return False return True def _follow_link(self, link_path_components: List[str], link: AnyFile) -> str: """Follow a link w.r.t. a path resolved so far. The component is either a real file, which is a no-op, or a symlink. In the case of a symlink, we have to modify the path as built up so far /a/b => ../c should yield /a/../c (which will normalize to /a/c) /a/b => x should yield /a/x /a/b => /x/y/z should yield /x/y/z The modified path may land us in a new spot which is itself a link, so we may repeat the process. Args: link_path_components: The resolved path built up to the link so far. link: The link object itself. Returns: (string) The updated path resolved after following the link. Raises: OSError: if there are too many levels of symbolic link """ link_path = link.contents if link_path is not None: # ignore UNC prefix for local files if self.is_windows_fs and link_path.startswith("\\\\?\\"): link_path = link_path[4:] sep = self.get_path_separator(link_path) # For links to absolute paths, we want to throw out everything # in the path built so far and replace with the link. For relative # links, we have to append the link to what we have so far, if not self._starts_with_root_path(link_path): # Relative path. Append remainder of path to what we have # processed so far, excluding the name of the link itself. # /a/b => ../c should yield /a/../c # (which will normalize to /c) # /a/b => d should yield a/d components = link_path_components[:-1] components.append(link_path) link_path = sep.join(components) # Don't call self.NormalizePath(), as we don't want to prepend # self.cwd. return self.normpath(link_path) # pytype: disable=bad-return-type raise ValueError("Invalid link") def get_object_from_normpath( self, file_path: AnyPath, check_read_perm: bool = True, check_exe_perm: bool = True, check_owner: bool = False, ) -> AnyFile: """Search for the specified filesystem object within the fake filesystem. Args: file_path: Specifies target FakeFile object to retrieve, with a path that has already been normalized/resolved. check_read_perm: If `True`, raises `OSError` if a parent directory does not have read permission check_exe_perm: If `True`, raises `OSError` if a parent directory does not have execute (e.g. search) permission check_owner: If `True`, and ``check_read_perm`` is also `True`, only checks read permission if the current user id is different from the file object user id Returns: The FakeFile object corresponding to file_path. Raises: OSError: if the object is not found. """ path = make_string_path(file_path) if path == matching_string(path, self.root.name): return self.root if path == matching_string(path, self.devnull): return self.dev_null path = self._original_path(path) path_components = self._path_components(path) target = self.root try: for component in path_components: if S_ISLNK(target.st_mode): if target.contents: target = cast(FakeDirectory, self.resolve(target.contents)) if not S_ISDIR(target.st_mode): if not self.is_windows_fs: self.raise_os_error(errno.ENOTDIR, path) self.raise_os_error(errno.ENOENT, path) target = target.get_entry(component) # type: ignore if ( not is_root() and (check_read_perm or check_exe_perm) and target and not self._can_read( target, check_read_perm, check_exe_perm, check_owner ) ): self.raise_os_error(errno.EACCES, target.path) except KeyError: self.raise_os_error(errno.ENOENT, path) return target @staticmethod def _can_read(target, check_read_perm, check_exe_perm, owner_can_read): if owner_can_read and target.st_uid == helpers.get_uid(): return True permission = helpers.PERM_READ if check_read_perm else 0 if S_ISDIR(target.st_mode) and check_exe_perm: permission \|= helpers.PERM_EXE if not permission: return True return target.has_permission(permission) def _get_object( self, file_path: AnyPath, check_read_perm: bool = True, check_exe_perm: bool = True, ) -> FakeFile: """Search for the specified filesystem object within the fake filesystem. By default, consider read and execute permissions. Args: file_path: Specifies the target :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object to retrieve. check_read_perm: If `True`, raises `OSError` if a parent directory does not have read permission check_exe_perm: If `True`, raises `OSError` if a parent directory does not have execute (e.g. search) permission Returns: The :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object corresponding to `file_path`. Raises: OSError: if the object is not found. """ path = make_string_path(file_path) path = self.absnormpath(self._original_path(path)) return self.get_object_from_normpath(path, check_read_perm, check_exe_perm) def get_object(self, file_path: AnyPath) -> FakeFile: """Search for the specified filesystem object within the fake filesystem and return it. Ignore any read permissions. Args: file_path: Specifies the target :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object to retrieve. Returns: The :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object corresponding to `file_path`. Raises: OSError: if the object is not found. """ return self._get_object(file_path, check_read_perm=False, check_exe_perm=False) def resolve( self, file_path: Union[AnyStr, int], follow_symlinks: bool = True, allow_fd: bool = False, check_read_perm: bool = True, check_exe_perm: bool = True, check_owner: bool = False, ) -> FakeFile: """Search for the specified filesystem object, resolving all links. Args: file_path: Specifies the target FakeFile object to retrieve. follow_symlinks: If `False`, the link itself is resolved, otherwise the object linked to. allow_fd: If `True`, `file_path` may be an open file descriptor check_read_perm: If `True`, raises `OSError` if a parent directory does not have read permission check_read_perm: If `True`, raises `OSError` if a parent directory does not have execute permission check_owner: If `True`, and ``check_read_perm`` is also `True`, only checks read permission if the current user id is different from the file object user id Returns: The FakeFile object corresponding to ``file_path``. Raises: OSError: if the object is not found. """ if isinstance(file_path, int): if allow_fd: open_file = self.get_open_file(file_path).get_object() assert isinstance(open_file, FakeFile) return open_file raise TypeError("path should be string, bytes or os.PathLike, not int") if follow_symlinks: return self.get_object_from_normpath( self.resolve_path(file_path, allow_fd), check_read_perm, check_exe_perm, check_owner, ) return self.lresolve(file_path) def lresolve(self, path: AnyPath) -> FakeFile: """Search for the specified object, resolving only parent links. This is analogous to the stat/lstat difference. This resolves links to the object but not of the final object itself. Args: path: Specifies target FakeFile object to retrieve. Returns: The FakeFile object corresponding to path. Raises: OSError: if the object is not found. """ path_str = make_string_path(path) if not path_str: self.raise_os_error(errno.ENOENT, path_str) if path_str == matching_string(path_str, self.root.name): # The root directory will never be a link return self.root # remove trailing separator path_str = self._path_without_trailing_separators(path_str) if path_str == matching_string(path_str, "."): path_str = matching_string(path_str, self.cwd) path_str = self._original_path(path_str) parent_directory, child_name = self.splitpath(path_str) if not parent_directory: parent_directory = matching_string(path_str, self.cwd) try: parent_obj = self.resolve(parent_directory) assert parent_obj if not isinstance(parent_obj, FakeDirectory): if not self.is_windows_fs and isinstance(parent_obj, FakeFile): self.raise_os_error(errno.ENOTDIR, path_str) self.raise_os_error(errno.ENOENT, path_str) if not parent_obj.has_permission(helpers.PERM_READ): self.raise_os_error(errno.EACCES, parent_directory) return ( parent_obj.get_entry(to_string(child_name)) if child_name else parent_obj ) except KeyError: pass self.raise_os_error(errno.ENOENT, path_str) def add_object(self, file_path: AnyStr, file_object: AnyFile) -> None: """Add a fake file or directory into the filesystem at file_path. Args: file_path: The path to the file to be added relative to self. file_object: File or directory to add. Raises: OSError: if file_path does not correspond to a directory. """ if not file_path: target_directory = self.root_dir else: target_directory = cast( FakeDirectory, self.resolve(file_path, check_read_perm=False, check_exe_perm=True), ) if not S_ISDIR(target_directory.st_mode): error = errno.ENOENT if self.is_windows_fs else errno.ENOTDIR self.raise_os_error(error, file_path) target_directory.add_entry(file_object) def rename( self, old_file_path: AnyPath, new_file_path: AnyPath, force_replace: bool = False, ) -> None: """Renames a FakeFile object at old_file_path to new_file_path, preserving all properties. Args: old_file_path: Path to filesystem object to rename. new_file_path: Path to where the filesystem object will live after this call. force_replace: If set and destination is an existing file, it will be replaced even under Windows if the user has permissions, otherwise replacement happens under Unix only. Raises: OSError: if old_file_path does not exist. OSError: if new_file_path is an existing directory (Windows, or Posix if old_file_path points to a regular file) OSError: if old_file_path is a directory and new_file_path a file OSError: if new_file_path is an existing file and force_replace not set (Windows only). OSError: if new_file_path is an existing file and could not be removed (Posix, or Windows with force_replace set). OSError: if dirname(new_file_path) does not exist. OSError: if the file would be moved to another filesystem (e.g. mount point). """ old_path = make_string_path(old_file_path) new_path = make_string_path(new_file_path) ends_with_sep = self.ends_with_path_separator(old_path) old_path = self.absnormpath(old_path) new_path = self.absnormpath(new_path) if not self.exists(old_path, check_link=True): self.raise_os_error(errno.ENOENT, old_path, 2) if ends_with_sep: self._handle_broken_link_with_trailing_sep(old_path) old_object = self.lresolve(old_path) if not self.is_windows_fs: self._handle_posix_dir_link_errors(new_path, old_path, ends_with_sep) if self.exists(new_path, check_link=True): renamed_path = self._rename_to_existing_path( force_replace, new_path, old_path, old_object, ends_with_sep ) if renamed_path is None: return else: new_path = renamed_path old_dir, old_name = self.splitpath(old_path) new_dir, new_name = self.splitpath(new_path) if not self.exists(new_dir): self.raise_os_error(errno.ENOENT, new_dir) old_dir_object = self.resolve(old_dir) new_dir_object = self.resolve(new_dir) if old_dir_object.st_dev != new_dir_object.st_dev: self.raise_os_error(errno.EXDEV, old_path) if not S_ISDIR(new_dir_object.st_mode): self.raise_os_error( errno.EINVAL if self.is_windows_fs else errno.ENOTDIR, new_path ) if new_dir_object.has_parent_object(old_object): self.raise_os_error(errno.EINVAL, new_path) self._do_rename(old_dir_object, old_name, new_dir_object, new_name) def _do_rename(self, old_dir_object, old_name, new_dir_object, new_name): object_to_rename = old_dir_object.get_entry(old_name) old_dir_object.remove_entry(old_name, recursive=False) object_to_rename.name = new_name new_name = new_dir_object._normalized_entryname(new_name) old_entry = ( new_dir_object.get_entry(new_name) if new_name in new_dir_object.entries else None ) try: if old_entry: # in case of overwriting remove the old entry first new_dir_object.remove_entry(new_name) new_dir_object.add_entry(object_to_rename) except OSError: # adding failed, roll back the changes before re-raising if old_entry and new_name not in new_dir_object.entries: new_dir_object.add_entry(old_entry) object_to_rename.name = old_name old_dir_object.add_entry(object_to_rename) raise def _handle_broken_link_with_trailing_sep(self, path: AnyStr) -> None: # note that the check for trailing sep has to be done earlier if self.islink(path): if not self.exists(path): error = errno.ENOENT if self.is_macos else errno.ENOTDIR self.raise_os_error(error, path) def _handle_posix_dir_link_errors( self, new_file_path: AnyStr, old_file_path: AnyStr, ends_with_sep: bool ) -> None: if self.isdir(old_file_path, follow_symlinks=False) and self.islink( new_file_path ): self.raise_os_error(errno.ENOTDIR, new_file_path) if self.isdir(new_file_path, follow_symlinks=False) and self.islink( old_file_path ): if ends_with_sep and self.is_macos: return error = errno.ENOTDIR if ends_with_sep else errno.EISDIR self.raise_os_error(error, new_file_path) if ( ends_with_sep and self.islink(old_file_path) and old_file_path == new_file_path and not self.is_windows_fs ): self.raise_os_error(errno.ENOTDIR, new_file_path) def _rename_to_existing_path( self, force_replace: bool, new_file_path: AnyStr, old_file_path: AnyStr, old_object: FakeFile, ends_with_sep: bool, ) -> Optional[AnyStr]: new_object = self._get_object(new_file_path) if old_file_path == new_file_path: if not S_ISLNK(new_object.st_mode) and ends_with_sep: error = errno.ENOTDIR if self.is_windows_fs else errno.ENOTDIR self.raise_os_error(error, old_file_path) return None # Nothing to do here if old_object == new_object: return self._rename_same_object(new_file_path, old_file_path) if S_ISDIR(new_object.st_mode) or S_ISLNK(new_object.st_mode): self._handle_rename_error_for_dir_or_link( force_replace, new_file_path, new_object, old_object, ends_with_sep, ) elif S_ISDIR(old_object.st_mode): error = errno.EEXIST if self.is_windows_fs else errno.ENOTDIR self.raise_os_error(error, new_file_path) elif self.is_windows_fs and not force_replace: self.raise_os_error(errno.EEXIST, new_file_path) else: self.remove_object(new_file_path) return new_file_path def _handle_rename_error_for_dir_or_link( self, force_replace: bool, new_file_path: AnyStr, new_object: FakeFile, old_object: FakeFile, ends_with_sep: bool, ) -> None: if self.is_windows_fs: if force_replace: self.raise_os_error(errno.EACCES, new_file_path) else: self.raise_os_error(errno.EEXIST, new_file_path) if not S_ISLNK(new_object.st_mode): if new_object.entries: if ( not S_ISLNK(old_object.st_mode) or not ends_with_sep or not self.is_macos ): self.raise_os_error(errno.ENOTEMPTY, new_file_path) if S_ISREG(old_object.st_mode): self.raise_os_error(errno.EISDIR, new_file_path) def _rename_same_object( self, new_file_path: AnyStr, old_file_path: AnyStr ) -> Optional[AnyStr]: do_rename = old_file_path.lower() == new_file_path.lower() if not do_rename: try: real_old_path = self.resolve_path(old_file_path) original_old_path = self._original_path(real_old_path) real_new_path = self.resolve_path(new_file_path) if real_new_path == original_old_path and ( new_file_path == real_old_path ) == (new_file_path.lower() == real_old_path.lower()): real_object = self.resolve(old_file_path, follow_symlinks=False) do_rename = ( os.path.basename(old_file_path) == real_object.name or not self.is_macos ) else: do_rename = real_new_path.lower() == real_old_path.lower() if do_rename: # only case is changed in case-insensitive file # system - do the rename parent, file_name = self.splitpath(new_file_path) new_file_path = self.joinpaths( self._original_path(parent), file_name ) except OSError: # ResolvePath may fail due to symlink loop issues or # similar - in this case just assume the paths # to be different pass if not do_rename: # hard links to the same file - nothing to do return None return new_file_path def remove_object(self, file_path: AnyStr) -> None: """Remove an existing file or directory. Args: file_path: The path to the file relative to self. Raises: OSError: if file_path does not correspond to an existing file, or if part of the path refers to something other than a directory. OSError: if the directory is in use (eg, if it is '/'). """ file_path = self.absnormpath(self._original_path(file_path)) if self._is_root_path(file_path): self.raise_os_error(errno.EBUSY, file_path) try: dirname, basename = self.splitpath(file_path) target_directory = self.resolve(dirname, check_read_perm=False) target_directory.remove_entry(basename) except KeyError: self.raise_os_error(errno.ENOENT, file_path) except AttributeError: self.raise_os_error(errno.ENOTDIR, file_path) def make_string_path(self, path: AnyPath) -> AnyStr: # type: ignore[type-var] path_str = make_string_path(path) os_sep = matching_string(path_str, os.sep) fake_sep = self.get_path_separator(path_str) return path_str.replace(os_sep, fake_sep) # type: ignore[return-value] def create_dir( self, directory_path: AnyPath, perm_bits: int = helpers.PERM_DEF, apply_umask: bool = True, ) -> FakeDirectory: """Create `directory_path` and all the parent directories, and return the created :py:class:`FakeDirectory<pyfakefs.fake_file.FakeDirectory>` object. Helper method to set up your test faster. Args: directory_path: The full directory path to create. perm_bits: The permission bits as set by ``chmod``. apply_umask: If `True` (default), the current umask is applied to `perm_bits`. Returns: The newly created :py:class:`FakeDirectory<pyfakefs.fake_file.FakeDirectory>` object. Raises: OSError: if the directory already exists. """ dir_path = self.make_string_path(directory_path) dir_path = self.absnormpath(dir_path) self._auto_mount_drive_if_needed(dir_path) if self.exists(dir_path, check_link=True) and dir_path not in self.mount_points: self.raise_os_error(errno.EEXIST, dir_path) path_components = self._path_components(dir_path) current_dir = self.root new_dirs = [] for component in [to_string(p) for p in path_components]: directory = self._directory_content(current_dir, to_string(component))[1] if not directory: new_dir = FakeDirectory(component, filesystem=self) new_dirs.append(new_dir) if self.is_windows_fs and current_dir == self.root: current_dir = self.root_dir current_dir.add_entry(new_dir) current_dir = new_dir else: if S_ISLNK(directory.st_mode): assert directory.contents directory = self.resolve(directory.contents) assert directory current_dir = cast(FakeDirectory, directory) if directory.st_mode & S_IFDIR != S_IFDIR: self.raise_os_error(errno.ENOTDIR, current_dir.path) # set the permission after creating the directories # to allow directory creation inside a read-only directory for new_dir in new_dirs: if apply_umask: perm_bits &= ~self.umask new_dir.st_mode = S_IFDIR \| perm_bits return current_dir def create_file( self, file_path: AnyPath, st_mode: int = S_IFREG \| helpers.PERM_DEF_FILE, contents: AnyString = "", st_size: Optional[int] = None, create_missing_dirs: bool = True, apply_umask: bool = True, encoding: Optional[str] = None, errors: Optional[str] = None, side_effect: Optional[Callable] = None, ) -> FakeFile: """Create `file_path`, including all the parent directories along the way, and return the created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object. This helper method can be used to set up tests more easily. Args: file_path: The path to the file to create. st_mode: The `stat` constant representing the file type. contents: the contents of the file. If not given and `st_size` is `None`, an empty file is assumed. st_size: file size; only valid if contents not given. If given, the file is considered to be in "large file mode" and trying to read from or write to the file will result in an exception. create_missing_dirs: If `True`, auto create missing directories. apply_umask: If `True` (default), the current umask is applied to `st_mode`. encoding: If ``contents`` is of type `str`, the encoding used for serialization. errors: The error mode used for encoding/decoding errors. side_effect: function handle that is executed when the file is written, must accept the file object as an argument. Returns: The newly created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object. Raises: OSError: if the file already exists. OSError: if the containing directory is required and missing. """ return self.create_file_internally( file_path, st_mode, contents, st_size, create_missing_dirs, apply_umask, encoding, errors, side_effect=side_effect, ) def add_real_file( self, source_path: AnyPath, read_only: bool = True, target_path: Optional[AnyPath] = None, ) -> FakeFile: """Create `file_path`, including all the parent directories along the way, for an existing real file, and return the created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object. The contents of the real file are read only on demand. Args: source_path: Path to an existing file in the real file system read_only: If `True` (the default), writing to the fake file raises an exception. Otherwise, writing to the file changes the fake file only. target_path: If given, the path of the target direction, otherwise it is equal to `source_path`. Returns: the newly created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object. Raises: OSError: if the file does not exist in the real file system. OSError: if the file already exists in the fake file system. .. note:: On most systems, accessing the fake file's contents may update both the real and fake files' `atime` (access time). In this particular case, `add_real_file()` violates the rule that `pyfakefs` must not modify the real file system. """ target_path = target_path or source_path source_path_str = make_string_path(source_path) real_stat = os.stat(source_path_str) fake_file = self.create_file_internally(target_path, read_from_real_fs=True) # for read-only mode, remove the write/executable permission bits fake_file.stat_result.set_from_stat_result(real_stat) if read_only: fake_file.st_mode &= 0o777444 fake_file.file_path = source_path_str self.change_disk_usage(fake_file.size, fake_file.name, fake_file.st_dev) return fake_file def add_real_symlink( self, source_path: AnyPath, target_path: Optional[AnyPath] = None ) -> FakeFile: """Create a symlink at `source_path` (or `target_path`, if given) and return the created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object. It will point to the same path as the symlink on the real filesystem. Relative symlinks will point relative to their new location. Absolute symlinks will point to the same, absolute path as on the real filesystem. Args: source_path: The path to the existing symlink. target_path: If given, the name of the symlink in the fake filesystem, otherwise, the same as `source_path`. Returns: the newly created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object. Raises: OSError: if the directory does not exist in the real file system. OSError: if the symlink could not be created (see :py:meth:`create_file`). OSError: if the directory already exists in the fake file system. """ source_path_str = make_string_path(source_path) # TODO: add test source_path_str = self._path_without_trailing_separators(source_path_str) if not os.path.exists(source_path_str) and not os.path.islink(source_path_str): self.raise_os_error(errno.ENOENT, source_path_str) target = os.readlink(source_path_str) if target_path: return self.create_symlink(target_path, target) else: return self.create_symlink(source_path_str, target) def add_real_directory( self, source_path: AnyPath, read_only: bool = True, lazy_read: bool = True, target_path: Optional[AnyPath] = None, ) -> FakeDirectory: """Create a fake directory corresponding to the real directory at the specified path, and return the created :py:class:`FakeDirectory<pyfakefs.fake_file.FakeDirectory>` object. Add entries in the fake directory corresponding to the entries in the real directory. Symlinks are supported. If the target directory already exists in the fake filesystem, the directory contents are merged. Overwriting existing files is not allowed. Args: source_path: The path to the existing directory. read_only: If set, all files under the directory are treated as read-only, e.g. a write access raises an exception; otherwise, writing to the files changes the fake files only as usually. lazy_read: If set (default), directory contents are only read when accessed, and only until the needed subdirectory level. .. note:: This means that the file system size is only updated at the time the directory contents are read; set this to `False` only if you are dependent on accurate file system size in your test target_path: If given, the target directory, otherwise, the target directory is the same as `source_path`. Returns: the newly created :py:class:`FakeDirectory<pyfakefs.fake_file.FakeDirectory>` object. Raises: OSError: if the directory does not exist in the real filesystem. OSError: if a file or link exists in the fake filesystem where a real file or directory shall be mapped. """ source_path_str = make_string_path(source_path) source_path_str = self._path_without_trailing_separators(source_path_str) if not os.path.exists(source_path_str): self.raise_os_error(errno.ENOENT, source_path_str) target_path_str = make_string_path(target_path or source_path_str) # get rid of inconsistencies between real and fake path separators if os.altsep is not None: target_path_str = os.path.normpath(target_path_str) if os.sep != self.path_separator: target_path_str = target_path_str.replace(os.sep, self.path_separator) self._auto_mount_drive_if_needed(target_path_str) if lazy_read: self._create_fake_from_real_dir_lazily( source_path_str, target_path_str, read_only ) else: self._create_fake_from_real_dir(source_path_str, target_path_str, read_only) return cast(FakeDirectory, self._get_object(target_path_str)) def _create_fake_from_real_dir(self, source_path_str, target_path_str, read_only): if not self.exists(target_path_str): self.create_dir(target_path_str) for base, _, files in os.walk(source_path_str): new_base = os.path.join( target_path_str, os.path.relpath(base, source_path_str), ) for file_entry in os.listdir(base): file_path = os.path.join(base, file_entry) if os.path.islink(file_path): self.add_real_symlink(file_path, os.path.join(new_base, file_entry)) for file_entry in files: path = os.path.join(base, file_entry) if not os.path.islink(path): self.add_real_file( path, read_only, os.path.join(new_base, file_entry) ) def _create_fake_from_real_dir_lazily( self, source_path_str, target_path_str, read_only ): if self.exists(target_path_str): if not self.isdir(target_path_str): raise OSError(errno.ENOTDIR, "Mapping target is not a directory") for entry in os.listdir(source_path_str): src_entry_path = os.path.join(source_path_str, entry) target_entry_path = os.path.join(target_path_str, entry) if os.path.isdir(src_entry_path): self.add_real_directory( src_entry_path, read_only, True, target_entry_path ) elif os.path.islink(src_entry_path): self.add_real_symlink(src_entry_path, target_entry_path) elif os.path.isfile(src_entry_path): self.add_real_file(src_entry_path, read_only, target_entry_path) return self._get_object(target_path_str) parent_path = os.path.split(target_path_str)[0] if self.exists(parent_path): parent_dir = self._get_object(parent_path) else: parent_dir = self.create_dir(parent_path) new_dir = FakeDirectoryFromRealDirectory( source_path_str, self, read_only, target_path_str ) parent_dir.add_entry(new_dir) return new_dir def add_real_paths( self, path_list: List[AnyStr], read_only: bool = True, lazy_dir_read: bool = True, ) -> None: """This convenience method adds multiple files and/or directories from the real file system to the fake file system. See :py:meth:`add_real_file` and :py:meth:`add_real_directory`. Args: path_list: List of file and directory paths in the real file system. read_only: If set, all files and files under the directories are treated as read-only, e.g. a write access raises an exception; otherwise, writing to the files changes the fake files only as usually. lazy_dir_read: Uses lazy reading of directory contents if set (see :py:meth:`add_real_directory`) Raises: OSError: if any of the files and directories in the list does not exist in the real file system. OSError: if a file or link exists in the fake filesystem where a real file or directory shall be mapped. """ for path in path_list: if os.path.isdir(path): self.add_real_directory(path, read_only, lazy_dir_read) else: self.add_real_file(path, read_only) def add_package_metadata(self, package_name: str) -> None: """Convenience function to add all metadata distribution files of the package with the given name to the fake filesystem. These are the files that are usually located under `<package_name>-<version>.dist-info` beside the actual package. The files are accessed by the functions in `importlib.metadata`. Args: package_name: Name of the package whose metadata shall be copied Raises: PackageNotFoundError: if the package with the given name is not found """ if sys.version_info < (3, 8): raise NotImplementedError("Not available in Python 3.7") from importlib.metadata import distribution, PackageNotFoundError # we have to pause patching to get the distribution # from the real filesystem if we are in patch mode if self.patcher: self.pause() try: dist_files = distribution(package_name).files if dist_files is None: raise PackageNotFoundError(package_name) for metadata_file in dist_files: self.add_real_file(metadata_file.locate()) finally: if self.patcher: self.resume() def create_file_internally( self, file_path: AnyPath, st_mode: int = S_IFREG \| helpers.PERM_DEF_FILE, contents: AnyString = "", st_size: Optional[int] = None, create_missing_dirs: bool = True, apply_umask: bool = True, encoding: Optional[str] = None, errors: Optional[str] = None, read_from_real_fs: bool = False, side_effect: Optional[Callable] = None, ) -> FakeFile: """Internal fake file creator that supports both normal fake files and fake files based on real files. Args: file_path: path to the file to create. st_mode: the stat.S_IF constant representing the file type. contents: the contents of the file. If not given and st_size is None, an empty file is assumed. st_size: file size; only valid if contents not given. If given, the file is considered to be in "large file mode" and trying to read from or write to the file will result in an exception. create_missing_dirs: if `True`, auto create missing directories. apply_umask: whether the current umask must be applied on st_mode. encoding: if contents is a Unicode string, the encoding used for serialization. errors: the error mode used for encoding/decoding errors read_from_real_fs: if `True`, the contents are read from the real file system on demand. side_effect: function handle that is executed when file is written, must accept the file object as an argument. """ path = self.make_string_path(file_path) path = self.absnormpath(path) if not is_int_type(st_mode): raise TypeError( "st_mode must be of int type - did you mean to set contents?" ) if self.exists(path, check_link=True): self.raise_os_error(errno.EEXIST, path) parent_directory, new_file = self.splitpath(path) if not parent_directory: parent_directory = matching_string(path, self.cwd) self._auto_mount_drive_if_needed(parent_directory) if not self.exists(parent_directory): if not create_missing_dirs: self.raise_os_error(errno.ENOENT, parent_directory) parent_directory = matching_string( path, self.create_dir(parent_directory).path, # type: ignore ) else: parent_directory = self._original_path(parent_directory) if apply_umask: st_mode &= ~self.umask file_object: FakeFile if read_from_real_fs: file_object = FakeFileFromRealFile( to_string(path), filesystem=self, side_effect=side_effect ) else: file_object = FakeFile( new_file, st_mode, filesystem=self, encoding=encoding, errors=errors, side_effect=side_effect, ) self.add_object(parent_directory, file_object) if st_size is None and contents is None: contents = "" if not read_from_real_fs and (contents is not None or st_size is not None): try: if st_size is not None: file_object.set_large_file_size(st_size) else: file_object.set_initial_contents(contents) # type: ignore except OSError: self.remove_object(path) raise return file_object def create_symlink( self, file_path: AnyPath, link_target: AnyPath, create_missing_dirs: bool = True, ) -> FakeFile: """Create the specified symlink, pointed at the specified link target, and return the created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object representing the link. Args: file_path: path to the symlink to create link_target: the target of the symlink create_missing_dirs: If `True`, any missing parent directories of `file_path` will be created Returns: The newly created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object. Raises: OSError: if the symlink could not be created (see :py:meth:`create_file`). """ link_path = self.make_string_path(file_path) link_target_path = self.make_string_path(link_target) link_path = self.normcase(link_path) # the link path cannot end with a path separator if self.ends_with_path_separator(link_path): if self.exists(link_path): self.raise_os_error(errno.EEXIST, link_path) if self.exists(link_target_path): if not self.is_windows_fs: self.raise_os_error(errno.ENOENT, link_path) else: if self.is_windows_fs: self.raise_os_error(errno.ENOTDIR, link_target_path) if not self.exists( self._path_without_trailing_separators(link_path), check_link=True, ): self.raise_os_error(errno.ENOENT, link_target_path) if self.is_macos: # to avoid EEXIST exception, remove the link # if it already exists if self.exists(link_path, check_link=True): self.remove_object(link_path) else: self.raise_os_error(errno.EEXIST, link_target_path) # resolve the link path only if it is not a link itself if not self.islink(link_path): link_path = self.resolve_path(link_path) permission = helpers.PERM_DEF_FILE if self.is_windows_fs else helpers.PERM_DEF return self.create_file_internally( link_path, st_mode=S_IFLNK \| permission, contents=link_target_path, create_missing_dirs=create_missing_dirs, apply_umask=self.is_macos, ) def create_link( self, old_path: AnyPath, new_path: AnyPath, follow_symlinks: bool = True, create_missing_dirs: bool = True, ) -> FakeFile: """Create a hard link at `new_path`, pointing at `old_path`, and return the created :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object representing the link. Args: old_path: An existing link to the target file. new_path: The destination path to create a new link at. follow_symlinks: If `False` and `old_path` is a symlink, link the symlink instead of the object it points to. create_missing_dirs: If `True`, any missing parent directories of `file_path` will be created Returns: The :py:class:`FakeFile<pyfakefs.fake_file.FakeFile>` object referred to by `old_path`. Raises: OSError: if something already exists at `new_path`. OSError: if `old_path` is a directory. OSError: if the parent directory doesn't exist. """ old_path_str = make_string_path(old_path) new_path_str = make_string_path(new_path) new_path_normalized = self.absnormpath(new_path_str) if self.exists(new_path_normalized, check_link=True): self.raise_os_error(errno.EEXIST, new_path_str) new_parent_directory, new_basename = self.splitpath(new_path_normalized) if not new_parent_directory: new_parent_directory = matching_string(new_path_str, self.cwd) if not self.exists(new_parent_directory): if create_missing_dirs: self.create_dir(new_parent_directory) else: self.raise_os_error(errno.ENOENT, new_parent_directory) if self.ends_with_path_separator(old_path_str): self.raise_os_error(errno.ENOTDIR, old_path_str) if not self.is_windows_fs and self.ends_with_path_separator(new_path): self.raise_os_error(errno.ENOENT, old_path_str) # Retrieve the target file try: old_file = self.resolve(old_path_str, follow_symlinks=follow_symlinks) except OSError: self.raise_os_error(errno.ENOENT, old_path_str) if old_file.st_mode & S_IFDIR: self.raise_os_error( errno.EACCES if self.is_windows_fs else errno.EPERM, old_path_str, ) # abuse the name field to control the filename of the # newly created link old_file.name = new_basename # type: ignore[assignment] self.add_object(new_parent_directory, old_file) return old_file def link( self, old_path: AnyPath, new_path: AnyPath, follow_symlinks: bool = True, ) -> FakeFile: """Create a hard link at new_path, pointing at old_path. Args: old_path: An existing link to the target file. new_path: The destination path to create a new link at. follow_symlinks: If `False` and ``old_path`` is a symlink, link the symlink instead of the object it points to. Returns: The FakeFile object referred to by ``old_path``. Raises: OSError: if something already exists at new_path. OSError: if ``old_path`` is a directory. OSError: if the parent directory doesn't exist. """ return self.create_link( old_path, new_path, follow_symlinks, create_missing_dirs=False ) def _is_circular_link(self, link_obj: FakeFile) -> bool: try: assert link_obj.contents self.resolve_path(link_obj.contents) except OSError as exc: return exc.errno == errno.ELOOP return False def readlink(self, path: AnyPath) -> str: """Read the target of a symlink. Args: path: symlink to read the target of. Returns: the string representing the path to which the symbolic link points. Raises: TypeError: if path is `None` OSError: (with errno=ENOENT) if path is not a valid path, or (with errno=EINVAL) if path is valid, but is not a symlink, or if the path ends with a path separator (Posix only) """ if path is None: raise TypeError link_path = make_string_path(path) link_obj = self.lresolve(link_path) if S_IFMT(link_obj.st_mode) != S_IFLNK: self.raise_os_error(errno.EINVAL, link_path) if self.ends_with_path_separator(link_path): if not self.is_windows_fs and self.exists(link_path): self.raise_os_error(errno.EINVAL, link_path) if not self.exists(link_obj.path): # type: ignore if self.is_windows_fs: error = errno.ENOTDIR elif self._is_circular_link(link_obj): if self.is_macos: return link_obj.path # type: ignore[return-value] error = errno.ELOOP else: error = errno.ENOENT self.raise_os_error(error, link_obj.path) assert link_obj.contents return link_obj.contents def makedir(self, dir_path: AnyPath, mode: int = helpers.PERM_DEF) -> None: """Create a leaf Fake directory. Args: dir_path: (str) Name of directory to create. Relative paths are assumed to be relative to '/'. mode: (int) Mode to create directory with. This argument defaults to 0o777. The umask is applied to this mode. Raises: OSError: if the directory name is invalid or parent directory is read only or as per :py:meth:`add_object`. """ dir_name = make_string_path(dir_path) ends_with_sep = self.ends_with_path_separator(dir_name) dir_name = self._path_without_trailing_separators(dir_name) if not dir_name: self.raise_os_error(errno.ENOENT, "") if self.is_windows_fs: dir_name = self.absnormpath(dir_name) parent_dir, rest = self.splitpath(dir_name) if parent_dir: base_dir = self.normpath(parent_dir) ellipsis = matching_string(parent_dir, self.path_separator + "..") if parent_dir.endswith(ellipsis) and not self.is_windows_fs: base_dir, dummy_dotdot, _ = parent_dir.partition(ellipsis) if self.is_windows_fs and not rest and not self.exists(base_dir): # under Windows, the parent dir may be a drive or UNC path # which has to be mounted self._auto_mount_drive_if_needed(parent_dir) if not self.exists(base_dir): self.raise_os_error(errno.ENOENT, base_dir) dir_name = self.absnormpath(dir_name) if self.exists(dir_name, check_link=True): if self.is_windows_fs and dir_name == self.root_dir_name: error_nr = errno.EACCES else: error_nr = errno.EEXIST if ends_with_sep and self.is_macos and not self.exists(dir_name): # to avoid EEXIST exception, remove the link self.remove_object(dir_name) else: self.raise_os_error(error_nr, dir_name) head, tail = self.splitpath(dir_name) self.add_object( to_string(head), FakeDirectory(to_string(tail), mode & ~self.umask, filesystem=self), ) def _path_without_trailing_separators(self, path: AnyStr) -> AnyStr: while self.ends_with_path_separator(path): path = path[:-1] return path def makedirs( self, dir_name: AnyStr, mode: int = helpers.PERM_DEF, exist_ok: bool = False ) -> None: """Create a leaf Fake directory and create any non-existent parent dirs. Args: dir_name: (str) Name of directory to create. mode: (int) Mode to create directory (and any necessary parent directories) with. This argument defaults to 0o777. The umask is applied to this mode. exist_ok: (boolean) If exist_ok is `False` (the default), an `OSError` is raised if the target directory already exists. Raises: OSError: if the directory already exists and exist_ok=False, or as per :py:meth:`create_dir`. """ if not dir_name: self.raise_os_error(errno.ENOENT, "") ends_with_sep = self.ends_with_path_separator(dir_name) dir_name = self.absnormpath(dir_name) if ( ends_with_sep and self.is_macos and self.exists(dir_name, check_link=True) and not self.exists(dir_name) ): # to avoid EEXIST exception, remove the link self.remove_object(dir_name) dir_name_str = to_string(dir_name) path_components = self._path_components(dir_name_str) # Raise a permission denied error if the first existing directory # is not writeable. current_dir = self.root_dir for component in path_components: if ( not hasattr(current_dir, "entries") or component not in current_dir.entries ): break else: current_dir = cast(FakeDirectory, current_dir.entries[component]) try: self.create_dir(dir_name, mode) except OSError as e: if e.errno == errno.EACCES: # permission denied - propagate exception raise if not exist_ok or not isinstance(self.resolve(dir_name), FakeDirectory): if self.is_windows_fs and e.errno == errno.ENOTDIR: e.errno = errno.ENOENT # mypy thinks that errno may be None self.raise_os_error(cast(int, e.errno), e.filename) def _is_of_type( self, path: AnyPath, st_flag: int, follow_symlinks: bool = True, ) -> bool: """Helper function to implement isdir(), islink(), etc. See the stat(2) man page for valid stat.S_I* flag values Args: path: Path to file to stat and test st_flag: The stat.S_I* flag checked for the file's st_mode follow_symlinks: If `False` and path points to a symlink, the link itself is checked instead of the linked object. Returns: (boolean) `True` if the st_flag is set in path's st_mode. Raises: TypeError: if path is `None` """ if path is None: raise TypeError file_path = make_string_path(path) try: obj = self.resolve(file_path, follow_symlinks, check_read_perm=False) if obj: self.raise_for_filepath_ending_with_separator( file_path, obj, macos_handling=not follow_symlinks ) return S_IFMT(obj.st_mode) == st_flag except OSError: return False return False def isdir(self, path: AnyPath, follow_symlinks: bool = True) -> bool: """Determine if path identifies a directory. Args: path: Path to filesystem object. follow_symlinks: If `False` and path points to a symlink, the link itself is checked instead of the linked object. Returns: `True` if path points to a directory (following symlinks). Raises: TypeError: if path is `None`. """ return self._is_of_type(path, S_IFDIR, follow_symlinks) def isfile(self, path: AnyPath, follow_symlinks: bool = True) -> bool: """Determine if path identifies a regular file. Args: path: Path to filesystem object. follow_symlinks: If `False` and path points to a symlink, the link itself is checked instead of the linked object. Returns: `True` if path points to a regular file (following symlinks). Raises: TypeError: if path is `None`. """ return self._is_of_type(path, S_IFREG, follow_symlinks) def islink(self, path: AnyPath) -> bool: """Determine if path identifies a symbolic link. Args: path: Path to filesystem object. Returns: `True` if path points to a symlink (S_IFLNK set in st_mode) Raises: TypeError: if path is `None`. """ return self._is_of_type(path, S_IFLNK, follow_symlinks=False) if sys.version_info >= (3, 12): def isjunction(self, path: AnyPath) -> bool: """Returns `False`. Junctions are never faked.""" return False def confirmdir( self, target_directory: AnyStr, check_read_perm: bool = True, check_exe_perm: bool = True, check_owner: bool = False, ) -> FakeDirectory: """Test that the target is actually a directory, raising OSError if not. Args: target_directory: Path to the target directory within the fake filesystem. check_read_perm: If `True`, raises `OSError` if the directory does not have read permission check_exe_perm: If `True`, raises `OSError` if the directory does not have execute (e.g. search) permission check_owner: If `True`, only checks read permission if the current user id is different from the file object user id Returns: The FakeDirectory object corresponding to target_directory. Raises: OSError: if the target is not a directory. """ directory = cast( FakeDirectory, self.resolve( target_directory, check_read_perm=check_read_perm, check_exe_perm=check_exe_perm, check_owner=check_owner, ), ) if not directory.st_mode & S_IFDIR: self.raise_os_error(errno.ENOTDIR, target_directory, 267) return directory def remove(self, path: AnyStr) -> None: """Remove the FakeFile object at the specified file path. Args: path: Path to file to be removed. Raises: OSError: if path points to a directory. OSError: if path does not exist. OSError: if removal failed. """ norm_path = make_string_path(path) norm_path = self.absnormpath(norm_path) if self.ends_with_path_separator(path): self._handle_broken_link_with_trailing_sep(norm_path) if self.exists(norm_path): obj = self.resolve(norm_path, check_read_perm=False) if S_IFMT(obj.st_mode) == S_IFDIR: link_obj = self.lresolve(norm_path) if S_IFMT(link_obj.st_mode) != S_IFLNK: if self.is_windows_fs: error = errno.EACCES elif self.is_macos: error = errno.EPERM else: error = errno.EISDIR self.raise_os_error(error, norm_path) if path.endswith(self.get_path_separator(path)): if self.is_windows_fs: error = errno.EACCES elif self.is_macos: error = errno.EPERM else: error = errno.ENOTDIR self.raise_os_error(error, norm_path) else: self.raise_for_filepath_ending_with_separator(path, obj) self.remove_object(norm_path) def rmdir(self, target_directory: AnyStr, allow_symlink: bool = False) -> None: """Remove a leaf Fake directory. Args: target_directory: (str) Name of directory to remove. allow_symlink: (bool) if `target_directory` is a symlink, the function just returns, otherwise it raises (Posix only) Raises: OSError: if target_directory does not exist. OSError: if target_directory does not point to a directory. OSError: if removal failed per FakeFilesystem.RemoveObject. Cannot remove '.'. """ if target_directory == matching_string(target_directory, "."): error_nr = errno.EACCES if self.is_windows_fs else errno.EINVAL self.raise_os_error(error_nr, target_directory) ends_with_sep = self.ends_with_path_separator(target_directory) target_directory = self.absnormpath(target_directory) if self.confirmdir(target_directory, check_owner=True): if not self.is_windows_fs and self.islink(target_directory): if allow_symlink: return if not ends_with_sep or not self.is_macos: self.raise_os_error(errno.ENOTDIR, target_directory) dir_object = self.resolve(target_directory, check_owner=True) if dir_object.entries: self.raise_os_error(errno.ENOTEMPTY, target_directory) self.remove_object(target_directory) def listdir(self, target_directory: AnyStr) -> List[AnyStr]: """Return a list of file names in target_directory. Args: target_directory: Path to the target directory within the fake filesystem. Returns: A list of file names within the target directory in arbitrary order. If `shuffle_listdir_results` is set, the order is not the same in subsequent calls to avoid tests relying on any ordering. Raises: OSError: if the target is not a directory. """ if target_directory is None: return [] target_directory = self.resolve_path(target_directory, allow_fd=True) directory = self.confirmdir(target_directory, check_exe_perm=False) directory_contents = list(directory.entries.keys()) if self.shuffle_listdir_results: random.shuffle(directory_contents) return directory_contents # type: ignore[return-value] def __str__(self) -> str: return str(self.root_dir) if sys.version_info >= (3, 13): # used for emulating Windows _WIN_RESERVED_NAMES = frozenset( {"CON", "PRN", "AUX", "NUL", "CONIN$", "CONOUT$"} \| {f"COM{c}" for c in "123456789\xb9\xb2\xb3"} \| {f"LPT{c}" for c in "123456789\xb9\xb2\xb3"} ) _WIN_RESERVED_CHARS = frozenset( {chr(i) for i in range(32)} \| {'"', "*", ":", "<", ">", "?", "\|", "/", "\\"} ) def isreserved(self, path): if not self.is_windows_fs: return False def is_reserved_name(name): if sys.platform == "win32": from os.path import _isreservedname # type: ignore[import-error] return _isreservedname(name) if name[-1:] in (".", " "): return name not in (".", "..") if self._WIN_RESERVED_CHARS.intersection(name): return True name = name.partition(".")[0].rstrip(" ").upper() return name in self._WIN_RESERVED_NAMES path = os.fsdecode(self.splitroot(path)[2]) if self.alternative_path_separator is not None: path = path.replace( self.alternative_path_separator, self.path_separator ) return any( is_reserved_name(name) for name in reversed(path.split(self.path_separator)) ) def _add_standard_streams(self) -> None: self.add_open_file(StandardStreamWrapper(sys.stdin)) self.add_open_file(StandardStreamWrapper(sys.stdout)) self.add_open_file(StandardStreamWrapper(sys.stderr)) def _tempdir_name(self): """This logic is extracted from tempdir._candidate_tempdir_list. We cannot rely on tempdir.gettempdir() in an empty filesystem, as it tries to write to the filesystem to ensure that the tempdir is valid. """ # reset the cached tempdir in tempfile tempfile.tempdir = None for env_name in "TMPDIR", "TEMP", "TMP": dir_name = os.getenv(env_name) if dir_name: return dir_name # we have to check the real OS temp path here, as this is what # tempfile assumes if os.name == "nt": return os.path.expanduser(r"~\AppData\Local\Temp") return "/tmp" def _create_temp_dir(self): # the temp directory is assumed to exist at least in `tempfile`, # so we create it here for convenience temp_dir = self._tempdir_name() if not self.exists(temp_dir): self.create_dir(temp_dir) if sys.platform != "win32" and not self.exists("/tmp"): # under Posix, we also create a link in /tmp if the path does not exist self.create_symlink("/tmp", temp_dir) # reset the used size to 0 to avoid having the link size counted # which would make disk size tests more complicated next(iter(self.mount_points.values()))["used_size"] = 0 def _run_doctest() -> TestResults: import doctest import pyfakefs return doctest.testmod(pyfakefs.fake_filesystem) def __getattr__(name): # backwards compatibility for read access to globals moved to helpers if name == "USER_ID": return helpers.USER_ID if name == "GROUP_ID": return helpers.GROUP_ID raise AttributeError(f"No attribute {name!r}.") if __name__ == "__main__": _run_doctest()	127.07 KB	2026-01-20 12:58
📄 fake_filesystem_shutil.py # Copyright 2009 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """A fake shutil module implementation that uses fake_filesystem for unit tests. Note that only `shutildisk_usage()` is faked, the rest of the functions shall work fine with the fake file system if `os`/`os.path` are patched. :Includes: FakeShutil: Uses a FakeFilesystem to provide a fake replacement for the shutil module. :Usage: The fake implementation is automatically involved if using `fake_filesystem_unittest.TestCase`, pytest fs fixture, or directly `Patcher`. """ import functools import os import shutil import sys from threading import RLock from typing import Callable class FakeShutilModule: """Uses a FakeFilesystem to provide a fake replacement for shutil module. Automatically created if using `fake_filesystem_unittest.TestCase`, the `fs` fixture, the `patchfs` decorator, or directly the `Patcher`. """ module_lock = RLock() use_copy_file_range = ( hasattr(shutil, "_USE_CP_COPY_FILE_RANGE") and shutil._USE_CP_COPY_FILE_RANGE # type: ignore[attr-defined] ) has_fcopy_file = hasattr(shutil, "_HAS_FCOPYFILE") and shutil._HAS_FCOPYFILE # type: ignore[attr-defined] use_sendfile = hasattr(shutil, "_USE_CP_SENDFILE") and shutil._USE_CP_SENDFILE # type: ignore[attr-defined] use_fd_functions = shutil._use_fd_functions # type: ignore[attr-defined] functions_to_patch = ["copy", "copyfile", "rmtree"] if sys.version_info < (3, 12) or sys.platform != "win32": functions_to_patch.extend(["copy2", "copytree", "move"]) @staticmethod def dir(): """Return the list of patched function names. Used for patching functions imported from the module. """ return ("disk_usage",) def __init__(self, filesystem): """Construct fake shutil module using the fake filesystem. Args: filesystem: FakeFilesystem used to provide file system information """ self.filesystem = filesystem self.shutil_module = shutil self._patch_level = 0 def _start_patching_global_vars(self): self._patch_level += 1 if self._patch_level > 1: return # nested call - already patched if self.has_fcopy_file: self.shutil_module._HAS_FCOPYFILE = False if self.use_copy_file_range: self.shutil_module._USE_CP_COPY_FILE_RANGE = False if self.use_sendfile: self.shutil_module._USE_CP_SENDFILE = False if self.use_fd_functions: if sys.version_info >= (3, 14): self.shutil_module._rmtree_impl = ( self.shutil_module._rmtree_unsafe # type: ignore[attr-defined] ) else: self.shutil_module._use_fd_functions = False def _stop_patching_global_vars(self): self._patch_level -= 1 if self._patch_level > 0: return # nested call - remains patched if self.has_fcopy_file: self.shutil_module._HAS_FCOPYFILE = True if self.use_copy_file_range: self.shutil_module._USE_CP_COPY_FILE_RANGE = True if self.use_sendfile: self.shutil_module._USE_CP_SENDFILE = True if self.use_fd_functions: if sys.version_info >= (3, 14): self.shutil_module._rmtree_impl = ( self.shutil_module._rmtree_safe_fd # type: ignore[attr-defined] ) else: self.shutil_module._use_fd_functions = True def with_patched_globals(self, f: Callable) -> Callable: """Function wrapper that patches global variables during function execution. Can be used in multi-threading code. """ @functools.wraps(f) def wrapped(args, kwargs): with self.module_lock: self._start_patching_global_vars() try: return f(args, *kwargs) finally: self._stop_patching_global_vars() return wrapped def disk_usage(self, path): """Return the total, used and free disk space in bytes as named tuple or placeholder holder values simulating unlimited space if not set. Args: path: defines the filesystem device which is queried """ return self.filesystem.get_disk_usage(path) if sys.version_info >= (3, 12) and sys.platform == "win32": def copy2(self, src, dst, , follow_symlinks=True): """Since Python 3.12, there is an optimization fow Windows, using the Windows API. We just remove this and fall back to the previous implementation. """ if self.filesystem.isdir(dst): dst = self.filesystem.joinpaths(dst, os.path.basename(src)) self.copyfile(src, dst, follow_symlinks=follow_symlinks) self.copystat(src, dst, follow_symlinks=follow_symlinks) return dst def copytree( self, src, dst, symlinks=False, ignore=None, copy_function=shutil.copy2, ignore_dangling_symlinks=False, dirs_exist_ok=False, ): """Make sure the default argument is patched.""" if copy_function == shutil.copy2: copy_function = self.copy2 return self.shutil_module.copytree( src, dst, symlinks, ignore, copy_function, ignore_dangling_symlinks, dirs_exist_ok, ) def move(self, src, dst, copy_function=shutil.copy2): """Make sure the default argument is patched.""" if copy_function == shutil.copy2: copy_function = self.copy2 return self.shutil_module.move(src, dst, copy_function) def __getattr__(self, name): """Forwards any non-faked calls to the standard shutil module.""" if name in self.functions_to_patch: return self.with_patched_globals(getattr(self.shutil_module, name)) return getattr(self.shutil_module, name)	6.64 KB	2026-01-20 12:58
📄 fake_filesystem_unittest.py # Copyright 2014 Altera Corporation. All Rights Reserved. # Copyright 2015-2017 John McGehee # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """This module provides a base class derived from `unittest.TestClass` for unit tests using the :py:class:`pyfakefs` module. `fake_filesystem_unittest.TestCase` searches `sys.modules` for modules that import the `os`, `io`, `path` `shutil`, and `pathlib` modules. The `setUpPyfakefs()` method binds these modules to the corresponding fake modules from `pyfakefs`. Further, the `open()` built-in is bound to a fake `open()`. It is expected that `setUpPyfakefs()` be invoked at the beginning of the derived class' `setUp()` method. There is no need to add anything to the derived class' `tearDown()` method. During the test, everything uses the fake file system and modules. This means that even in your test fixture, familiar functions like `open()` and `os.makedirs()` manipulate the fake file system. Existing unit tests that use the real file system can be retrofitted to use pyfakefs by simply changing their base class from `:py:class`unittest.TestCase` to `:py:class`pyfakefs.fake_filesystem_unittest.TestCase`. """ import _io # type:ignore[import] import doctest import functools import genericpath import glob import inspect import io import linecache import os import sys import tempfile import tokenize import unittest import warnings from importlib import reload from importlib.abc import Loader, MetaPathFinder from importlib.machinery import ModuleSpec from importlib.util import spec_from_file_location, module_from_spec from types import ModuleType, TracebackType, FunctionType from typing import ( Any, Callable, Dict, List, Set, Tuple, Optional, Union, Type, Iterator, cast, ItemsView, Sequence, ) from unittest import TestSuite from pyfakefs import fake_filesystem, fake_io, fake_os, fake_open, fake_path, fake_file from pyfakefs import fake_filesystem_shutil from pyfakefs import fake_legacy_modules from pyfakefs import fake_pathlib from pyfakefs import mox3_stubout from pyfakefs.fake_filesystem import ( set_uid, set_gid, reset_ids, PatchMode, FakeFilesystem, ) from pyfakefs.fake_os import use_original_os from pyfakefs.helpers import IS_PYPY, IS_WIN from pyfakefs.legacy_packages import pathlib2, scandir from pyfakefs.mox3_stubout import StubOutForTesting OS_MODULE = "nt" if sys.platform == "win32" else "posix" PATH_MODULE = "ntpath" if sys.platform == "win32" else "posixpath" class TempfilePatcher: """Handles tempfile patching for Posix systems.""" def __init__(self): self.tempfile_cleanup = None self.tempfile_rmtree = None def start_patching(self): if self.tempfile_cleanup is not None: return if sys.version_info >= (3, 12): def cleanup(self_, windows=(os.name == "nt"), unlink=None): self.tempfile_cleanup(self_, windows, unlink or os.unlink) self.tempfile_cleanup = tempfile._TemporaryFileCloser.cleanup # type: ignore[module-attr] tempfile._TemporaryFileCloser.cleanup = cleanup # type: ignore[module-attr] if sys.version_info >= (3, 13) and hasattr(tempfile, "_rmtree"): # Debian patches tempfile by importing or copying shutil.rmtree as _rmtree # we patch this to use the original (patched) version def _rmtree(args, kwargs): return tempfile._shutil.rmtree(args, *kwargs) self.tempfile_rmtree = tempfile._rmtree # type: ignore[module-attr] tempfile._rmtree = _rmtree # type: ignore[module-attr] elif sys.platform != "win32": def close(self_, unlink=None): self.tempfile_cleanup(self_, unlink or os.unlink) self.tempfile_cleanup = tempfile._TemporaryFileCloser.close # type: ignore[module-attr] tempfile._TemporaryFileCloser.close = close # type: ignore[module-attr] def stop_patching(self): if self.tempfile_cleanup is None: return if sys.version_info < (3, 12): tempfile._TemporaryFileCloser.close = self.tempfile_cleanup # type: ignore[module-attr] else: tempfile._TemporaryFileCloser.cleanup = self.tempfile_cleanup # type: ignore[module-attr] self.tempfile_cleanup = None if self.tempfile_rmtree is not None: tempfile._rmtree = self.tempfile_rmtree # type: ignore[module-attr] self.tempfile_rmtree = None # reset the cached tempdir in tempfile tempfile.tempdir = None class LineCachePatcher: """Handles linecache patching for newer Python versions.""" def __init__(self): self.linecache_updatecache = None self.linecache_checkcache = None def start_patching(self): if self.linecache_updatecache is not None: return def checkcache(filename=None): """Calls the original linecache.checkcache making sure no fake OS calls are used.""" with use_original_os(): return self.linecache_checkcache(filename) def updatecache(filename, module_globals=None): """Calls the original linecache.updatecache making sure no fake OS calls are used.""" with use_original_os(): # workaround for updatecache problem with pytest under Windows, see #1096 if not IS_WIN or not filename.lower().endswith( r"pytest.exe\__main__.py" ): return self.linecache_updatecache(filename, module_globals) return [] if ( sys.version_info >= (3, 13) or sys.platform == "win32" and sys.version_info >= (3, 12) ): # in linecache, 'os' is now imported locally, which involves the # dynamic patcher, therefore we patch the affected functions self.linecache_updatecache = linecache.updatecache linecache.updatecache = updatecache self.linecache_checkcache = linecache.checkcache linecache.checkcache = checkcache def stop_patching(self): if self.linecache_updatecache is None: return linecache.updatecache = self.linecache_updatecache linecache.checkcache = self.linecache_checkcache self.linecache_updatecache = None self.linecache_checkcache = None def patchfs( _func: Optional[Callable] = None, , additional_skip_names: Optional[List[Union[str, ModuleType]]] = None, modules_to_reload: Optional[List[ModuleType]] = None, modules_to_patch: Optional[Dict[str, ModuleType]] = None, allow_root_user: bool = True, use_known_patches: bool = True, patch_open_code: PatchMode = PatchMode.OFF, patch_default_args: bool = False, use_cache: bool = True, use_dynamic_patch: bool = True, ) -> Callable: """Convenience decorator to use patcher with additional parameters in a test function. Usage:: @patchfs def test_my_function(fake_fs): fake_fs.create_file('foo') @patchfs(allow_root_user=False) def test_with_patcher_args(fs): os.makedirs('foo/bar') """ def wrap_patchfs(f: Callable) -> Callable: @functools.wraps(f) def wrapped(args, kwargs): with Patcher( additional_skip_names=additional_skip_names, modules_to_reload=modules_to_reload, modules_to_patch=modules_to_patch, allow_root_user=allow_root_user, use_known_patches=use_known_patches, patch_open_code=patch_open_code, patch_default_args=patch_default_args, use_cache=use_cache, use_dynamic_patch=use_dynamic_patch, ) as p: args = list(args) args.append(p.fs) return f(args, *kwargs) return wrapped if _func: if not callable(_func): raise TypeError( "Decorator argument is not a function.\n" "Did you mean `@patchfs(additional_skip_names=...)`?" ) if hasattr(_func, "patchings"): _func.nr_patches = len(_func.patchings) # type: ignore return wrap_patchfs(_func) return wrap_patchfs DOCTEST_PATCHER = None def load_doctests( loader: Any, tests: TestSuite, ignore: Any, module: ModuleType, additional_skip_names: Optional[List[Union[str, ModuleType]]] = None, modules_to_reload: Optional[List[ModuleType]] = None, modules_to_patch: Optional[Dict[str, ModuleType]] = None, allow_root_user: bool = True, use_known_patches: bool = True, patch_open_code: PatchMode = PatchMode.OFF, patch_default_args: bool = False, use_dynamic_patch: bool = True, ) -> TestSuite: # pylint:disable=unused-argument """Load the doctest tests for the specified module into unittest. Args: loader, tests, ignore : arguments passed in from `load_tests()` module: module under test remaining args: see :py:class:`TestCase` for an explanation File `example_test.py` in the pyfakefs release provides a usage example. """ has_patcher = Patcher.DOC_PATCHER is not None if not has_patcher: Patcher.DOC_PATCHER = Patcher( additional_skip_names=additional_skip_names, modules_to_reload=modules_to_reload, modules_to_patch=modules_to_patch, allow_root_user=allow_root_user, use_known_patches=use_known_patches, patch_open_code=patch_open_code, patch_default_args=patch_default_args, use_dynamic_patch=use_dynamic_patch, is_doc_test=True, ) assert Patcher.DOC_PATCHER is not None globs = Patcher.DOC_PATCHER.replace_globs(vars(module)) tests.addTests( doctest.DocTestSuite( module, globs=globs, setUp=Patcher.DOC_PATCHER.setUp, tearDown=Patcher.DOC_PATCHER.tearDown, ) ) return tests class TestCaseMixin: """Test case mixin that automatically replaces file-system related modules by fake implementations. Attributes: additional_skip_names: names of modules where no module replacement shall be performed, in addition to the names in :py:attr:`fake_filesystem_unittest.Patcher.SKIPNAMES`. Instead of the module names, the modules themselves may be used. modules_to_reload: A list of modules that need to be reloaded to be patched dynamically; may be needed if the module imports file system modules under an alias .. caution:: Reloading modules may have unwanted side effects. modules_to_patch: A dictionary of fake modules mapped to the fully qualified patched module names. Can be used to add patching of modules not provided by `pyfakefs`. If you specify some of these attributes here, and you have DocTests, consider also specifying the same arguments to :py:func:`load_doctests`. Example usage in derived test classes:: from unittest import TestCase from fake_filesystem_unittest import TestCaseMixin class MyTestCase(TestCase, TestCaseMixin): def __init__(self, methodName='runTest'): super(MyTestCase, self).__init__( methodName=methodName, additional_skip_names=['posixpath']) import sut class AnotherTestCase(TestCase, TestCaseMixin): def __init__(self, methodName='runTest'): super(MyTestCase, self).__init__( methodName=methodName, modules_to_reload=[sut]) """ additional_skip_names: Optional[List[Union[str, ModuleType]]] = None modules_to_reload: Optional[List[ModuleType]] = None modules_to_patch: Optional[Dict[str, ModuleType]] = None @property def patcher(self): if hasattr(self, "_patcher"): return self._patcher or Patcher.PATCHER return Patcher.PATCHER @property def fs(self) -> FakeFilesystem: return cast(FakeFilesystem, self.patcher.fs) def setUpPyfakefs( self, additional_skip_names: Optional[List[Union[str, ModuleType]]] = None, modules_to_reload: Optional[List[ModuleType]] = None, modules_to_patch: Optional[Dict[str, ModuleType]] = None, allow_root_user: bool = True, use_known_patches: bool = True, patch_open_code: PatchMode = PatchMode.OFF, patch_default_args: bool = False, use_cache: bool = True, use_dynamic_patch: bool = True, ) -> None: """Bind the file-related modules to the :py:class:`pyfakefs` fake file system instead of the real file system. Also bind the fake `open()` function. Invoke this at the beginning of the `setUp()` method in your unit test class. For the arguments, see the `TestCaseMixin` attribute description. If any of the arguments is not `None`, it overwrites the settings for the current test case. Settings the arguments here may be a more convenient way to adapt the setting than overwriting `__init__()`. """ # if the class has already a patcher setup, we use this one if Patcher.PATCHER is not None: return if additional_skip_names is None: additional_skip_names = self.additional_skip_names if modules_to_reload is None: modules_to_reload = self.modules_to_reload if modules_to_patch is None: modules_to_patch = self.modules_to_patch self._patcher = Patcher( additional_skip_names=additional_skip_names, modules_to_reload=modules_to_reload, modules_to_patch=modules_to_patch, allow_root_user=allow_root_user, use_known_patches=use_known_patches, patch_open_code=patch_open_code, patch_default_args=patch_default_args, use_cache=use_cache, use_dynamic_patch=use_dynamic_patch, ) self._patcher.setUp() cast(TestCase, self).addCleanup(self._patcher.tearDown) @classmethod def setUpClassPyfakefs( cls, additional_skip_names: Optional[List[Union[str, ModuleType]]] = None, modules_to_reload: Optional[List[ModuleType]] = None, modules_to_patch: Optional[Dict[str, ModuleType]] = None, allow_root_user: bool = True, use_known_patches: bool = True, patch_open_code: PatchMode = PatchMode.OFF, patch_default_args: bool = False, use_cache: bool = True, use_dynamic_patch: bool = True, ) -> None: """Similar to :py:func:`setUpPyfakefs`, but as a class method that can be used in `setUpClass` instead of in `setUp`. The fake filesystem will live in all test methods in the test class and can be used in the usual way. Note that using both :py:func:`setUpClassPyfakefs` and :py:func:`setUpPyfakefs` in the same class will not work correctly. .. note:: This method is only available from Python 3.8 onwards. .. note:: If using `pytest` as testrunner, you need at least pytest 6.2 for this method to work. """ if sys.version_info < (3, 8): raise NotImplementedError( "setUpClassPyfakefs is only available in " "Python versions starting from 3.8" ) # if the class has already a patcher setup, we use this one if Patcher.PATCHER is not None: return if additional_skip_names is None: additional_skip_names = cls.additional_skip_names if modules_to_reload is None: modules_to_reload = cls.modules_to_reload if modules_to_patch is None: modules_to_patch = cls.modules_to_patch Patcher.PATCHER = Patcher( additional_skip_names=additional_skip_names, modules_to_reload=modules_to_reload, modules_to_patch=modules_to_patch, allow_root_user=allow_root_user, use_known_patches=use_known_patches, patch_open_code=patch_open_code, patch_default_args=patch_default_args, use_cache=use_cache, use_dynamic_patch=use_dynamic_patch, ) Patcher.PATCHER.setUp() cast(TestCase, cls).addClassCleanup(Patcher.PATCHER.tearDown) @classmethod def fake_fs(cls): """Convenience class method for accessing the fake filesystem. For use inside `setUpClass`, after :py:func:`setUpClassPyfakefs` has been called. """ if Patcher.PATCHER: return Patcher.PATCHER.fs return None def pause(self) -> None: """Pause the patching of the file system modules until `resume` is called. After that call, all file system calls are executed in the real file system. Calling pause() twice is silently ignored. """ self.patcher.pause() def resume(self) -> None: """Resume the patching of the file system modules if `pause` has been called before. After that call, all file system calls are executed in the fake file system. Does nothing if patching is not paused. """ self.patcher.resume() class TestCase(unittest.TestCase, TestCaseMixin): """Test case class that automatically replaces file-system related modules by fake implementations. Inherits :py:class:`TestCaseMixin`. The arguments are explained in :py:class:`TestCaseMixin`. """ def __init__( self, methodName: str = "runTest", additional_skip_names: Optional[List[Union[str, ModuleType]]] = None, modules_to_reload: Optional[List[ModuleType]] = None, modules_to_patch: Optional[Dict[str, ModuleType]] = None, ): """Creates the test class instance and the patcher used to stub out file system related modules. Args: methodName: The name of the test method (same as in unittest.TestCase) """ super().__init__(methodName) self.additional_skip_names = additional_skip_names self.modules_to_reload = modules_to_reload self.modules_to_patch = modules_to_patch def tearDownPyfakefs(self) -> None: """This method is deprecated and exists only for backward compatibility. It does nothing. """ class Patcher: """ Instantiate a stub creator to bind and un-bind the file-related modules to the :py:mod:`pyfakefs` fake modules. The arguments are explained in :py:class:`TestCaseMixin`. :py:class:`Patcher` is used in :py:class:`TestCaseMixin`. :py:class:`Patcher` also works as a context manager for other tests:: with Patcher(): doStuff() """ """Stub nothing that is imported within these modules. `sys` is included to prevent `sys.path` from being stubbed with the fake `os.path`. The `linecache` module is used to read the test file in case of test failure to get traceback information before test tear down. In order to make sure that reading the test file is not faked, we skip faking the module. We also have to set back the cached open function in tokenize. """ SKIPMODULES = { None, fake_filesystem, fake_filesystem_shutil, fake_os, fake_io, fake_open, fake_path, fake_file, sys, linecache, tokenize, os, io, _io, genericpath, os.path, } if sys.platform == "win32": import nt # type:ignore[import] import ntpath SKIPMODULES.add(nt) SKIPMODULES.add(ntpath) else: import posix import posixpath import fcntl SKIPMODULES.add(posix) SKIPMODULES.add(posixpath) SKIPMODULES.add(fcntl) # a list of modules detected at run-time # each tool defines one or more module name prefixes for modules to be skipped RUNTIME_SKIPMODULES = { "pydevd": ["_pydevd_", "pydevd", "_pydev_"], # Python debugger (PyCharm/VSCode) "_jb_runner_tools": ["_jb_"], # JetBrains tools } # caches all modules that do not have file system modules or function # to speed up _find_modules CACHED_MODULES: Set[ModuleType] = set() FS_MODULES: Dict[str, Set[Tuple[ModuleType, str]]] = {} FS_FUNCTIONS: Dict[Tuple[str, str, str], Set[ModuleType]] = {} FS_DEFARGS: List[Tuple[FunctionType, int, Callable[..., Any]]] = [] SKIPPED_FS_MODULES: Dict[str, Set[Tuple[ModuleType, str]]] = {} assert None in SKIPMODULES, "sys.modules contains 'None' values; must skip them." IS_WINDOWS = sys.platform in ("win32", "cygwin") SKIPNAMES: Set[str] = set() # hold values from last call - if changed, the cache has to be invalidated PATCHED_MODULE_NAMES: Set[str] = set() ADDITIONAL_SKIP_NAMES: Set[str] = set() PATCH_DEFAULT_ARGS = False PATCHER: Optional["Patcher"] = None DOC_PATCHER: Optional["Patcher"] = None REF_COUNT = 0 DOC_REF_COUNT = 0 def __new__(cls, args, *kwargs): if kwargs.get("is_doc_test", False): if cls.DOC_PATCHER is None: cls.DOC_PATCHER = super().__new__(cls) return cls.DOC_PATCHER if cls.PATCHER is None: cls.PATCHER = super().__new__(cls) return cls.PATCHER def __init__( self, additional_skip_names: Optional[List[Union[str, ModuleType]]] = None, modules_to_reload: Optional[List[ModuleType]] = None, modules_to_patch: Optional[Dict[str, ModuleType]] = None, allow_root_user: bool = True, use_known_patches: bool = True, patch_open_code: PatchMode = PatchMode.OFF, patch_default_args: bool = False, use_cache: bool = True, use_dynamic_patch: bool = True, is_doc_test: bool = False, ) -> None: """ Args: additional_skip_names: names of modules where no module replacement shall be performed, in addition to the names in :py:attr:`fake_filesystem_unittest.Patcher.SKIPNAMES`. Instead of the module names, the modules themselves may be used. modules_to_reload: A list of modules that need to be reloaded to be patched dynamically; may be needed if the module imports file system modules under an alias .. caution:: Reloading modules may have unwanted side effects. modules_to_patch: A dictionary of fake modules mapped to the fully qualified patched module names. Can be used to add patching of modules not provided by `pyfakefs`. allow_root_user: If `True` (default), if the test is run as root user, the user in the fake file system is also considered a root user, otherwise it is always considered a regular user. use_known_patches: If `True` (the default), some patches for commonly used packages are applied which make them usable with pyfakefs. patch_open_code: If `True`, `io.open_code` is patched. The default is not to patch it, as it mostly is used to load compiled modules that are not in the fake file system. patch_default_args: If `True`, default arguments are checked for file system functions, which are patched. This check is expansive, so it is off by default. use_cache: If `True` (default), patched and non-patched modules are cached between tests for performance reasons. As this is a new feature, this argument allows to turn it off in case it causes any problems. use_dynamic_patch: If `True`, dynamic patching after setup is used (for example for modules loaded locally inside of functions). Can be switched off if it causes unwanted side effects. """ self.is_doc_test = is_doc_test if is_doc_test: if self.DOC_REF_COUNT > 0: return elif self.REF_COUNT > 0: if ( additional_skip_names or modules_to_reload or modules_to_patch or not allow_root_user or not use_known_patches or patch_open_code != PatchMode.OFF or patch_default_args or not use_cache or not use_dynamic_patch ): warnings.warn( "Nested fake filesystem invocation using custom arguments - " "using the existing fake filesystem, discarding custom arguments!" ) return if not allow_root_user: # set non-root IDs even if the real user is root set_uid(1) set_gid(1) self.skip_names = self.SKIPNAMES.copy() # save the original open function for use in pytest plugin self.original_open = open self.patch_open_code = patch_open_code self.tempfile_patcher = TempfilePatcher() self.linecache_patcher = LineCachePatcher() if additional_skip_names is not None: skip_names = [ cast(ModuleType, m).__name__ if inspect.ismodule(m) else cast(str, m) for m in additional_skip_names ] self.skip_names.update(skip_names) self._fake_module_classes: Dict[str, Any] = {} self._unfaked_module_classes: Dict[str, Any] = {} self._class_modules: Dict[str, List[str]] = {} self._init_fake_module_classes() # reload tempfile under posix to patch default argument self.modules_to_reload: List[ModuleType] = [] if modules_to_reload is not None: self.modules_to_reload.extend(modules_to_reload) self.patch_default_args = patch_default_args self.use_cache = use_cache self.use_dynamic_patch = use_dynamic_patch self.cleanup_handlers: Dict[str, Callable[[str], bool]] = {} # Attributes set by _refresh() self._stubs: Optional[StubOutForTesting] = None self.fs: Optional[FakeFilesystem] = None self.fake_modules: Dict[str, Any] = {} self.unfaked_modules: Dict[str, Any] = {} # _isStale is set by tearDown(), reset by _refresh() self._isStale = True self._dyn_patcher: Optional[DynamicPatcher] = None self._patching = False self._paused = False self.has_copy_file_range = False self.has_copy_file = False if use_known_patches: from pyfakefs.patched_packages import ( get_modules_to_patch, get_classes_to_patch, get_fake_module_classes, get_cleanup_handlers, ) modules_to_patch = modules_to_patch or {} modules_to_patch.update(get_modules_to_patch()) self._class_modules.update(get_classes_to_patch()) self._fake_module_classes.update(get_fake_module_classes()) self.cleanup_handlers.update(get_cleanup_handlers()) if modules_to_patch is not None: for name, fake_module in modules_to_patch.items(): self._fake_module_classes[name] = fake_module patched_module_names = set(modules_to_patch) else: patched_module_names = set() clear_cache = not use_cache if use_cache: if patched_module_names != self.PATCHED_MODULE_NAMES: self.__class__.PATCHED_MODULE_NAMES = patched_module_names clear_cache = True if self.skip_names != self.ADDITIONAL_SKIP_NAMES: self.__class__.ADDITIONAL_SKIP_NAMES = self.skip_names clear_cache = True if patch_default_args != self.PATCH_DEFAULT_ARGS: self.__class__.PATCH_DEFAULT_ARGS = patch_default_args clear_cache = True if clear_cache: self.clear_cache() self._fake_module_functions: Dict[str, Dict] = {} self._init_fake_module_functions() @classmethod def clear_fs_cache(cls) -> None: """Clear the module cache.""" cls.CACHED_MODULES = set() cls.FS_MODULES = {} cls.FS_FUNCTIONS = {} cls.FS_DEFARGS = [] cls.SKIPPED_FS_MODULES = {} def clear_cache(self) -> None: """Clear the module cache (convenience instance method).""" self.__class__.clear_fs_cache() def register_cleanup_handler(self, name: str, handler: Callable[[str], bool]): """Register a handler for cleaning up a module after it had been loaded by the dynamic patcher. This allows to handle modules that cannot be reloaded without unwanted side effects. Args: name: The fully qualified module name. handler: A callable that may do any module cleanup, or do nothing and return `True` in case reloading shall be prevented. Returns: `True` if no further cleanup/reload shall occur after the handler is executed, `False` if the cleanup/reload shall still happen. """ self.cleanup_handlers[name] = handler def _init_fake_module_classes(self) -> None: # IMPORTANT TESTING NOTE: Whenever you add a new module below, test # it by adding an attribute in fixtures/module_with_attributes.py # and a test in fake_filesystem_unittest_test.py, class # TestAttributesWithFakeModuleNames. # we instantiate the fake pathlib library with `from_patcher` set # to avoid faking pathlib.os (already faked by the patcher) def fake_pathlib_module(fs: FakeFilesystem): return fake_pathlib.FakePathlibModule(fs, from_patcher=True) def fake_path_module(fs: FakeFilesystem): return fake_pathlib.FakePathlibPathModule(fs, from_patcher=True) self._fake_module_classes = { "os": fake_os.FakeOsModule, "shutil": fake_filesystem_shutil.FakeShutilModule, "io": fake_io.FakeIoModule, "pathlib": fake_pathlib_module, } if sys.version_info[:2] == (3, 13): # for Python 3.13, we need both pathlib (path with __init__.py) and # pathlib._local (has the actual implementation); # depending on how pathlib is imported, either may be used self._fake_module_classes["pathlib._local"] = fake_pathlib_module if IS_PYPY or sys.version_info >= (3, 12): # in PyPy and later cpython versions, the module is referenced as _io self._fake_module_classes["_io"] = fake_io.FakeIoModule2 if sys.platform == "win32": self._fake_module_classes["nt"] = fake_path.FakeNtModule else: self._fake_module_classes["fcntl"] = fake_filesystem.FakeFcntlModule # class modules maps class names against a list of modules they can # be contained in - this allows for alternative modules like # `pathlib` and `pathlib2` self._class_modules["Path"] = ["pathlib"] if sys.version_info[:2] == (3, 13): self._class_modules["Path"].append("pathlib._local") self._unfaked_module_classes["pathlib"] = fake_pathlib.RealPathlibModule if sys.version_info[:2] == (3, 13): self._unfaked_module_classes["pathlib._local"] = ( fake_pathlib.RealPathlibModule ) if pathlib2: self._fake_module_classes["pathlib2"] = ( fake_legacy_modules.FakePathlib2Module ) self._class_modules["Path"].append("pathlib2") self._unfaked_module_classes["pathlib2"] = fake_pathlib.RealPathlibModule if scandir: self._fake_module_classes["scandir"] = fake_legacy_modules.FakeScanDirModule self._fake_module_classes["Path"] = fake_path_module self._unfaked_module_classes["Path"] = fake_pathlib.RealPathlibPathModule def _init_fake_module_functions(self) -> None: # handle patching function imported separately like # `from os import stat` # each patched function name has to be looked up separately for mod_name, fake_module in self._fake_module_classes.items(): if hasattr(fake_module, "dir"): module_dir = fake_module.dir if inspect.isfunction(module_dir): for fct_name in fake_module.dir(): module_attr = (getattr(fake_module, fct_name), mod_name) self._fake_module_functions.setdefault(fct_name, {})[ mod_name ] = module_attr if mod_name == "os": self._fake_module_functions.setdefault(fct_name, {})[ OS_MODULE ] = module_attr # special handling for functions in os.path fake_module = fake_filesystem.FakePathModule for fct_name in fake_module.dir(): module_attr = (getattr(fake_module, fct_name), PATH_MODULE) self._fake_module_functions.setdefault(fct_name, {})["genericpath"] = ( module_attr ) self._fake_module_functions.setdefault(fct_name, {})[PATH_MODULE] = ( module_attr ) def __enter__(self) -> "Patcher": """Context manager for usage outside of fake_filesystem_unittest.TestCase. Ensure that all patched modules are removed in case of an unhandled exception. """ self.setUp() return self def __exit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: self.tearDown() def _is_fs_module( self, mod: ModuleType, name: str, module_names: List[str] ) -> bool: try: return ( inspect.ismodule(mod) and mod.__name__ in module_names or inspect.isclass(mod) and mod.__module__ in self._class_modules.get(name, []) ) except Exception: # handle cases where the module has no __name__ or __module__ # attribute - see #460, and any other exception triggered # by inspect functions return False def _is_fs_function(self, fct: FunctionType) -> bool: try: return ( (inspect.isfunction(fct) or inspect.isbuiltin(fct)) and fct.__name__ in self._fake_module_functions and fct.__module__ in self._fake_module_functions[fct.__name__] ) except Exception: # handle cases where the function has no __name__ or __module__ # attribute, or any other exception in inspect functions return False def _def_values( self, item: FunctionType ) -> Iterator[Tuple[FunctionType, int, Any]]: """Find default arguments that are file-system functions to be patched in top-level functions and members of top-level classes.""" # check for module-level functions try: if item.__defaults__ and inspect.isfunction(item): for i, d in enumerate(item.__defaults__): if self._is_fs_function(d): yield item, i, d except Exception: pass try: if inspect.isclass(item): # check for methods in class # (nested classes are ignored for now) # inspect.getmembers is very expansive! for m in inspect.getmembers(item, predicate=inspect.isfunction): f = cast(FunctionType, m[1]) if f.__defaults__: for i, d in enumerate(f.__defaults__): if self._is_fs_function(d): yield f, i, d except Exception: # Ignore any exception, examples: # ImportError: No module named '_gdbm' # _DontDoThat() (see #523) pass def _find_def_values(self, module_items: ItemsView[str, FunctionType]) -> None: for _, fct in module_items: for f, i, d in self._def_values(fct): self.__class__.FS_DEFARGS.append((f, i, d)) def _find_modules(self) -> None: """Find and cache all modules that import file system modules. Later, `setUp()` will stub these with the fake file system modules. """ module_names = list(self._fake_module_classes.keys()) + [PATH_MODULE] items = sys.modules.copy().items() for name, module in items: try: if ( self.use_cache and module in self.CACHED_MODULES or not inspect.ismodule(module) ): continue except Exception: # workaround for some py (part of pytest) versions # where py.error has no __name__ attribute # see https://github.com/pytest-dev/py/issues/73 # and any other exception triggered by inspect.ismodule if self.use_cache: try: self.__class__.CACHED_MODULES.add(module) except TypeError: # unhashable module - don't cache it pass continue skipped = module in self.SKIPMODULES or any( [sn.startswith(module.__name__) for sn in self.skip_names] ) module_items = module.__dict__.copy().items() modules = { name: mod for name, mod in module_items if self._is_fs_module(mod, name, module_names) } if skipped: for name, mod in modules.items(): self.__class__.SKIPPED_FS_MODULES.setdefault(name, set()).add( (module, mod.__name__) ) else: for name, mod in modules.items(): self.__class__.FS_MODULES.setdefault(name, set()).add( (module, mod.__name__) ) functions = { name: fct for name, fct in module_items if self._is_fs_function(fct) } for name, fct in functions.items(): self.__class__.FS_FUNCTIONS.setdefault( (name, fct.__name__, fct.__module__), set() ).add(module) # find default arguments that are file system functions if self.patch_default_args: self._find_def_values(module_items) if self.use_cache: self.__class__.CACHED_MODULES.add(module) def _refresh(self) -> None: """Renew the fake file system and set the _isStale flag to `False`.""" if self._stubs is not None: self._stubs.smart_unset_all() self._stubs = mox3_stubout.StubOutForTesting() self.fs = fake_filesystem.FakeFilesystem(patcher=self, create_temp_dir=True) self.fs.patch_open_code = self.patch_open_code for name in self._fake_module_classes: self.fake_modules[name] = self._fake_module_classes[name](self.fs) if hasattr(self.fake_modules[name], "skip_names"): self.fake_modules[name].skip_names = self.skip_names self.fake_modules[PATH_MODULE] = self.fake_modules["os"].path for name in self._unfaked_module_classes: self.unfaked_modules[name] = self._unfaked_module_classes[name]() self._isStale = False def setUp(self, doctester: Any = None) -> None: """Bind the file-related modules to the :py:mod:`pyfakefs` fake modules real ones. Also bind the fake `file()` and `open()` functions. """ if self.is_doc_test: self.__class__.DOC_REF_COUNT += 1 if self.__class__.DOC_REF_COUNT > 1: return else: self.__class__.REF_COUNT += 1 if self.__class__.REF_COUNT > 1: return with warnings.catch_warnings(): # ignore warnings, see #542 and #614 warnings.filterwarnings("ignore") self._find_modules() self._refresh() if doctester is not None: doctester.globs = self.replace_globs(doctester.globs) self.start_patching() linecache.open = self.original_open # type: ignore[attr-defined] tokenize._builtin_open = self.original_open # type: ignore def start_patching(self) -> None: if not self._patching: self._patching = True self._paused = False self.linecache_patcher.start_patching() self.patch_modules() self.patch_functions() self.patch_defaults() self._set_glob_os_functions() self.tempfile_patcher.start_patching() self._dyn_patcher = DynamicPatcher(self) sys.meta_path.insert(0, self._dyn_patcher) for module in self.modules_to_reload: if sys.modules.get(module.__name__) is module: reload(module) if not self.use_dynamic_patch: self._dyn_patcher.cleanup() sys.meta_path.pop(0) def _set_glob_os_functions(self): # make sure the os functions cached in glob are patched if sys.version_info >= (3, 13): globber = glob._StringGlobber # type: ignore[module-attr] globber.lstat = staticmethod(os.lstat) if sys.version_info < (3, 14): globber.scandir = staticmethod(os.scandir) if sys.version_info >= (3, 14): globber.lexists = staticmethod(os.path.lexists) def patch_functions(self) -> None: assert self._stubs is not None for (name, ft_name, ft_mod), modules in self.FS_FUNCTIONS.items(): method, mod_name = self._fake_module_functions[ft_name][ft_mod] fake_module = self.fake_modules[mod_name] attr = method.__get__( fake_module, fake_module.__class__ ) # pytype: disable=attribute-error for module in modules: self._stubs.smart_set(module, name, attr) def patch_modules(self) -> None: skip_prefix_list = [] for rt_skip_module, prefixes in self.RUNTIME_SKIPMODULES.items(): if rt_skip_module in sys.modules: skip_prefix_list.extend(prefixes) skip_prefixes = tuple(skip_prefix_list) assert self._stubs is not None for name, modules in self.FS_MODULES.items(): for module, attr in modules: try: if not skip_prefixes or not module.__name__.startswith( skip_prefixes ): self._stubs.smart_set(module, name, self.fake_modules[attr]) elif attr in self.unfaked_modules: self._stubs.smart_set(module, name, self.unfaked_modules[attr]) except Exception: # handle the rare case that a module has no __name__ pass for name, modules in self.SKIPPED_FS_MODULES.items(): for module, attr in modules: if attr in self.unfaked_modules: self._stubs.smart_set(module, name, self.unfaked_modules[attr]) def patch_defaults(self) -> None: for fct, idx, ft in self.FS_DEFARGS: method, mod_name = self._fake_module_functions[ft.__name__][ft.__module__] fake_module = self.fake_modules[mod_name] attr = method.__get__( fake_module, fake_module.__class__ ) # pytype: disable=attribute-error new_defaults = [] assert fct.__defaults__ is not None for i, d in enumerate(fct.__defaults__): if i == idx: new_defaults.append(attr) else: new_defaults.append(d) fct.__defaults__ = tuple(new_defaults) def replace_globs(self, globs_: Dict[str, Any]) -> Dict[str, Any]: globs = globs_.copy() if self._isStale: self._refresh() for name in self._fake_module_classes: if name in globs: globs[name] = self._fake_module_classes[name](self.fs) return globs def tearDown(self, doctester: Any = None): """Clear the fake filesystem bindings created by `setUp()`.""" if self.is_doc_test: self.__class__.DOC_REF_COUNT -= 1 if self.__class__.DOC_REF_COUNT > 0: return else: self.__class__.REF_COUNT -= 1 if self.__class__.REF_COUNT > 0: return self.stop_patching() reset_ids() if self.is_doc_test: self.__class__.DOC_PATCHER = None else: self.__class__.PATCHER = None def stop_patching(self, temporary=False) -> None: if self._patching: self._isStale = True self._patching = False self._paused = temporary if self._stubs: self._stubs.smart_unset_all() self.unset_defaults() if self.use_dynamic_patch and self._dyn_patcher: self._dyn_patcher.cleanup() sys.meta_path.pop(0) self.tempfile_patcher.stop_patching() self.linecache_patcher.stop_patching() self._set_glob_os_functions() @property def is_patching(self): return self._patching def unset_defaults(self) -> None: for fct, idx, ft in self.FS_DEFARGS: new_defaults = [] for i, d in enumerate(cast(Tuple, fct.__defaults__)): if i == idx: new_defaults.append(ft) else: new_defaults.append(d) fct.__defaults__ = tuple(new_defaults) def pause(self) -> None: """Pause the patching of the file system modules until `resume` is called. After that call, all file system calls are executed in the real file system. Calling pause() twice is silently ignored. """ self.stop_patching(temporary=True) def resume(self) -> None: """Resume the patching of the file system modules if `pause` has been called before. After that call, all file system calls are executed in the fake file system. Does nothing if patching is not paused. """ if self._paused: self.start_patching() class Pause: """Simple context manager that allows to pause/resume patching the filesystem. Patching is paused in the context manager, and resumed after going out of its scope. """ def __init__(self, caller: Union[Patcher, TestCaseMixin, FakeFilesystem]): """Initializes the context manager with the fake filesystem. Args: caller: either the FakeFilesystem instance, the Patcher instance or the pyfakefs test case. """ if isinstance(caller, (Patcher, TestCaseMixin)): assert caller.fs is not None self._fs: FakeFilesystem = caller.fs elif isinstance(caller, FakeFilesystem): self._fs = caller else: raise ValueError( "Invalid argument - should be of type " '"fake_filesystem_unittest.Patcher", ' '"fake_filesystem_unittest.TestCase" ' 'or "fake_filesystem.FakeFilesystem"' ) def __enter__(self) -> FakeFilesystem: self._fs.pause() return self._fs def __exit__(self, args: Any) -> None: self._fs.resume() class DynamicPatcher(MetaPathFinder, Loader): """A file loader that replaces file system related modules by their fake implementation if they are loaded after calling `setUpPyfakefs()`. Implements the protocol needed for import hooks. """ def __init__(self, patcher: Patcher) -> None: self._patcher = patcher self.sysmodules = {} self.modules = self._patcher.fake_modules self._loaded_module_names: Set[str] = set() self.cleanup_handlers = patcher.cleanup_handlers # remove all modules that have to be patched from `sys.modules`, # otherwise the find_... methods will not be called for name in self.modules: if self.needs_patch(name) and name in sys.modules: self.sysmodules[name] = sys.modules[name] del sys.modules[name] for name, module in self.modules.items(): sys.modules[name] = module def cleanup(self) -> None: for module_name in self.sysmodules: sys.modules[module_name] = self.sysmodules[module_name] for module in self._patcher.modules_to_reload: if module.__name__ in sys.modules: reload(module) reloaded_module_names = [ module.__name__ for module in self._patcher.modules_to_reload ] # Delete all modules loaded during the test, ensuring that # they are reloaded after the test. for name in self._loaded_module_names: if name in sys.modules and name not in reloaded_module_names: if name in self.cleanup_handlers and self.cleanup_handlers[name](name): continue del sys.modules[name] def needs_patch(self, name: str) -> bool: """Checks if the module with the given name shall be replaced.""" if name not in self.modules: self._loaded_module_names.add(name) return False if name in sys.modules and type(sys.modules[name]) is self.modules[name]: return False return True def fake_module_path(self, name: str) -> str: """Checks if the module with the given name is a module existing in the fake filesystem and returns its path in this case. """ fs = self._patcher.fs # we assume that the module name is the absolute module path if fs is not None: base_path = name.replace(".", fs.path_separator) for path in sys.path: module_path = fs.joinpaths(path, base_path) py_module_path = module_path + ".py" if fs.exists(py_module_path): return fs.absnormpath(py_module_path) init_path = fs.joinpaths(module_path, "__init__.py") if fs.exists(init_path): return fs.absnormpath(init_path) return "" def find_spec( self, fullname: str, path: Optional[Sequence[Union[bytes, str]]], target: Optional[ModuleType] = None, ) -> Optional[ModuleSpec]: """Module finder.""" if self.needs_patch(fullname): return ModuleSpec(fullname, self) if self._patcher.patch_open_code != PatchMode.OFF: # handle modules created in the fake filesystem module_path = self.fake_module_path(fullname) if module_path: spec = spec_from_file_location(fullname, module_path) if spec: module = module_from_spec(spec) sys.modules[fullname] = module return ModuleSpec(fullname, self) return None def load_module(self, fullname: str) -> ModuleType: """Replaces the module by its fake implementation.""" sys.modules[fullname] = self.modules[fullname] return self.modules[fullname]	52.17 KB	2026-01-20 12:58
📄 fake_io.py # Copyright 2009 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Uses :py:class:`FakeIoModule` to provide a fake ``io`` module replacement. """ import _io # pytype: disable=import-error import io import sys from enum import Enum from typing import ( List, Optional, Callable, Union, Any, AnyStr, IO, TYPE_CHECKING, ) from pyfakefs.fake_file import AnyFileWrapper from pyfakefs.fake_open import fake_open from pyfakefs.helpers import IS_PYPY, is_called_from_skipped_module if TYPE_CHECKING: from pyfakefs.fake_filesystem import FakeFilesystem class PatchMode(Enum): """Defines if patching shall be on, off, or in automatic mode. Currently only used for `patch_open_code` option. """ OFF = 1 AUTO = 2 ON = 3 class FakeIoModule: """Uses FakeFilesystem to provide a fake io module replacement. You need a fake_filesystem to use this:: filesystem = fake_filesystem.FakeFilesystem() my_io_module = fake_io.FakeIoModule(filesystem) """ @staticmethod def dir() -> List[str]: """Return the list of patched function names. Used for patching functions imported from the module. """ _dir = ["open"] if sys.version_info >= (3, 8): _dir.append("open_code") return _dir def __init__(self, filesystem: "FakeFilesystem"): """ Args: filesystem: FakeFilesystem used to provide file system information. """ self.filesystem = filesystem self.skip_names: List[str] = [] self._io_module = io def open( self, file: Union[AnyStr, int], mode: str = "r", buffering: int = -1, encoding: Optional[str] = None, errors: Optional[str] = None, newline: Optional[str] = None, closefd: bool = True, opener: Optional[Callable] = None, is_fake_open_code: bool = False, ) -> Union[AnyFileWrapper, IO[Any]]: """Redirect the call to FakeFileOpen. See FakeFileOpen.call() for description. """ return fake_open( self.filesystem, self.skip_names, file, mode, buffering, encoding, errors, newline, closefd, opener, is_fake_open_code, ) if sys.version_info >= (3, 8): def open_code(self, path): """Redirect the call to open. Note that the behavior of the real function may be overridden by an earlier call to the PyFile_SetOpenCodeHook(). This behavior is not reproduced here. """ if not isinstance(path, str) and not IS_PYPY: raise TypeError("open_code() argument 'path' must be str, not int") patch_mode = self.filesystem.patch_open_code if ( patch_mode == PatchMode.AUTO and self.filesystem.exists(path) or patch_mode == PatchMode.ON ): return self.open(path, mode="rb", is_fake_open_code=True) # mostly this is used for compiled code - # don't patch these, as the files are probably in the real fs return self._io_module.open_code(path) def __getattr__(self, name): """Forwards any unfaked calls to the standard io module.""" return getattr(self._io_module, name) class FakeIoModule2(FakeIoModule): """Similar to ``FakeIoModule``, but fakes `_io` instead of `io`.""" def __init__(self, filesystem: "FakeFilesystem"): """ Args: filesystem: FakeFilesystem used to provide file system information. """ super().__init__(filesystem) self._io_module = _io if sys.platform != "win32": import fcntl class FakeFcntlModule: """Replaces the fcntl module. Only valid under Linux/MacOS, currently just mocks the functionality away. """ @staticmethod def dir() -> List[str]: """Return the list of patched function names. Used for patching functions imported from the module. """ return ["fcntl", "ioctl", "flock", "lockf"] def __init__(self, filesystem: "FakeFilesystem"): """ Args: filesystem: FakeFilesystem used to provide file system information (currently not used). """ self.filesystem = filesystem self._fcntl_module = fcntl def fcntl(self, fd: int, cmd: int, arg: int = 0) -> Union[int, bytes]: return 0 if isinstance(arg, int) else arg def ioctl( self, fd: int, request: int, arg: int = 0, mutate_flag: bool = True ) -> Union[int, bytes]: return 0 if isinstance(arg, int) else arg def flock(self, fd: int, operation: int) -> None: pass def lockf( self, fd: int, cmd: int, len: int = 0, start: int = 0, whence=0 ) -> Any: pass def __getattribute__(self, name): """Prevents patching of skipped modules.""" fs: FakeFilesystem = object.__getattribute__(self, "filesystem") fnctl_module = object.__getattribute__(self, "_fcntl_module") if fs.patcher: if is_called_from_skipped_module( skip_names=fs.patcher.skip_names, case_sensitive=fs.is_case_sensitive, ): # remove the `self` argument for FakeOsModule methods return getattr(fnctl_module, name) return object.__getattribute__(self, name) def __getattr__(self, name): """Forwards any unfaked calls to the standard fcntl module.""" return getattr(self._fcntl_module, name)	6.31 KB	2026-01-20 12:58
📄 fake_legacy_modules.py # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import warnings from pyfakefs.fake_pathlib import FakePathlibModule from pyfakefs.fake_scandir import scandir, walk def legacy_warning(module_name): msg = ( f"You are using the legacy package '{module_name}' instead of the " f"built-in module." "Patching this package will no longer be supported in pyfakefs >= 6" ) warnings.warn(msg, category=DeprecationWarning) class FakePathlib2Module(FakePathlibModule): """Uses FakeFilesystem to provide a fake pathlib module replacement. for the `pathlib2` package available on PyPi. The usage of `pathlib2` is deprecated and will no longer be supported in future pyfakefs versions. """ has_warned = False def __getattribute__(self, name): attr = object.__getattribute__(self, name) if hasattr(attr, "__call__") and not FakePathlib2Module.has_warned: FakePathlib2Module.has_warned = True legacy_warning("pathlib2") return attr class FakeScanDirModule: """Uses FakeFilesystem to provide a fake module replacement for the `scandir` package available on PyPi. The usage of the `scandir` package is deprecated and will no longer be supported in future pyfakefs versions. You need a fake_filesystem to use this: `filesystem = fake_filesystem.FakeFilesystem()` `fake_scandir_module = fake_filesystem.FakeScanDirModule(filesystem)` """ @staticmethod def dir(): """Return the list of patched function names. Used for patching functions imported from the module. """ return "scandir", "walk" def __init__(self, filesystem): self.filesystem = filesystem has_warned = False def scandir(self, path="."): """Return an iterator of DirEntry objects corresponding to the entries in the directory given by path. Args: path: Path to the target directory within the fake filesystem. Returns: an iterator to an unsorted list of os.DirEntry objects for each entry in path. Raises: OSError: if the target is not a directory. """ if not self.has_warned: self.__class__.has_warned = True legacy_warning("scandir") return scandir(self.filesystem, path) def walk(self, top, topdown=True, onerror=None, followlinks=False): """Perform a walk operation over the fake filesystem. Args: top: The root directory from which to begin walk. topdown: Determines whether to return the tuples with the root as the first entry (`True`) or as the last, after all the child directory tuples (`False`). onerror: If not `None`, function which will be called to handle the `os.error` instance provided when `os.listdir()` fails. followlinks: If `True`, symbolic links are followed. Yields: (path, directories, nondirectories) for top and each of its subdirectories. See the documentation for the builtin os module for further details. """ if not self.has_warned: self.__class__.has_warned = True legacy_warning("scandir") return walk(self.filesystem, top, topdown, onerror, followlinks)	3.81 KB	2026-01-20 12:58
📄 fake_open.py # Copyright 2009 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """A fake open() function replacement. See ``fake_filesystem`` for usage.""" import errno import io import os import sys from stat import ( S_ISDIR, ) from typing import ( Optional, Union, Any, Tuple, cast, AnyStr, TYPE_CHECKING, Callable, IO, List, ) from pyfakefs import helpers from pyfakefs.fake_file import ( FakePipeWrapper, FakeFileWrapper, FakeFile, AnyFileWrapper, ) from pyfakefs.helpers import ( AnyString, is_called_from_skipped_module, is_root, PERM_READ, PERM_WRITE, _OpenModes, ) if TYPE_CHECKING: from pyfakefs.fake_filesystem import FakeFilesystem # Work around pyupgrade auto-rewriting `io.open()` to `open()`. io_open = io.open _OPEN_MODE_MAP = { # mode name:(file must exist, can read, can write, # truncate, append, must not exist) "r": (True, True, False, False, False, False), "w": (False, False, True, True, False, False), "a": (False, False, True, False, True, False), "r+": (True, True, True, False, False, False), "w+": (False, True, True, True, False, False), "a+": (False, True, True, False, True, False), "x": (False, False, True, False, False, True), "x+": (False, True, True, False, False, True), } def fake_open( filesystem: "FakeFilesystem", skip_names: List[str], file: Union[AnyStr, int], mode: str = "r", buffering: int = -1, encoding: Optional[str] = None, errors: Optional[str] = None, newline: Optional[str] = None, closefd: bool = True, opener: Optional[Callable] = None, is_fake_open_code: bool = False, ) -> Union[AnyFileWrapper, IO[Any]]: """Redirect the call to FakeFileOpen. See FakeFileOpen.call() for description. """ # We don't need to check this if we are in an `open_code` call # from a faked file (and this might cause recursions in `linecache`) if not is_fake_open_code and is_called_from_skipped_module( skip_names=skip_names, case_sensitive=filesystem.is_case_sensitive, check_open_code=sys.version_info >= (3, 12), ): return io_open( # pytype: disable=wrong-arg-count file, mode, buffering, encoding, errors, newline, closefd, opener, ) fake_file_open = FakeFileOpen(filesystem) return fake_file_open( file, mode, buffering, encoding, errors, newline, closefd, opener ) class FakeFileOpen: """Faked `file()` and `open()` function replacements. Returns :py:class:`FakeDirectory<pyfakefs.fake_file.FakeFile>` objects in a fake filesystem; replaces the `open()` function. """ __name__ = "FakeFileOpen" def __init__( self, filesystem: "FakeFilesystem", delete_on_close: bool = False, raw_io: bool = False, ): """ Args: filesystem: FakeFilesystem used to provide file system information delete_on_close: optional boolean, deletes file on close() """ self.filesystem = filesystem self._delete_on_close = delete_on_close self.raw_io = raw_io def __call__(self, args: Any, kwargs: Any) -> AnyFileWrapper: """Redirects calls to file() or open() to appropriate method.""" return self.call(args, *kwargs) def call( self, file_: Union[AnyStr, int], mode: str = "r", buffering: int = -1, encoding: Optional[str] = None, errors: Optional[str] = None, newline: Optional[str] = None, closefd: bool = True, opener: Any = None, open_modes: Optional[_OpenModes] = None, ) -> AnyFileWrapper: """Return a file-like object with the contents of the target file object. Args: file_: Path to target file or a file descriptor. mode: Additional file modes (all modes in `open()` are supported). buffering: the buffer size used for writing. Data will only be flushed if buffer size is exceeded. The default (-1) uses a system specific default buffer size. Text line mode (e.g. buffering=1 in text mode) is not supported. encoding: The encoding used to encode Unicode strings / decode bytes. errors: (str) Defines how encoding errors are handled. newline: Controls universal newlines, passed to stream object. closefd: If a file descriptor rather than file name is passed, and this is set to `False`, then the file descriptor is kept open when file is closed. opener: an optional function object that will be called with `file_` and the open flags (derived from `mode`) and returns a file descriptor. open_modes: Modes for opening files if called from low-level API. Returns: A file-like object containing the contents of the target file. Raises: OSError depending on Python version / call mode: - if the target object is a directory - on an invalid path - if the file does not exist when it should - if the file exists but should not - if permission is denied ValueError: for an invalid mode or mode combination """ binary = "b" in mode if binary and encoding: raise ValueError("binary mode doesn't take an encoding argument") newline, open_modes = self._handle_file_mode(mode, newline, open_modes) opened_as_fd = isinstance(file_, int) # the pathlib opener is defined in a Path instance that may not be # patched under some circumstances; as it just calls standard open(), # we may ignore it, as it would not change the behavior if opener is not None and opener.__module__ not in ( "pathlib", "pathlib._local", ): # opener shall return a file descriptor, which will be handled # here as if directly passed file_ = opener(file_, self._open_flags_from_open_modes(open_modes)) file_object, file_path, filedes, real_path, can_write = self._handle_file_arg( file_ ) if file_object is None and file_path is None: # file must be a fake pipe wrapper, find it... if ( filedes is None or len(self.filesystem.open_files) <= filedes or not self.filesystem.open_files[filedes] ): raise OSError(errno.EBADF, "invalid pipe file descriptor") wrappers = self.filesystem.open_files[filedes] assert wrappers is not None existing_wrapper = wrappers[0] assert isinstance(existing_wrapper, FakePipeWrapper) wrapper = FakePipeWrapper( self.filesystem, existing_wrapper.fd, existing_wrapper.can_write, mode, ) file_des = self.filesystem.add_open_file(wrapper) wrapper.filedes = file_des return wrapper assert file_path is not None if not filedes: closefd = True if ( not opener and open_modes.must_not_exist and ( file_object or self.filesystem.islink(file_path) and not self.filesystem.is_windows_fs ) ): self.filesystem.raise_os_error(errno.EEXIST, file_path) assert real_path is not None file_object = self._init_file_object( file_object, file_path, open_modes, real_path, check_file_permission=not opened_as_fd, ) if S_ISDIR(file_object.st_mode): if self.filesystem.is_windows_fs: self.filesystem.raise_os_error(errno.EACCES, file_path) else: self.filesystem.raise_os_error(errno.EISDIR, file_path) # If you print obj.name, the argument to open() must be printed. # Not the abspath, not the filename, but the actual argument. file_object.opened_as = file_path if open_modes.truncate: current_time = helpers.now() file_object.st_mtime = current_time if not self.filesystem.is_windows_fs: file_object.st_ctime = current_time fakefile = FakeFileWrapper( file_object, file_path, open_modes=open_modes, allow_update=open_modes.can_write and can_write, delete_on_close=self._delete_on_close, filesystem=self.filesystem, newline=newline, binary=binary, closefd=closefd, encoding=encoding, errors=errors, buffering=buffering, raw_io=self.raw_io, opened_as_fd=opened_as_fd, ) if filedes is not None: fakefile.filedes = filedes # replace the file wrapper open_files_list = self.filesystem.open_files[filedes] assert open_files_list is not None open_files_list.append(fakefile) else: fakefile.filedes = self.filesystem.add_open_file(fakefile) return fakefile @staticmethod def _open_flags_from_open_modes(open_modes: _OpenModes) -> int: flags = 0 if open_modes.can_read and open_modes.can_write: flags \|= os.O_RDWR elif open_modes.can_read: flags \|= os.O_RDONLY elif open_modes.can_write: flags \|= os.O_WRONLY if open_modes.append: flags \|= os.O_APPEND if open_modes.truncate: flags \|= os.O_TRUNC if not open_modes.must_exist and open_modes.can_write: flags \|= os.O_CREAT if open_modes.must_not_exist and open_modes.can_write: flags \|= os.O_EXCL return flags def _init_file_object( self, file_object: Optional[FakeFile], file_path: AnyStr, open_modes: _OpenModes, real_path: AnyString, check_file_permission: bool, ) -> FakeFile: if file_object: if ( check_file_permission and not is_root() and ( (open_modes.can_read and not file_object.has_permission(PERM_READ)) or ( open_modes.can_write and not file_object.has_permission(PERM_WRITE) ) ) ): self.filesystem.raise_os_error(errno.EACCES, file_path) if open_modes.can_write: if open_modes.truncate: file_object.set_contents("") else: if open_modes.must_exist: self.filesystem.raise_os_error(errno.ENOENT, file_path) if self.filesystem.islink(file_path): link_object = self.filesystem.resolve(file_path, follow_symlinks=False) assert link_object.contents is not None target_path = cast( AnyStr, link_object.contents ) # pytype: disable=invalid-annotation else: target_path = file_path if self.filesystem.ends_with_path_separator(target_path): error = ( errno.EINVAL if self.filesystem.is_windows_fs else errno.ENOENT if self.filesystem.is_macos else errno.EISDIR ) self.filesystem.raise_os_error(error, file_path) file_object = self.filesystem.create_file_internally( real_path, create_missing_dirs=False, apply_umask=True ) return file_object def _handle_file_arg( self, file_: Union[AnyStr, int] ) -> Tuple[ Optional[FakeFile], Optional[AnyStr], Optional[int], Optional[AnyStr], bool ]: file_object = None if isinstance(file_, int): # opening a file descriptor filedes: int = file_ wrapper = self.filesystem.get_open_file(filedes) can_write = True if isinstance(wrapper, FakePipeWrapper): return None, None, filedes, None, can_write if isinstance(wrapper, FakeFileWrapper): self._delete_on_close = wrapper.delete_on_close can_write = wrapper.allow_update file_object = cast( FakeFile, self.filesystem.get_open_file(filedes).get_object() ) assert file_object is not None path = file_object.name return ( # pytype: disable=bad-return-type file_object, cast(AnyStr, path), # pytype: disable=invalid-annotation filedes, cast(AnyStr, path), # pytype: disable=invalid-annotation can_write, ) # open a file by path file_path = cast(AnyStr, file_) # pytype: disable=invalid-annotation if file_path == self.filesystem.dev_null.name: file_object = self.filesystem.dev_null real_path = file_path else: real_path = self.filesystem.resolve_path(file_path) if self.filesystem.exists(file_path): file_object = self.filesystem.get_object_from_normpath( real_path, check_read_perm=False ) return file_object, file_path, None, real_path, True def _handle_file_mode( self, mode: str, newline: Optional[str], open_modes: Optional[_OpenModes], ) -> Tuple[Optional[str], _OpenModes]: orig_modes = mode # Save original modes for error messages. # Normalize modes. Handle 't' and 'U'. if ("b" in mode and "t" in mode) or ( sys.version_info > (3, 10) and "U" in mode ): raise ValueError("Invalid mode: " + mode) mode = mode.replace("t", "").replace("b", "") mode = mode.replace("rU", "r").replace("U", "r") if not self.raw_io: if mode not in _OPEN_MODE_MAP: raise ValueError("Invalid mode: %r" % orig_modes) open_modes = _OpenModes(_OPEN_MODE_MAP[mode]) assert open_modes is not None return newline, open_modes	14.95 KB	2026-01-20 12:58
📄 fake_os.py # Copyright 2009 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Uses :py:class:`FakeOsModule` to provide a fake :py:mod:`os` module replacement. """ import errno import functools import inspect import os import sys import uuid from contextlib import contextmanager from stat import ( S_IFREG, S_IFSOCK, ) from typing import ( List, Optional, Callable, Union, Any, Tuple, cast, AnyStr, TYPE_CHECKING, Set, ) from pyfakefs.fake_file import ( FakeDirectory, FakeDirWrapper, StandardStreamWrapper, FakeFileWrapper, FakePipeWrapper, FakeFile, AnyFileWrapper, ) from pyfakefs.fake_open import FakeFileOpen, _OpenModes from pyfakefs.fake_path import FakePathModule from pyfakefs.fake_scandir import scandir, walk, ScanDirIter from pyfakefs.helpers import ( FakeStatResult, is_called_from_skipped_module, is_int_type, is_byte_string, make_string_path, IS_PYPY, to_string, matching_string, AnyString, to_bytes, PERM_EXE, PERM_DEF, is_root, get_uid, get_gid, ) if TYPE_CHECKING: from pyfakefs.fake_filesystem import FakeFilesystem NR_STD_STREAMS = 3 class FakeOsModule: """Uses FakeFilesystem to provide a fake os module replacement. Do not create `os.path` separately from `os`, as there is a necessary circular dependency between `os` and `os.path` to replicate the behavior of the standard Python modules. What you want to do is to just let `FakeOsModule` take care of `os.path` setup itself. If you are creating the fake filesystem manually, you need to do:: filesystem = fake_filesystem.FakeFilesystem() my_os_module = fake_os.FakeOsModule(filesystem) """ use_original = False @staticmethod def dir() -> List[str]: """Return the list of patched function names. Used for patching functions imported from the module. """ _dir = [ "access", "chdir", "chmod", "chown", "close", "dup", "dup2", "fstat", "fsync", "getcwd", "lchmod", "link", "listdir", "lseek", "lstat", "makedirs", "mkdir", "mknod", "open", "read", "readlink", "remove", "removedirs", "rename", "rmdir", "scandir", "stat", "symlink", "umask", "unlink", "utime", "walk", "write", "getcwdb", "replace", ] if sys.platform.startswith("linux"): _dir += [ "fdatasync", "getxattr", "listxattr", "removexattr", "setxattr", ] if sys.platform != "win32": _dir += [ "getgid", "getuid", ] return _dir def __init__(self, filesystem: "FakeFilesystem"): """Also exposes self.path (to fake os.path). Args: filesystem: FakeFilesystem used to provide file system information """ self.filesystem = filesystem self.os_module: Any = os self.path = FakePathModule(self.filesystem, self) self._supports_follow_symlinks: Optional[Set] = None self._supports_dir_fd: Optional[Set] = None self._supports_effective_ids: Optional[Set] = None self._supports_fd: Optional[Set] = None @property def devnull(self) -> str: return self.path.devnull @property def sep(self) -> str: return self.path.sep @property def altsep(self) -> Optional[str]: return self.path.altsep @property def linesep(self) -> str: return self.path.linesep @property def pathsep(self) -> str: return self.path.pathsep def fdopen(self, fd: int, args: Any, kwargs: Any) -> AnyFileWrapper: """Redirector to open() builtin function. Args: fd: The file descriptor of the file to open. args: Pass through args. *kwargs: Pass through kwargs. Returns: File object corresponding to file_des. Raises: TypeError: if file descriptor is not an integer. """ if not is_int_type(fd): raise TypeError("an integer is required") return FakeFileOpen(self.filesystem)(fd, args, *kwargs) def _umask(self) -> int: """Return the current umask.""" if self.filesystem.is_windows_fs: # windows always returns 0 - it has no real notion of umask return 0 if sys.platform == "win32": # if we are testing Unix under Windows we assume a default mask return 0o002 else: # under Unix, we return the real umask; # there is no pure getter for umask, so we have to first # set a mode to get the previous one and then re-set that mask = os.umask(0) os.umask(mask) return mask def open( self, path: AnyStr, flags: int, mode: Optional[int] = None, , dir_fd: Optional[int] = None, ) -> int: """Return the file descriptor for a FakeFile. Args: path: the path to the file flags: low-level bits to indicate io operation mode: bits to define default permissions Note: only basic modes are supported, OS-specific modes are ignored dir_fd: If not `None`, the file descriptor of a directory, with `file_path` being relative to this directory. Returns: A file descriptor. Raises: OSError: if the path cannot be found ValueError: if invalid mode is given NotImplementedError: if `os.O_EXCL` is used without `os.O_CREAT` """ path = self._path_with_dir_fd(path, self.open, dir_fd) if mode is None: if self.filesystem.is_windows_fs: mode = 0o666 else: mode = 0o777 & ~self._umask() has_directory_flag = ( hasattr(os, "O_DIRECTORY") and flags & os.O_DIRECTORY == os.O_DIRECTORY ) if ( has_directory_flag and self.filesystem.exists(path) and not self.filesystem.isdir(path) ): self.filesystem.raise_os_error(errno.ENOTDIR, path) has_follow_flag = ( hasattr(os, "O_NOFOLLOW") and flags & os.O_NOFOLLOW == os.O_NOFOLLOW ) if has_follow_flag and self.filesystem.islink(path): self.filesystem.raise_os_error(errno.ELOOP, path) has_tmpfile_flag = ( hasattr(os, "O_TMPFILE") and flags & os.O_TMPFILE == os.O_TMPFILE ) open_modes = _OpenModes( must_exist=not flags & os.O_CREAT and not has_tmpfile_flag, can_read=not flags & os.O_WRONLY, can_write=flags & (os.O_RDWR \| os.O_WRONLY) != 0, truncate=flags & os.O_TRUNC != 0, append=flags & os.O_APPEND != 0, must_not_exist=flags & os.O_EXCL != 0, ) if open_modes.must_not_exist and open_modes.must_exist: raise NotImplementedError("O_EXCL without O_CREAT mode is not supported") if has_tmpfile_flag: # this is a workaround for tempfiles that do not have a filename # as we do not support this directly, we just add a unique filename # and set the file to delete on close path = self.filesystem.joinpaths( path, matching_string(path, str(uuid.uuid4())) ) if not self.filesystem.is_windows_fs and self.filesystem.exists(path): # handle opening directory - only allowed under Posix # with read-only mode obj = self.filesystem.resolve(path) if isinstance(obj, FakeDirectory): if ( not open_modes.must_exist and not self.filesystem.is_macos ) or open_modes.can_write: self.filesystem.raise_os_error(errno.EISDIR, path) dir_wrapper = FakeDirWrapper(obj, path, self.filesystem) file_des = self.filesystem.add_open_file(dir_wrapper) dir_wrapper.filedes = file_des return file_des # low level open is always binary str_flags = "b" delete_on_close = has_tmpfile_flag if hasattr(os, "O_TEMPORARY"): delete_on_close = flags & os.O_TEMPORARY == os.O_TEMPORARY fake_file = FakeFileOpen( self.filesystem, delete_on_close=delete_on_close, raw_io=True )(path, str_flags, open_modes=open_modes) assert not isinstance(fake_file, StandardStreamWrapper) if fake_file.file_object != self.filesystem.dev_null: self.chmod(path, mode) return fake_file.fileno() def close(self, fd: int) -> None: """Close a file descriptor. Args: fd: An integer file descriptor for the file object requested. Raises: OSError: bad file descriptor. TypeError: if file descriptor is not an integer. """ file_handle = self.filesystem.get_open_file(fd) file_handle.close_fd(fd) def dup(self, fd: int) -> int: file_handle = self.filesystem.get_open_file(fd) return self.filesystem.add_open_file(file_handle) def dup2(self, fd: int, fd2: int, inheritable: bool = True) -> int: if fd == fd2: return fd file_handle = self.filesystem.get_open_file(fd) return self.filesystem.add_open_file(file_handle, fd2) def read(self, fd: int, n: int) -> bytes: """Read number of bytes from a file descriptor, returns bytes read. Args: fd: An integer file descriptor for the file object requested. n: Number of bytes to read from file. Returns: Bytes read from file. Raises: OSError: bad file descriptor. TypeError: if file descriptor is not an integer. """ file_handle = self.filesystem.get_open_file(fd) if isinstance(file_handle, FakeFileWrapper): file_handle.raw_io = True if isinstance(file_handle, FakeDirWrapper): self.filesystem.raise_os_error(errno.EBADF, file_handle.file_path) return file_handle.read(n) def write(self, fd: int, contents: bytes) -> int: """Write string to file descriptor, returns number of bytes written. Args: fd: An integer file descriptor for the file object requested. contents: String of bytes to write to file. Returns: Number of bytes written. Raises: OSError: bad file descriptor. TypeError: if file descriptor is not an integer. """ file_handle = cast(FakeFileWrapper, self.filesystem.get_open_file(fd)) if isinstance(file_handle, FakeDirWrapper): self.filesystem.raise_os_error(errno.EBADF, file_handle.file_path) if isinstance(file_handle, FakePipeWrapper): return file_handle.write(contents) file_handle.raw_io = True file_handle._sync_io() file_handle.update_flush_pos() file_handle.write(contents) file_handle.flush() return len(contents) def lseek(self, fd: int, pos: int, whence: int): file_handle = self.filesystem.get_open_file(fd) if isinstance(file_handle, FakeFileWrapper): file_handle.seek(pos, whence) else: self.filesystem.raise_os_error(errno.EBADF) def pipe(self) -> Tuple[int, int]: read_fd, write_fd = os.pipe() read_wrapper = FakePipeWrapper(self.filesystem, read_fd, False) file_des = self.filesystem.add_open_file(read_wrapper) read_wrapper.filedes = file_des write_wrapper = FakePipeWrapper(self.filesystem, write_fd, True) file_des = self.filesystem.add_open_file(write_wrapper) write_wrapper.filedes = file_des return read_wrapper.filedes, write_wrapper.filedes def fstat(self, fd: int) -> FakeStatResult: """Return the os.stat-like tuple for the FakeFile object of file_des. Args: fd: The file descriptor of filesystem object to retrieve. Returns: The FakeStatResult object corresponding to entry_path. Raises: OSError: if the filesystem object doesn't exist. """ # stat should return the tuple representing return value of os.stat file_object = self.filesystem.get_open_file(fd).get_object() assert isinstance(file_object, FakeFile) return file_object.stat_result.copy() def umask(self, mask: int) -> int: """Change the current umask. Args: mask: (int) The new umask value. Returns: The old umask. Raises: TypeError: if new_mask is of an invalid type. """ if not is_int_type(mask): raise TypeError("an integer is required") old_umask = self.filesystem.umask self.filesystem.umask = mask return old_umask def chdir(self, path: AnyStr) -> None: """Change current working directory to target directory. Args: path: The path to new current working directory. Raises: OSError: if user lacks permission to enter the argument directory or if the target is not a directory. """ try: path = self.filesystem.resolve_path(path, allow_fd=True) except OSError as exc: if self.filesystem.is_macos and exc.errno == errno.EBADF: self.filesystem.raise_os_error(errno.ENOTDIR, str(path)) elif ( self.filesystem.is_windows_fs and exc.errno == errno.ENOENT and path == "" ): self.filesystem.raise_os_error(errno.EINVAL, str(path)) raise try: self.filesystem.confirmdir(path) except OSError as exc: if exc.errno == errno.EACCES: # no access rights to the parent directory - do nothing return raise directory = self.filesystem.resolve(path) # A full implementation would check permissions all the way # up the tree. if not is_root() and not directory.has_permission(PERM_EXE): self.filesystem.raise_os_error(errno.EACCES, directory.name) self.filesystem.cwd = path # type: ignore[assignment] def getcwd(self) -> str: """Return current working directory.""" return to_string(self.filesystem.cwd) def getcwdb(self) -> bytes: """Return current working directory as bytes.""" return to_bytes(self.filesystem.cwd) def listdir(self, path: AnyStr) -> List[AnyStr]: """Return a list of file names in target_directory. Args: path: Path to the target directory within the fake filesystem. Returns: A list of file names within the target directory in arbitrary order. Raises: OSError: if the target is not a directory. """ return self.filesystem.listdir(path) XATTR_CREATE = 1 XATTR_REPLACE = 2 def getxattr( self, path: AnyStr, attribute: AnyString, , follow_symlinks: bool = True ) -> Optional[bytes]: """Return the value of the given extended filesystem attribute for `path`. Args: path: File path, file descriptor or path-like object. attribute: (str or bytes) The attribute name. follow_symlinks: (bool) If `True` (the default), symlinks in the path are traversed. Returns: The contents of the extended attribute as bytes or `None` if the attribute does not exist. Raises: OSError: if the path does not exist. """ if not self.filesystem.is_linux: raise AttributeError("module 'os' has no attribute 'getxattr'") if isinstance(attribute, bytes): attribute = attribute.decode(sys.getfilesystemencoding()) file_obj = self.filesystem.resolve(path, follow_symlinks, allow_fd=True) if attribute not in file_obj.xattr: self.filesystem.raise_os_error(errno.ENODATA, path) return file_obj.xattr.get(attribute) def listxattr( self, path: Optional[AnyStr] = None, , follow_symlinks: bool = True ) -> List[str]: """Return a list of the extended filesystem attributes on `path`. Args: path: File path, file descriptor or path-like object. If `None`, the current directory is used. follow_symlinks: (bool) If `True` (the default), symlinks in the path are traversed. Returns: A list of all attribute names for the given path as str. Raises: OSError: if the path does not exist. """ if not self.filesystem.is_linux: raise AttributeError("module 'os' has no attribute 'listxattr'") path_str = self.filesystem.cwd if path is None else path file_obj = self.filesystem.resolve( cast(AnyStr, path_str), # pytype: disable=invalid-annotation follow_symlinks, allow_fd=True, ) return list(file_obj.xattr.keys()) def removexattr( self, path: AnyStr, attribute: AnyString, , follow_symlinks: bool = True ) -> None: """Removes the extended filesystem attribute from `path`. Args: path: File path, file descriptor or path-like object attribute: (str or bytes) The attribute name. follow_symlinks: (bool) If `True` (the default), symlinks in the path are traversed. Raises: OSError: if the path does not exist. """ if not self.filesystem.is_linux: raise AttributeError("module 'os' has no attribute 'removexattr'") if isinstance(attribute, bytes): attribute = attribute.decode(sys.getfilesystemencoding()) file_obj = self.filesystem.resolve(path, follow_symlinks, allow_fd=True) if attribute in file_obj.xattr: del file_obj.xattr[attribute] def setxattr( self, path: AnyStr, attribute: AnyString, value: bytes, flags: int = 0, , follow_symlinks: bool = True, ) -> None: """Sets the value of the given extended filesystem attribute for `path`. Args: path: File path, file descriptor or path-like object. attribute: The attribute name (str or bytes). value: (byte-like) The value to be set. follow_symlinks: (bool) If `True` (the default), symlinks in the path are traversed. Raises: OSError: if the path does not exist. TypeError: if `value` is not a byte-like object. """ if not self.filesystem.is_linux: raise AttributeError("module 'os' has no attribute 'setxattr'") if isinstance(attribute, bytes): attribute = attribute.decode(sys.getfilesystemencoding()) if not is_byte_string(value): raise TypeError("a bytes-like object is required") file_obj = self.filesystem.resolve(path, follow_symlinks, allow_fd=True) exists = attribute in file_obj.xattr if exists and flags == self.XATTR_CREATE: self.filesystem.raise_os_error(errno.ENODATA, file_obj.path) if not exists and flags == self.XATTR_REPLACE: self.filesystem.raise_os_error(errno.EEXIST, file_obj.path) file_obj.xattr[attribute] = value def scandir(self, path: str = ".") -> ScanDirIter: """Return an iterator of DirEntry objects corresponding to the entries in the directory given by path. Args: path: Path to the target directory within the fake filesystem. Returns: An iterator to an unsorted list of os.DirEntry objects for each entry in path. Raises: OSError: if the target is not a directory. """ return scandir(self.filesystem, path) def walk( self, top: AnyStr, topdown: bool = True, onerror: Optional[bool] = None, followlinks: bool = False, ): """Perform an os.walk operation over the fake filesystem. Args: top: The root directory from which to begin walk. topdown: Determines whether to return the tuples with the root as the first entry (`True`) or as the last, after all the child directory tuples (`False`). onerror: If not `None`, function which will be called to handle the `os.error` instance provided when `os.listdir()` fails. followlinks: If `True`, symbolic links are followed. Yields: (path, directories, nondirectories) for top and each of its subdirectories. See the documentation for the builtin os module for further details. """ return walk(self.filesystem, top, topdown, onerror, followlinks) def readlink(self, path: AnyStr, dir_fd: Optional[int] = None) -> str: """Read the target of a symlink. Args: path: Symlink to read the target of. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. Returns: the string representing the path to which the symbolic link points. Raises: TypeError: if `path` is `None` OSError: (with errno=ENOENT) if path is not a valid path, or (with errno=EINVAL) if path is valid, but is not a symlink """ path = self._path_with_dir_fd(path, self.readlink, dir_fd) return self.filesystem.readlink(path) def stat( self, path: AnyStr, , dir_fd: Optional[int] = None, follow_symlinks: bool = True, ) -> FakeStatResult: """Return the os.stat-like tuple for the FakeFile object of entry_path. Args: path: path to filesystem object to retrieve. dir_fd: (int) If not `None`, the file descriptor of a directory, with `entry_path` being relative to this directory. follow_symlinks: (bool) If `False` and `entry_path` points to a symlink, the link itself is changed instead of the linked object. Returns: The FakeStatResult object corresponding to entry_path. Raises: OSError: if the filesystem object doesn't exist. """ path = self._path_with_dir_fd(path, self.stat, dir_fd) return self.filesystem.stat(path, follow_symlinks) def lstat(self, path: AnyStr, , dir_fd: Optional[int] = None) -> FakeStatResult: """Return the os.stat-like tuple for entry_path, not following symlinks. Args: path: path to filesystem object to retrieve. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. Returns: the FakeStatResult object corresponding to `path`. Raises: OSError: if the filesystem object doesn't exist. """ # stat should return the tuple representing return value of os.stat path = self._path_with_dir_fd(path, self.lstat, dir_fd, check_supported=False) return self.filesystem.stat(path, follow_symlinks=False) def remove(self, path: AnyStr, dir_fd: Optional[int] = None) -> None: """Remove the FakeFile object at the specified file path. Args: path: Path to file to be removed. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. Raises: OSError: if path points to a directory. OSError: if path does not exist. OSError: if removal failed. """ path = self._path_with_dir_fd(path, self.remove, dir_fd, check_supported=False) self.filesystem.remove(path) def unlink(self, path: AnyStr, , dir_fd: Optional[int] = None) -> None: """Remove the FakeFile object at the specified file path. Args: path: Path to file to be removed. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. Raises: OSError: if path points to a directory. OSError: if path does not exist. OSError: if removal failed. """ path = self._path_with_dir_fd(path, self.unlink, dir_fd) self.filesystem.remove(path) def rename( self, src: AnyStr, dst: AnyStr, , src_dir_fd: Optional[int] = None, dst_dir_fd: Optional[int] = None, ) -> None: """Rename a FakeFile object at old_file_path to new_file_path, preserving all properties. Also replaces existing new_file_path object, if one existed (Unix only). Args: src: Path to filesystem object to rename. dst: Path to where the filesystem object will live after this call. src_dir_fd: If not `None`, the file descriptor of a directory, with `src` being relative to this directory. dst_dir_fd: If not `None`, the file descriptor of a directory, with `dst` being relative to this directory. Raises: OSError: if old_file_path does not exist. OSError: if new_file_path is an existing directory. OSError: if new_file_path is an existing file (Windows only) OSError: if new_file_path is an existing file and could not be removed (Unix) OSError: if `dirname(new_file)` does not exist OSError: if the file would be moved to another filesystem (e.g. mount point) """ src = self._path_with_dir_fd(src, self.rename, src_dir_fd) dst = self._path_with_dir_fd(dst, self.rename, dst_dir_fd) self.filesystem.rename(src, dst) def renames(self, old: AnyStr, new: AnyStr): """Fakes `os.renames`, documentation taken from there. Super-rename; create directories as necessary and delete any left empty. Works like rename, except creation of any intermediate directories needed to make the new pathname good is attempted first. After the rename, directories corresponding to rightmost path segments of the old name will be pruned until either the whole path is consumed or a nonempty directory is found. Note: this function can fail with the new directory structure made if you lack permissions needed to unlink the leaf directory or file. """ head, tail = self.filesystem.splitpath(new) if head and tail and not self.filesystem.exists(head): self.makedirs(head) self.rename(old, new) head, tail = self.filesystem.splitpath(old) if head and tail: try: self.removedirs(head) except OSError: pass def replace( self, src: AnyStr, dst: AnyStr, , src_dir_fd: Optional[int] = None, dst_dir_fd: Optional[int] = None, ) -> None: """Renames a FakeFile object at old_file_path to new_file_path, preserving all properties. Also replaces existing new_file_path object, if one existed. Arg src: Path to filesystem object to rename. dst: Path to where the filesystem object will live after this call. src_dir_fd: If not `None`, the file descriptor of a directory, with `src` being relative to this directory. dst_dir_fd: If not `None`, the file descriptor of a directory, with `dst` being relative to this directory. Raises: OSError: if old_file_path does not exist. OSError: if new_file_path is an existing directory. OSError: if new_file_path is an existing file and could not be removed OSError: if `dirname(new_file)` does not exist OSError: if the file would be moved to another filesystem (e.g. mount point) """ src = self._path_with_dir_fd( src, self.rename, src_dir_fd, check_supported=False ) dst = self._path_with_dir_fd( dst, self.rename, dst_dir_fd, check_supported=False ) self.filesystem.rename(src, dst, force_replace=True) def rmdir(self, path: AnyStr, , dir_fd: Optional[int] = None) -> None: """Remove a leaf Fake directory. Args: path: (str) Name of directory to remove. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. Raises: OSError: if `path` does not exist or is not a directory, or as per FakeFilesystem.remove_object. Cannot remove '.'. """ path = self._path_with_dir_fd(path, self.rmdir, dir_fd) self.filesystem.rmdir(path) def removedirs(self, name: AnyStr) -> None: """Remove a leaf fake directory and all empty intermediate ones. Args: name: the directory to be removed. Raises: OSError: if target_directory does not exist or is not a directory. OSError: if target_directory is not empty. """ name = self.filesystem.absnormpath(name) directory = self.filesystem.confirmdir(name) if directory.entries: self.filesystem.raise_os_error(errno.ENOTEMPTY, self.path.basename(name)) else: self.rmdir(name) head, tail = self.path.split(name) if not tail: head, tail = self.path.split(head) while head and tail: head_dir = self.filesystem.confirmdir(head) if head_dir.entries: break # only the top-level dir may not be a symlink self.filesystem.rmdir(head, allow_symlink=True) head, tail = self.path.split(head) def mkdir( self, path: AnyStr, mode: int = PERM_DEF, , dir_fd: Optional[int] = None ) -> None: """Create a leaf Fake directory. Args: path: (str) Name of directory to create. Relative paths are assumed to be relative to '/'. mode: (int) Mode to create directory with. This argument defaults to 0o777. The umask is applied to this mode. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. Raises: OSError: if the directory name is invalid or parent directory is read only or as per FakeFilesystem.add_object. """ path = self._path_with_dir_fd(path, self.mkdir, dir_fd) try: self.filesystem.makedir(path, mode) except OSError as e: if e.errno == errno.EACCES: self.filesystem.raise_os_error(e.errno, path) raise def makedirs( self, name: AnyStr, mode: int = PERM_DEF, exist_ok: Optional[bool] = None ) -> None: """Create a leaf Fake directory + create any non-existent parent dirs. Args: name: (str) Name of directory to create. mode: (int) Mode to create directory (and any necessary parent directories) with. This argument defaults to 0o777. The umask is applied to this mode. exist_ok: (boolean) If exist_ok is `False` (the default), an `OSError` is raised if the target directory already exists. Raises: OSError: if the directory already exists and exist_ok=False, or as per :py:meth:`FakeFilesystem.create_dir`. """ if exist_ok is None: exist_ok = False # copied and adapted from real implementation in os.py (Python 3.12) head, tail = self.filesystem.splitpath(name) if not tail: head, tail = self.filesystem.splitpath(head) if head and tail and not self.filesystem.exists(head): try: self.makedirs(head, exist_ok=exist_ok) except FileExistsError: pass cdir = self.filesystem.cwd if isinstance(tail, bytes): if tail == bytes(cdir, "ASCII"): return elif tail == cdir: return try: self.mkdir(name, mode) except OSError: if not exist_ok or not self.filesystem.isdir(name): raise def _path_with_dir_fd( self, path: AnyStr, fct: Callable, dir_fd: Optional[int], check_supported: bool = True, ) -> AnyStr: """Return the path considering dir_fd. Raise on invalid parameters.""" try: path = make_string_path(path) except TypeError: # the error is handled later path = path if dir_fd is not None: # check if fd is supported for the built-in real function if check_supported and (fct not in self.supports_dir_fd): raise NotImplementedError("dir_fd unavailable on this platform") if isinstance(path, int): raise ValueError( "%s: Can't specify dir_fd without matching path_str" % fct.__name__ ) if not self.path.isabs(path): open_file = self.filesystem.get_open_file(dir_fd) return self.path.join( # type: ignore[type-var, return-value] cast(FakeFile, open_file.get_object()).path, path ) return path def truncate(self, path: AnyStr, length: int) -> None: """Truncate the file corresponding to path, so that it is length bytes in size. If length is larger than the current size, the file is filled up with zero bytes. Args: path: (str or int) Path to the file, or an integer file descriptor for the file object. length: (int) Length of the file after truncating it. Raises: OSError: if the file does not exist or the file descriptor is invalid. """ file_object = self.filesystem.resolve(path, allow_fd=True) file_object.size = length def ftruncate(self, fd: int, length: int) -> None: """Truncate the file corresponding to fd, so that it is length bytes in size. If length is larger than the current size, the file is filled up with zero bytes. Args: fd: (int) File descriptor for the file object. length: (int) Maximum length of the file after truncating it. Raises: OSError: if the file descriptor is invalid """ file_object = self.filesystem.get_open_file(fd).get_object() if isinstance(file_object, FakeFileWrapper): file_object.size = length else: self.filesystem.raise_os_error(errno.EBADF) def access( self, path: AnyStr, mode: int, , dir_fd: Optional[int] = None, effective_ids: Optional[bool] = None, follow_symlinks: Optional[bool] = None, ) -> bool: """Check if a file exists and has the specified permissions. Args: path: (str) Path to the file. mode: (int) Permissions represented as a bitwise-OR combination of os.F_OK, os.R_OK, os.W_OK, and os.X_OK. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. effective_ids: (bool) Unused. Only here to match the signature. follow_symlinks: (bool) If `False` and `path` points to a symlink, the link itself is queried instead of the linked object. Returns: bool, `True` if file is accessible, `False` otherwise. """ if effective_ids is not None and self.filesystem.is_windows_fs: raise NotImplementedError( "access: effective_ids unavailable on this platform" ) if follow_symlinks is None: # different behavior under Windows follow_symlinks = not self.filesystem.is_windows_fs elif self.filesystem.is_windows_fs: raise NotImplementedError( "access: follow_symlinks unavailable on this platform" ) path = self._path_with_dir_fd(path, self.access, dir_fd) try: stat_result = self.stat(path, follow_symlinks=follow_symlinks) except OSError as os_error: if os_error.errno == errno.ENOENT: return False raise if is_root(): if mode & os.X_OK: return stat_result.st_mode & 0o111 != 0 return True return (mode & ((stat_result.st_mode >> 6) & 7)) == mode def fchmod( self, fd: int, mode: int, ) -> None: """Change the permissions of an open file as encoded in integer mode. Args: fd: (int) File descriptor. mode: (int) Permissions. """ if self.filesystem.is_windows_fs and sys.version_info < (3, 13): raise AttributeError( "module 'os' has no attribute 'fchmod'. Did you mean: 'chmod'?" ) self.filesystem.chmod(fd, mode) def chmod( self, path: AnyStr, mode: int, , dir_fd: Optional[int] = None, follow_symlinks: bool = True, ) -> None: """Change the permissions of a file as encoded in integer mode. Args: path: (str) Path to the file. mode: (int) Permissions. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. follow_symlinks: (bool) If `False` and `path` points to a symlink, the link itself is queried instead of the linked object. """ if not follow_symlinks and ( self.chmod not in self.supports_follow_symlinks or IS_PYPY ): raise NotImplementedError( "`follow_symlinks` for chmod() is not available on this system" ) path = self._path_with_dir_fd(path, self.chmod, dir_fd) self.filesystem.chmod(path, mode, follow_symlinks) def lchmod(self, path: AnyStr, mode: int) -> None: """Change the permissions of a file as encoded in integer mode. If the file is a link, the permissions of the link are changed. Args: path: (str) Path to the file. mode: (int) Permissions. """ if self.filesystem.is_windows_fs: raise NameError("name 'lchmod' is not defined") self.filesystem.chmod(path, mode, follow_symlinks=False) def utime( self, path: AnyStr, times: Optional[Tuple[Union[int, float], Union[int, float]]] = None, ns: Optional[Tuple[int, int]] = None, dir_fd: Optional[int] = None, follow_symlinks: bool = True, ) -> None: """Change the access and modified times of a file. Args: path: (str) Path to the file. times: 2-tuple of int or float numbers, of the form (atime, mtime) which is used to set the access and modified times in seconds. If `None`, both times are set to the current time. ns: 2-tuple of int numbers, of the form (atime, mtime) which is used to set the access and modified times in nanoseconds. If `None`, both times are set to the current time. dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. follow_symlinks: (bool) If `False` and `path` points to a symlink, the link itself is queried instead of the linked object. Raises: TypeError: If anything other than the expected types is specified in the passed ``times`` or ``ns`` tuple, or if the tuple length is not equal to 2. ValueError: If both ``times`` and ``ns`` are specified. """ path = self._path_with_dir_fd(path, self.utime, dir_fd) self.filesystem.utime(path, times=times, ns=ns, follow_symlinks=follow_symlinks) def chown( self, path: AnyStr, uid: int, gid: int, , dir_fd: Optional[int] = None, follow_symlinks: bool = True, ) -> None: """Set ownership of a faked file. Args: path: (str) Path to the file or directory. uid: (int) Numeric uid to set the file or directory to. gid: (int) Numeric gid to set the file or directory to. dir_fd: (int) If not `None`, the file descriptor of a directory, with `path` being relative to this directory. follow_symlinks: (bool) If `False` and path points to a symlink, the link itself is changed instead of the linked object. Raises: OSError: if path does not exist. `None` is also allowed for `uid` and `gid`. This permits `os.rename` to use `os.chown` even when the source file `uid` and `gid` are `None` (unset). """ path = self._path_with_dir_fd(path, self.chown, dir_fd) file_object = self.filesystem.resolve(path, follow_symlinks, allow_fd=True) if not isinstance(uid, int) or not isinstance(gid, int): raise TypeError("An integer is required") if uid != -1: file_object.st_uid = uid if gid != -1: file_object.st_gid = gid def mknod( self, path: AnyStr, mode: Optional[int] = None, device: int = 0, , dir_fd: Optional[int] = None, ) -> None: """Create a filesystem node named 'filename'. Does not support device special files or named pipes as the real os module does. Args: path: (str) Name of the file to create mode: (int) Permissions to use and type of file to be created. Default permissions are 0o666. Only the stat.S_IFREG file type is supported by the fake implementation. The umask is applied to this mode. device: not supported in fake implementation dir_fd: If not `None`, the file descriptor of a directory, with `path` being relative to this directory. Raises: OSError: if called with unsupported options or the file can not be created. """ if self.filesystem.is_windows_fs: raise AttributeError("module 'os' has no attribute 'mknode'") if mode is None: # note that a default value of 0o600 without a device type is # documented - this is not how it seems to work mode = S_IFREG \| 0o600 if device or not mode & S_IFREG and not is_root(): self.filesystem.raise_os_error(errno.EPERM) path = self._path_with_dir_fd(path, self.mknod, dir_fd) head, tail = self.path.split(path) if not tail: if self.filesystem.exists(head, check_link=True): self.filesystem.raise_os_error(errno.EEXIST, path) self.filesystem.raise_os_error(errno.ENOENT, path) if tail in (matching_string(tail, "."), matching_string(tail, "..")): self.filesystem.raise_os_error(errno.ENOENT, path) if self.filesystem.exists(path, check_link=True): self.filesystem.raise_os_error(errno.EEXIST, path) self.filesystem.add_object( head, FakeFile(tail, mode & ~self.filesystem.umask, filesystem=self.filesystem), ) def symlink( self, src: AnyStr, dst: AnyStr, target_is_directory: bool = False, , dir_fd: Optional[int] = None, ) -> None: """Creates the specified symlink, pointed at the specified link target. Args: src: The target of the symlink. dst: Path to the symlink to create. target_is_directory: Currently ignored. dir_fd: If not `None`, the file descriptor of a directory, with `dst` being relative to this directory. Raises: OSError: if the file already exists. """ dst = self._path_with_dir_fd(dst, self.symlink, dir_fd) self.filesystem.create_symlink(dst, src, create_missing_dirs=False) def link( self, src: AnyStr, dst: AnyStr, , src_dir_fd: Optional[int] = None, dst_dir_fd: Optional[int] = None, follow_symlinks: Optional[bool] = None, ) -> None: """Create a hard link at dst, pointing at src. Args: src: An existing path to the target file. dst: The destination path to create a new link at. src_dir_fd: If not `None`, the file descriptor of a directory, with `src` being relative to this directory. dst_dir_fd: If not `None`, the file descriptor of a directory, with `dst` being relative to this directory. follow_symlinks: (bool) If `True` (the default), symlinks in the path are traversed. Raises: OSError: if something already exists at new_path. OSError: if the parent directory doesn't exist. """ if IS_PYPY and follow_symlinks is not None: raise OSError(errno.EINVAL, "Invalid argument: follow_symlinks") if follow_symlinks is None: follow_symlinks = True src = self._path_with_dir_fd(src, self.link, src_dir_fd) dst = self._path_with_dir_fd(dst, self.link, dst_dir_fd) self.filesystem.link(src, dst, follow_symlinks=follow_symlinks) def fsync(self, fd: int) -> None: """Perform fsync for a fake file (in other words, do nothing). Args: fd: The file descriptor of the open file. Raises: OSError: file_des is an invalid file descriptor. TypeError: file_des is not an integer. """ # Throw an error if file_des isn't valid if 0 <= fd < NR_STD_STREAMS: self.filesystem.raise_os_error(errno.EINVAL) file_object = cast(FakeFileWrapper, self.filesystem.get_open_file(fd)) if self.filesystem.is_windows_fs: if not hasattr(file_object, "allow_update") or not file_object.allow_update: self.filesystem.raise_os_error(errno.EBADF, file_object.file_path) def fdatasync(self, fd: int) -> None: """Perform fdatasync for a fake file (in other words, do nothing). Args: fd: The file descriptor of the open file. Raises: OSError: `fd` is an invalid file descriptor. TypeError: `fd` is not an integer. """ if self.filesystem.is_windows_fs or self.filesystem.is_macos: raise AttributeError("module 'os' has no attribute 'fdatasync'") # Throw an error if file_des isn't valid if 0 <= fd < NR_STD_STREAMS: self.filesystem.raise_os_error(errno.EINVAL) self.filesystem.get_open_file(fd) def sendfile(self, fd_out: int, fd_in: int, offset: int, count: int) -> int: """Copy count bytes from file descriptor fd_in to file descriptor fd_out starting at offset. Args: fd_out: The file descriptor of the destination file. fd_in: The file descriptor of the source file. offset: The offset in bytes where to start the copy in the source file. If `None` (Linux only), copying is started at the current position, and the position is updated. count: The number of bytes to copy. If 0, all remaining bytes are copied (MacOs only). Raises: OSError: If `fd_in` or `fd_out` is an invalid file descriptor. TypeError: If `fd_in` or `fd_out` is not an integer. TypeError: If `offset` is None under MacOs. """ if self.filesystem.is_windows_fs: raise AttributeError("module 'os' has no attribute 'sendfile'") if 0 <= fd_in < NR_STD_STREAMS: self.filesystem.raise_os_error(errno.EINVAL) if 0 <= fd_out < NR_STD_STREAMS: self.filesystem.raise_os_error(errno.EINVAL) source = cast(FakeFileWrapper, self.filesystem.get_open_file(fd_in)) dest = cast(FakeFileWrapper, self.filesystem.get_open_file(fd_out)) if self.filesystem.is_macos: if dest.get_object().stat_result.st_mode & 0o777000 != S_IFSOCK: raise OSError("Socket operation on non-socket") if offset is None: if self.filesystem.is_macos: raise TypeError("None is not a valid offset") contents = source.read(count) else: position = source.tell() source.seek(offset) if count == 0 and self.filesystem.is_macos: contents = source.read() else: contents = source.read(count) source.seek(position) if contents: written = dest.write(contents) dest.flush() return written return 0 def getuid(self) -> int: """Returns the user id set in the fake filesystem. If not changed using ``set_uid``, this is the uid of the real system. """ if self.filesystem.is_windows_fs: raise NameError("name 'getuid' is not defined") return get_uid() def getgid(self) -> int: """Returns the group id set in the fake filesystem. If not changed using ``set_gid``, this is the gid of the real system. """ if self.filesystem.is_windows_fs: raise NameError("name 'getgid' is not defined") return get_gid() if sys.version_info >= (3, 14): def readinto(self, fd, buffer): """Read from a file descriptor fd into a mutable buffer object buffer.""" wrapper = self.filesystem.get_open_file(fd) contents = wrapper.read(len(buffer)) count = len(contents) buffer[:count] = contents return count def fake_functions(self, original_functions) -> Set: functions = set() for fn in original_functions: if hasattr(self, fn.__name__): functions.add(getattr(self, fn.__name__)) else: functions.add(fn) return functions @property def supports_follow_symlinks(self) -> Set[Callable]: if self._supports_follow_symlinks is None: self._supports_follow_symlinks = self.fake_functions( self.os_module.supports_follow_symlinks ) return self._supports_follow_symlinks @property def supports_dir_fd(self) -> Set[Callable]: if self._supports_dir_fd is None: self._supports_dir_fd = self.fake_functions(self.os_module.supports_dir_fd) return self._supports_dir_fd @property def supports_fd(self) -> Set[Callable]: if self._supports_fd is None: self._supports_fd = self.fake_functions(self.os_module.supports_fd) return self._supports_fd @property def supports_effective_ids(self) -> Set[Callable]: if self._supports_effective_ids is None: self._supports_effective_ids = self.fake_functions( self.os_module.supports_effective_ids ) return self._supports_effective_ids def __getattr__(self, name: str) -> Any: """Forwards any unfaked calls to the standard os module.""" return getattr(self.os_module, name) def handle_original_call(f: Callable) -> Callable: """Decorator used for real pathlib Path methods to ensure that real os functions instead of faked ones are used. Applied to all non-private methods of `FakeOsModule`.""" @functools.wraps(f) def wrapped(args, *kwargs): should_use_original = FakeOsModule.use_original if not should_use_original and args: self = args[0] fs: FakeFilesystem = self.filesystem if self.filesystem.patcher: skip_names = fs.patcher.skip_names if is_called_from_skipped_module( skip_names=skip_names, case_sensitive=fs.is_case_sensitive, ): should_use_original = True if should_use_original: # remove the `self` argument for FakeOsModule methods if args and isinstance(args[0], FakeOsModule): args = args[1:] return getattr(os, f.__name__)(args, *kwargs) return f(args, **kwargs) return wrapped for name, fn in inspect.getmembers(FakeOsModule, inspect.isfunction): if not fn.__name__.startswith("_"): setattr(FakeOsModule, name, handle_original_call(fn)) @contextmanager def use_original_os(): """Temporarily use original os functions instead of faked ones. Used to ensure that skipped modules do not use faked calls. """ try: FakeOsModule.use_original = True yield finally: FakeOsModule.use_original = False	54.22 KB	2026-01-20 12:58
📄 fake_path.py # Copyright 2009 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Faked ``os.path`` module replacement. See ``fake_filesystem`` for usage.""" import errno import functools import inspect import os import sys from stat import ( S_IFDIR, S_IFMT, ) from types import ModuleType from typing import ( Callable, List, Optional, Union, Any, Dict, Tuple, AnyStr, overload, ClassVar, TYPE_CHECKING, ) from pyfakefs.helpers import ( is_called_from_skipped_module, make_string_path, to_string, matching_string, to_bytes, ) if TYPE_CHECKING: from pyfakefs.fake_filesystem import FakeFilesystem from pyfakefs.fake_os import FakeOsModule def _copy_module(old: ModuleType) -> ModuleType: """Recompiles and creates new module object.""" saved = sys.modules.pop(old.__name__, None) new = __import__(old.__name__) if saved is not None: sys.modules[old.__name__] = saved return new class FakePathModule: """Faked os.path module replacement. `FakePathModule` should only be instantiated by `FakeOsModule`. See the :py:class:`FakeDirectory<pyfakefs.fake_os.FakeOsModule>` docstring for details. """ _OS_PATH_COPY: Any = _copy_module(os.path) devnull: ClassVar[str] = "" sep: ClassVar[str] = "" altsep: ClassVar[Optional[str]] = None linesep: ClassVar[str] = "" pathsep: ClassVar[str] = "" @staticmethod def dir() -> List[str]: """Return the list of patched function names. Used for patching functions imported from the module. """ dir_list = [ "abspath", "dirname", "exists", "expanduser", "getatime", "getctime", "getmtime", "getsize", "isabs", "isdir", "isfile", "islink", "ismount", "join", "lexists", "normcase", "normpath", "realpath", "relpath", "split", "splitdrive", "samefile", ] if sys.version_info >= (3, 12): dir_list += ["isjunction", "splitroot"] return dir_list def __init__(self, filesystem: "FakeFilesystem", os_module: "FakeOsModule"): """Init. Args: filesystem: FakeFilesystem used to provide file system information """ self.filesystem = filesystem self._os_path = self._OS_PATH_COPY self._os_path.os = self.os = os_module # type: ignore[attr-defined] self.reset(filesystem) @classmethod def reset(cls, filesystem: "FakeFilesystem") -> None: cls.sep = filesystem.path_separator cls.altsep = filesystem.alternative_path_separator cls.linesep = filesystem.line_separator cls.devnull = filesystem.devnull cls.pathsep = filesystem.pathsep def exists(self, path: AnyStr) -> bool: """Determine whether the file object exists within the fake filesystem. Args: path: The path to the file object. Returns: (bool) `True` if the file exists. """ return self.filesystem.exists(path) def lexists(self, path: AnyStr) -> bool: """Test whether a path exists. Returns `True` for broken symbolic links. Args: path: path to the symlink object. Returns: bool (if file exists). """ return self.filesystem.exists(path, check_link=True) def getsize(self, path: AnyStr): """Return the file object size in bytes. Args: path: path to the file object. Returns: file size in bytes. """ file_obj = self.filesystem.resolve(path) if ( self.filesystem.ends_with_path_separator(path) and S_IFMT(file_obj.st_mode) != S_IFDIR ): self.filesystem.raise_os_error(errno.ENOTDIR, path) return file_obj.st_size def isabs(self, path: AnyStr) -> bool: """Return `True` if path is an absolute pathname.""" empty = matching_string(path, "") if self.filesystem.is_windows_fs: drive, path = self.splitdrive(path) else: drive = empty path = make_string_path(path) if not self.filesystem.starts_with_sep(path): return False if self.filesystem.is_windows_fs and sys.version_info >= (3, 13): # from Python 3.13 on, a path under Windows starting with a single separator # (e.g. not a drive and not an UNC path) is no more considered absolute return drive != empty return True def isdir(self, path: AnyStr) -> bool: """Determine if path identifies a directory.""" return self.filesystem.isdir(path) def isfile(self, path: AnyStr) -> bool: """Determine if path identifies a regular file.""" return self.filesystem.isfile(path) def islink(self, path: AnyStr) -> bool: """Determine if path identifies a symbolic link. Args: path: Path to filesystem object. Returns: `True` if path points to a symbolic link. Raises: TypeError: if path is `None`. """ return self.filesystem.islink(path) if sys.version_info >= (3, 12): def isjunction(self, path: AnyStr) -> bool: """Returns `False`. Junctions are never faked.""" return self.filesystem.isjunction(path) def splitroot(self, path: AnyStr): """Split a pathname into drive, root and tail. Implementation taken from ntpath and posixpath. """ return self.filesystem.splitroot(path) if sys.version_info >= (3, 13): def isreserved(self, path): if not self.filesystem.is_windows_fs: raise AttributeError("module 'os' has no attribute 'isreserved'") return self.filesystem.isreserved(path) def getmtime(self, path: AnyStr) -> float: """Returns the modification time of the fake file. Args: path: the path to fake file. Returns: (int, float) the modification time of the fake file in number of seconds since the epoch. Raises: OSError: if the file does not exist. """ try: file_obj = self.filesystem.resolve(path) return file_obj.st_mtime except OSError: self.filesystem.raise_os_error( errno.ENOENT, winerror=3 ) # pytype: disable=bad-return-type def getatime(self, path: AnyStr) -> float: """Returns the last access time of the fake file. Note: Access time is not set automatically in fake filesystem on access. Args: path: the path to fake file. Returns: (int, float) the access time of the fake file in number of seconds since the epoch. Raises: OSError: if the file does not exist. """ try: file_obj = self.filesystem.resolve(path) except OSError: self.filesystem.raise_os_error(errno.ENOENT) return file_obj.st_atime # pytype: disable=name-error def getctime(self, path: AnyStr) -> float: """Returns the creation time of the fake file. Args: path: the path to fake file. Returns: (int, float) the creation time of the fake file in number of seconds since the epoch. Raises: OSError: if the file does not exist. """ try: file_obj = self.filesystem.resolve(path) except OSError: self.filesystem.raise_os_error(errno.ENOENT) return file_obj.st_ctime # pytype: disable=name-error def abspath(self, path: AnyStr) -> AnyStr: """Return the absolute version of a path.""" def getcwd(): """Return the current working directory.""" # pylint: disable=undefined-variable if isinstance(path, bytes): return self.os.getcwdb() else: return self.os.getcwd() path = make_string_path(path) if not self.isabs(path): path = self.join(getcwd(), path) elif self.filesystem.is_windows_fs and self.filesystem.starts_with_sep(path): cwd = getcwd() if self.filesystem.starts_with_drive_letter(cwd): path = self.join(cwd[:2], path) return self.normpath(path) def join(self, p: AnyStr) -> AnyStr: """Return the completed path with a separator of the parts.""" return self.filesystem.joinpaths(p) def split(self, path: AnyStr) -> Tuple[AnyStr, AnyStr]: """Split the path into the directory and the filename of the path.""" return self.filesystem.splitpath(path) def splitdrive(self, path: AnyStr) -> Tuple[AnyStr, AnyStr]: """Split the path into the drive part and the rest of the path, if supported.""" return self.filesystem.splitdrive(path) def normpath(self, path: AnyStr) -> AnyStr: """Normalize path, eliminating double slashes, etc.""" return self.filesystem.normpath(path) def normcase(self, path: AnyStr) -> AnyStr: """Convert to lower case under windows, replaces additional path separator.""" path = self.filesystem.normcase(path) if self.filesystem.is_windows_fs: path = path.lower() return path def relpath(self, path: AnyStr, start: Optional[AnyStr] = None) -> AnyStr: """We mostly rely on the native implementation and adapt the path separator.""" if not path: raise ValueError("no path specified") path = make_string_path(path) path = self.filesystem.replace_windows_root(path) sep = matching_string(path, self.filesystem.path_separator) if start is not None: start = make_string_path(start) else: start = matching_string(path, self.filesystem.cwd) start = self.filesystem.replace_windows_root(start) system_sep = matching_string(path, self._os_path.sep) if self.filesystem.alternative_path_separator is not None: altsep = matching_string(path, self.filesystem.alternative_path_separator) path = path.replace(altsep, system_sep) start = start.replace(altsep, system_sep) path = path.replace(sep, system_sep) start = start.replace(sep, system_sep) path = self._os_path.relpath(path, start) return path.replace(system_sep, sep) def realpath(self, filename: AnyStr, strict: Optional[bool] = None) -> AnyStr: """Return the canonical path of the specified filename, eliminating any symbolic links encountered in the path. """ has_allow_missing = hasattr(os.path, "ALLOW_MISSING") # the strict argument was backported to Python 3.9.23 # together with support for os.path.ALLOW_MISSING if strict is not None and sys.version_info < (3, 10) and not has_allow_missing: raise TypeError("realpath() got an unexpected keyword argument 'strict'") if has_allow_missing and strict == os.path.ALLOW_MISSING: # type: ignore[attr-defined] # ignores non-existing file, but not other errors ignored_error: Any = FileNotFoundError elif strict: # raises on any errors ignored_error = () else: # ignores any OSError while resolving the filename ignored_error = OSError try: self.filesystem.resolve(filename) except ignored_error: pass if self.filesystem.is_windows_fs: return self.abspath(filename) filename = make_string_path(filename) path, ok = self._join_real_path(filename[:0], filename, {}) path = self.abspath(path) return path def samefile(self, path1: AnyStr, path2: AnyStr) -> bool: """Return whether path1 and path2 point to the same file. Args: path1: first file path or path object path2: second file path or path object Raises: OSError: if one of the paths does not point to an existing file system object. """ stat1 = self.filesystem.stat(path1) stat2 = self.filesystem.stat(path2) return stat1.st_ino == stat2.st_ino and stat1.st_dev == stat2.st_dev @overload def _join_real_path( self, path: str, rest: str, seen: Dict[str, Optional[str]] ) -> Tuple[str, bool]: ... @overload def _join_real_path( self, path: bytes, rest: bytes, seen: Dict[bytes, Optional[bytes]] ) -> Tuple[bytes, bool]: ... def _join_real_path( self, path: AnyStr, rest: AnyStr, seen: Dict[AnyStr, Optional[AnyStr]] ) -> Tuple[AnyStr, bool]: """Join two paths, normalizing and eliminating any symbolic links encountered in the second path. Taken from Python source and adapted. """ curdir = matching_string(path, ".") pardir = matching_string(path, "..") sep = self.filesystem.get_path_separator(path) if self.isabs(rest): rest = rest[1:] path = sep while rest: name, _, rest = rest.partition(sep) if not name or name == curdir: # current dir continue if name == pardir: # parent dir if path: path, name = self.filesystem.splitpath(path) if name == pardir: path = self.filesystem.joinpaths(path, pardir, pardir) else: path = pardir continue newpath = self.filesystem.joinpaths(path, name) if not self.filesystem.islink(newpath): path = newpath continue # Resolve the symbolic link if newpath in seen: # Already seen this path seen_path = seen[newpath] if seen_path is not None: # use cached value path = seen_path continue # The symlink is not resolved, so we must have a symlink loop. # Return already resolved part + rest of the path unchanged. return self.filesystem.joinpaths(newpath, rest), False seen[newpath] = None # not resolved symlink path, ok = self._join_real_path( path, matching_string(path, self.filesystem.readlink(newpath)), seen, ) if not ok: return self.filesystem.joinpaths(path, rest), False seen[newpath] = path # resolved symlink return path, True def dirname(self, path: AnyStr) -> AnyStr: """Returns the first part of the result of `split()`.""" return self.split(path)[0] def expanduser(self, path: AnyStr) -> AnyStr: """Return the argument with an initial component of ~ or ~user replaced by that user's home directory. """ path = self._os_path.expanduser(path) return path.replace( matching_string(path, self._os_path.sep), matching_string(path, self.sep), ) def ismount(self, path: AnyStr) -> bool: """Return true if the given path is a mount point. Args: path: Path to filesystem object to be checked Returns: `True` if path is a mount point added to the fake file system. Under Windows also returns `True` for drive and UNC roots (independent of their existence). """ if not path: return False path_str = to_string(make_string_path(path)) normed_path = self.filesystem.absnormpath(path_str) sep = self.filesystem.path_separator if self.filesystem.is_windows_fs: path_seps: Union[Tuple[str, Optional[str]], Tuple[str]] if self.filesystem.alternative_path_separator is not None: path_seps = (sep, self.filesystem.alternative_path_separator) else: path_seps = (sep,) drive, rest = self.filesystem.splitdrive(normed_path) if drive and drive[:1] in path_seps: return (not rest) or (rest in path_seps) if rest in path_seps: return True for mount_point in self.filesystem.mount_points: if to_string(normed_path).rstrip(sep) == to_string(mount_point).rstrip(sep): return True return False def __getattr__(self, name: str) -> Any: """Forwards any non-faked calls to the real os.path.""" return getattr(self._os_path, name) if sys.platform == "win32": class FakeNtModule: """Under windows, a few function of `os.path` are taken from the `nt` module for performance reasons. These are patched here. """ @staticmethod def dir(): if sys.version_info >= (3, 12): return ["_path_exists", "_path_isfile", "_path_isdir", "_path_islink"] else: return ["_isdir"] def __init__(self, filesystem: "FakeFilesystem"): """Init. Args: filesystem: FakeFilesystem used to provide file system information """ import nt # type:ignore[import] self.filesystem = filesystem self.nt_module: Any = nt def getcwd(self) -> str: """Return current working directory.""" return to_string(self.filesystem.cwd) def getcwdb(self) -> bytes: """Return current working directory as bytes.""" return to_bytes(self.filesystem.cwd) if sys.version_info >= (3, 12): def _path_isdir(self, path: AnyStr) -> bool: return self.filesystem.isdir(path) def _path_isfile(self, path: AnyStr) -> bool: return self.filesystem.isfile(path) def _path_islink(self, path: AnyStr) -> bool: return self.filesystem.islink(path) def _path_exists(self, path: AnyStr) -> bool: return self.filesystem.exists(path) else: def _isdir(self, path: AnyStr) -> bool: return self.filesystem.isdir(path) def __getattr__(self, name: str) -> Any: """Forwards any non-faked calls to the real nt module.""" return getattr(self.nt_module, name) def handle_original_call(f: Callable) -> Callable: """Decorator used for real pathlib Path methods to ensure that real os functions instead of faked ones are used. Applied to all non-private methods of `FakePathModule`.""" @functools.wraps(f) def wrapped(args, kwargs): if args: self = args[0] should_use_original = self.os.use_original if not should_use_original and self.filesystem.patcher: skip_names = self.filesystem.patcher.skip_names if is_called_from_skipped_module( skip_names=skip_names, case_sensitive=self.filesystem.is_case_sensitive, ): should_use_original = True if should_use_original: # remove the `self` argument for FakePathModule methods if args and isinstance(args[0], FakePathModule): args = args[1:] return getattr(os.path, f.__name__)(args, *kwargs) return f(args, **kwargs) return wrapped for name, fn in inspect.getmembers(FakePathModule, inspect.isfunction): if not fn.__name__.startswith("_"): setattr(FakePathModule, name, handle_original_call(fn))	20.33 KB	2026-01-20 12:58
📄 fake_pathlib.py # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """A fake implementation for pathlib working with FakeFilesystem. New in pyfakefs 3.0. Usage: If using `fake_filesystem_unittest.TestCase`, pytest fs fixture, or directly `Patcher`, pathlib gets replaced by fake_pathlib together with other file system related modules. Note: as the implementation is based on FakeFilesystem, all faked classes (including PurePosixPath, PosixPath, PureWindowsPath and WindowsPath) get the properties of the underlying fake filesystem. """ import errno import fnmatch import functools import inspect import ntpath import os import pathlib import posixpath import re import sys import warnings from pathlib import PurePath from typing import Callable, List, Optional from unittest import mock from urllib.parse import quote_from_bytes as urlquote_from_bytes from pyfakefs import fake_scandir from pyfakefs.fake_filesystem import FakeFilesystem from pyfakefs.fake_open import fake_open from pyfakefs.fake_os import FakeOsModule, use_original_os from pyfakefs.fake_path import FakePathModule from pyfakefs.helpers import IS_PYPY, is_called_from_skipped_module, FSType _WIN_RESERVED_NAMES = ( {"CON", "PRN", "AUX", "NUL"} \| {"COM%d" % i for i in range(1, 10)} \| {"LPT%d" % i for i in range(1, 10)} ) def init_module(filesystem): """Initializes the fake module with the fake file system.""" # pylint: disable=protected-access FakePath.filesystem = filesystem if sys.version_info < (3, 12): FakePathlibModule.WindowsPath._flavour = _FakeWindowsFlavour(filesystem) FakePathlibModule.PosixPath._flavour = _FakePosixFlavour(filesystem) # Pure POSIX path separators must be filesystem-independent. fake_pure_posix_flavour = _FakePosixFlavour(filesystem) fake_pure_posix_flavour.sep = "/" fake_pure_posix_flavour.altsep = None FakePathlibModule.PurePosixPath._flavour = fake_pure_posix_flavour # Pure Windows path separators must be filesystem-independent. fake_pure_nt_flavour = _FakeWindowsFlavour(filesystem) fake_pure_nt_flavour.sep = "\\" fake_pure_nt_flavour.altsep = "/" FakePathlibModule.PureWindowsPath._flavour = fake_pure_nt_flavour else: # in Python > 3.11, the flavour is no longer a separate class, # but points to the os-specific path module (posixpath/ntpath) fake_os_posix = FakeOsModule(filesystem) if filesystem.is_windows_fs: fake_os_posix.path = FakePosixPathModule(filesystem, fake_os_posix) fake_os_windows = FakeOsModule(filesystem) if not filesystem.is_windows_fs: fake_os_windows.path = FakeWindowsPathModule(filesystem, fake_os_windows) parser_name = "_flavour" if sys.version_info < (3, 13) else "parser" # Pure POSIX path properties must be filesystem independent. setattr(FakePathlibModule.PurePosixPath, parser_name, fake_os_posix.path) # Pure Windows path properties must be filesystem independent. setattr(FakePathlibModule.PureWindowsPath, parser_name, fake_os_windows.path) def _wrap_strfunc(fake_fct, original_fct): @functools.wraps(fake_fct) def _wrapped(pathobj, args, kwargs): fs: FakeFilesystem = pathobj.filesystem if fs.patcher: if is_called_from_skipped_module( skip_names=fs.patcher.skip_names, case_sensitive=fs.is_case_sensitive, ): return original_fct(str(pathobj), args, *kwargs) return fake_fct(fs, str(pathobj), args, *kwargs) return staticmethod(_wrapped) def _wrap_binary_strfunc(fake_fct, original_fct): @functools.wraps(fake_fct) def _wrapped(pathobj1, pathobj2, args): fs: FakeFilesystem = pathobj1.filesystem if fs.patcher: if is_called_from_skipped_module( skip_names=fs.patcher.skip_names, case_sensitive=fs.is_case_sensitive, ): return original_fct(str(pathobj1), str(pathobj2), args) return fake_fct(fs, str(pathobj1), str(pathobj2), args) return staticmethod(_wrapped) def _wrap_binary_strfunc_reverse(fake_fct, original_fct): @functools.wraps(fake_fct) def _wrapped(pathobj1, pathobj2, args): fs: FakeFilesystem = pathobj2.filesystem if fs.patcher: if is_called_from_skipped_module( skip_names=fs.patcher.skip_names, case_sensitive=fs.is_case_sensitive, ): return original_fct(str(pathobj2), str(pathobj1), args) return fake_fct(fs, str(pathobj2), str(pathobj1), args) return staticmethod(_wrapped) try: accessor = pathlib._Accessor # type: ignore[attr-defined] except AttributeError: accessor = object class _FakeAccessor(accessor): # type: ignore[valid-type, misc] """Accessor which forwards some of the functions to FakeFilesystem methods. """ stat = _wrap_strfunc(FakeFilesystem.stat, os.stat) lstat = _wrap_strfunc( lambda fs, path: FakeFilesystem.stat(fs, path, follow_symlinks=False), os.lstat ) listdir = _wrap_strfunc(FakeFilesystem.listdir, os.listdir) scandir = _wrap_strfunc(fake_scandir.scandir, os.scandir) if hasattr(os, "lchmod"): lchmod = _wrap_strfunc( lambda fs, path, mode: FakeFilesystem.chmod( fs, path, mode, follow_symlinks=False ), os.lchmod, ) else: def lchmod(self, pathobj, args, *kwargs): """Raises not implemented for Windows systems.""" raise NotImplementedError("lchmod() not available on this system") def chmod(self, pathobj, args, *kwargs): if "follow_symlinks" in kwargs: if sys.version_info < (3, 10): raise TypeError( "chmod() got an unexpected keyword argument 'follow_symlinks'" ) if not kwargs["follow_symlinks"] and ( os.chmod not in os.supports_follow_symlinks or IS_PYPY ): raise NotImplementedError( "`follow_symlinks` for chmod() is not available on this system" ) return pathobj.filesystem.chmod(str(pathobj), args, *kwargs) mkdir = _wrap_strfunc(FakeFilesystem.makedir, os.mkdir) unlink = _wrap_strfunc(FakeFilesystem.remove, os.unlink) rmdir = _wrap_strfunc(FakeFilesystem.rmdir, os.rmdir) rename = _wrap_binary_strfunc(FakeFilesystem.rename, os.rename) replace = _wrap_binary_strfunc( lambda fs, old_path, new_path: FakeFilesystem.rename( fs, old_path, new_path, force_replace=True ), os.replace, ) symlink = _wrap_binary_strfunc_reverse( lambda fs, fpath, target, target_is_dir: FakeFilesystem.create_symlink( fs, fpath, target, create_missing_dirs=False ), os.symlink, ) if (3, 8) <= sys.version_info: link_to = _wrap_binary_strfunc( lambda fs, file_path, link_target: FakeFilesystem.link( fs, file_path, link_target ), os.link, ) if sys.version_info >= (3, 10): link = _wrap_binary_strfunc( lambda fs, file_path, link_target: FakeFilesystem.link( fs, file_path, link_target ), os.link, ) # this will use the fake filesystem because os is patched def getcwd(self): return os.getcwd() readlink = _wrap_strfunc(FakeFilesystem.readlink, os.readlink) utime = _wrap_strfunc(FakeFilesystem.utime, os.utime) _fake_accessor = _FakeAccessor() if sys.version_info < (3, 12): flavour = pathlib._Flavour # type: ignore[attr-defined] class _FakeFlavour(flavour): # type: ignore[valid-type, misc] """Fake Flavour implementation used by PurePath and _Flavour""" filesystem = None ext_namespace_prefix = "\\\\?\\" drive_letters = {chr(x) for x in range(ord("a"), ord("z") + 1)} \| { chr(x) for x in range(ord("A"), ord("Z") + 1) } def __init__(self, filesystem): self.filesystem = filesystem super().__init__() @staticmethod def _split_extended_path(path, ext_prefix=ext_namespace_prefix): prefix = "" if path.startswith(ext_prefix): prefix = path[:4] path = path[4:] if path.startswith("UNC\\"): prefix += path[:3] path = "\\" + path[3:] return prefix, path def _splitroot_with_drive(self, path, sep): first = path[0:1] second = path[1:2] if second == sep and first == sep: # extended paths should also disable the collapsing of "." # components (according to MSDN docs). prefix, path = self._split_extended_path(path) first = path[0:1] second = path[1:2] else: prefix = "" third = path[2:3] if second == sep and first == sep and third != sep: # is a UNC path: # vvvvvvvvvvvvvvvvvvvvv root # \\machine\mountpoint\directory\etc\... # directory ^^^^^^^^^^^^^^ index = path.find(sep, 2) if index != -1: index2 = path.find(sep, index + 1) # a UNC path can't have two slashes in a row # (after the initial two) if index2 != index + 1: if index2 == -1: index2 = len(path) if prefix: return prefix + path[1:index2], sep, path[index2 + 1 :] return path[:index2], sep, path[index2 + 1 :] drv = root = "" if second == ":" and first in self.drive_letters: drv = path[:2] path = path[2:] first = third if first == sep: root = first path = path.lstrip(sep) return prefix + drv, root, path @staticmethod def _splitroot_posix(path, sep): if path and path[0] == sep: stripped_part = path.lstrip(sep) if len(path) - len(stripped_part) == 2: return "", sep 2, stripped_part return "", sep, stripped_part else: return "", "", path def splitroot(self, path, sep=None): """Split path into drive, root and rest.""" is_windows = isinstance(self, _FakeWindowsFlavour) if sep is None: if is_windows == self.filesystem.is_windows_fs: sep = self.filesystem.path_separator else: sep = self.sep if is_windows: return self._splitroot_with_drive(path, sep) return self._splitroot_posix(path, sep) def casefold(self, path): """Return the lower-case version of s for a Windows filesystem.""" if self.filesystem.is_windows_fs: return path.lower() return path def casefold_parts(self, parts): """Return the lower-case version of parts for a Windows filesystem.""" if self.filesystem.is_windows_fs: return [p.lower() for p in parts] return parts def _resolve_posix(self, path, strict): sep = self.sep seen = {} def _resolve(path, rest): if rest.startswith(sep): path = "" for name in rest.split(sep): if not name or name == ".": # current dir continue if name == "..": # parent dir path, _, _ = path.rpartition(sep) continue newpath = path + sep + name if newpath in seen: # Already seen this path path = seen[newpath] if path is not None: # use cached value continue # The symlink is not resolved, so we must have # a symlink loop. raise RuntimeError("Symlink loop from %r" % newpath) # Resolve the symbolic link try: target = self.filesystem.readlink(newpath) except OSError as e: if e.errno != errno.EINVAL and strict: raise # Not a symlink, or non-strict mode. We just leave the path # untouched. path = newpath else: seen[newpath] = None # not resolved symlink path = _resolve(path, target) seen[newpath] = path # resolved symlink return path # NOTE: according to POSIX, getcwd() cannot contain path components # which are symlinks. base = "" if path.is_absolute() else self.filesystem.cwd return _resolve(base, str(path)) or sep def _resolve_windows(self, path, strict): path = str(path) if not path: return os.getcwd() previous_s = None if strict: if not self.filesystem.exists(path): self.filesystem.raise_os_error(errno.ENOENT, path) return self.filesystem.resolve_path(path) else: while True: try: path = self.filesystem.resolve_path(path) except OSError: previous_s = path path = self.filesystem.splitpath(path)[0] else: if previous_s is None: return path return self.filesystem.joinpaths( path, os.path.basename(previous_s) ) def resolve(self, path, strict): """Make the path absolute, resolving any symlinks.""" if self.filesystem.is_windows_fs: return self._resolve_windows(path, strict) return self._resolve_posix(path, strict) def gethomedir(self, username): """Return the home directory of the current user.""" if not username: try: return os.environ["HOME"] except KeyError: import pwd return pwd.getpwuid(os.getuid()).pw_dir else: import pwd try: return pwd.getpwnam(username).pw_dir except KeyError: raise RuntimeError( "Can't determine home directory for %r" % username ) class _FakeWindowsFlavour(_FakeFlavour): """Flavour used by PureWindowsPath with some Windows specific implementations independent of FakeFilesystem properties. """ sep = "\\" altsep = "/" has_drv = True pathmod = ntpath def is_reserved(self, parts): """Return `True` if the path is considered reserved under Windows.""" # NOTE: the rules for reserved names seem somewhat complicated # (e.g. r"..\NUL" is reserved but not r"foo\NUL"). # We err on the side of caution and return True for paths which are # not considered reserved by Windows. if not parts: return False if self.filesystem.is_windows_fs and parts[0].startswith("\\\\"): # UNC paths are never reserved return False return parts[-1].partition(".")[0].upper() in _WIN_RESERVED_NAMES def make_uri(self, path): """Return a file URI for the given path""" # Under Windows, file URIs use the UTF-8 encoding. # original version, not faked drive = path.drive if len(drive) == 2 and drive[1] == ":": # It's a path on a local drive => 'file:///c:/a/b' rest = path.as_posix()[2:].lstrip("/") return "file:///{}/{}".format( drive, urlquote_from_bytes(rest.encode("utf-8")), ) else: # It's a path on a network drive => 'file://host/share/a/b' return "file:" + urlquote_from_bytes(path.as_posix().encode("utf-8")) def gethomedir(self, username): """Return the home directory of the current user.""" # original version, not faked if "HOME" in os.environ: userhome = os.environ["HOME"] elif "USERPROFILE" in os.environ: userhome = os.environ["USERPROFILE"] elif "HOMEPATH" in os.environ: try: drv = os.environ["HOMEDRIVE"] except KeyError: drv = "" userhome = drv + os.environ["HOMEPATH"] else: raise RuntimeError("Can't determine home directory") if username: # Try to guess user home directory. By default all users # directories are located in the same place and are named by # corresponding usernames. If current user home directory points # to nonstandard place, this guess is likely wrong. if os.environ["USERNAME"] != username: drv, root, parts = self.parse_parts((userhome,)) if parts[-1] != os.environ["USERNAME"]: raise RuntimeError( "Can't determine home directory for %r" % username ) parts[-1] = username if drv or root: userhome = drv + root + self.join(parts[1:]) else: userhome = self.join(parts) return userhome def compile_pattern(self, pattern): return re.compile(fnmatch.translate(pattern), re.IGNORECASE).fullmatch class _FakePosixFlavour(_FakeFlavour): """Flavour used by PurePosixPath with some Unix specific implementations independent of FakeFilesystem properties. """ sep = "/" altsep: Optional[str] = None has_drv = False pathmod = posixpath def is_reserved(self, parts): return False def make_uri(self, path): # We represent the path using the local filesystem encoding, # for portability to other applications. bpath = bytes(path) return "file://" + urlquote_from_bytes(bpath) def gethomedir(self, username): # original version, not faked if not username: try: return os.environ["HOME"] except KeyError: import pwd return pwd.getpwuid(os.getuid()).pw_dir else: import pwd try: return pwd.getpwnam(username).pw_dir except KeyError: raise RuntimeError( "Can't determine home directory for %r" % username ) def compile_pattern(self, pattern): return re.compile(fnmatch.translate(pattern)).fullmatch else: # Python >= 3.12 class FakePosixPathModule(FakePathModule): def __init__(self, filesystem: "FakeFilesystem", os_module: "FakeOsModule"): super().__init__(filesystem, os_module) with self.filesystem.use_fs_type(FSType.POSIX): self.reset(self.filesystem) class FakeWindowsPathModule(FakePathModule): def __init__(self, filesystem: "FakeFilesystem", os_module: "FakeOsModule"): super().__init__(filesystem, os_module) with self.filesystem.use_fs_type(FSType.WINDOWS): self.reset(self.filesystem) def with_fs_type(f: Callable, fs_type: FSType) -> Callable: """Decorator used for fake_path methods to ensure that the correct filesystem type is used.""" @functools.wraps(f) def wrapped(self, args, kwargs): with self.filesystem.use_fs_type(fs_type): return f(self, args, *kwargs) return wrapped # decorate all public functions to use the correct fs type for fct_name in FakePathModule.dir(): fn = getattr(FakePathModule, fct_name) setattr(FakeWindowsPathModule, fct_name, with_fs_type(fn, FSType.WINDOWS)) setattr(FakePosixPathModule, fct_name, with_fs_type(fn, FSType.POSIX)) class FakePath(pathlib.Path): """Replacement for pathlib.Path. Reimplement some methods to use fake filesystem. The rest of the methods work as they are, as they will use the fake accessor. New in pyfakefs 3.0. """ # the underlying fake filesystem filesystem = None skip_names: List[str] = [] def __new__(cls, args, *kwargs): """Creates the correct subclass based on OS.""" if cls is FakePathlibModule.Path: cls = ( FakePathlibModule.WindowsPath if cls.filesystem.is_windows_fs else FakePathlibModule.PosixPath ) if sys.version_info < (3, 12): return cls._from_parts(args) # pytype: disable=attribute-error else: return object.__new__(cls) if sys.version_info[:2] == (3, 10): # Overwritten class methods to call _init to set the fake accessor, # which is not done in Python 3.10, and not needed from Python 3.11 on @classmethod def _from_parts(cls, args): self = object.__new__(cls) self._init() drv, root, parts = self._parse_args(args) # pytype: disable=attribute-error self._drv = drv self._root = root self._parts = parts return self @classmethod def _from_parsed_parts(cls, drv, root, parts): self = object.__new__(cls) self._drv = drv self._root = root self._parts = parts self._init() return self if sys.version_info < (3, 11): def _init(self, template=None): """Initializer called from base class.""" # only needed until Python 3.10 self._accessor = _fake_accessor # only needed until Python 3.8 self._closed = False def _path(self): """Returns the underlying path string as used by the fake filesystem. """ return str(self) @classmethod def cwd(cls): """Return a new path pointing to the current working directory (as returned by os.getcwd()). """ return cls(cls.filesystem.cwd) if sys.version_info < (3, 12): # in 3.12, we can use the pathlib implementation def resolve(self, strict=None): """Make the path absolute, resolving all symlinks on the way and also normalizing it (for example turning slashes into backslashes under Windows). Args: strict: If `False` (default) no exception is raised if the path does not exist. Raises: OSError: if the path doesn't exist (strict=True) """ if strict is None: strict = False self._raise_on_closed() path = self._flavour.resolve( self, strict=strict ) # pytype: disable=attribute-error if path is None: self.stat() path = str(self.absolute()) path = self.filesystem.absnormpath(path) return FakePath(path) def open(self, mode="r", buffering=-1, encoding=None, errors=None, newline=None): """Open the file pointed by this path and return a fake file object. Raises: OSError: if the target object is a directory, the path is invalid or permission is denied. """ self._raise_on_closed() return fake_open( self.filesystem, self.skip_names, self._path(), mode, buffering, encoding, errors, newline, ) def read_bytes(self): """Open the fake file in bytes mode, read it, and close the file. Raises: OSError: if the target object is a directory, the path is invalid or permission is denied. """ with fake_open( self.filesystem, self.skip_names, self._path(), mode="rb", ) as f: return f.read() def read_text(self, encoding=None, errors=None): """ Open the fake file in text mode, read it, and close the file. """ with fake_open( self.filesystem, self.skip_names, self._path(), mode="r", encoding=encoding, errors=errors, ) as f: return f.read() def write_bytes(self, data): """Open the fake file in bytes mode, write to it, and close the file. Args: data: the bytes to be written Raises: OSError: if the target object is a directory, the path is invalid or permission is denied. """ # type-check for the buffer interface before truncating the file view = memoryview(data) with fake_open( self.filesystem, self.skip_names, self._path(), mode="wb", ) as f: return f.write(view) def write_text(self, data, encoding=None, errors=None, newline=None): """Open the fake file in text mode, write to it, and close the file. Args: data: the string to be written encoding: the encoding used for the string; if not given, the default locale encoding is used errors: (str) Defines how encoding errors are handled. newline: Controls universal newlines, passed to stream object. New in Python 3.10. Raises: TypeError: if data is not of type 'str'. OSError: if the target object is a directory, the path is invalid or permission is denied. """ if not isinstance(data, str): raise TypeError("data must be str, not %s" % data.__class__.__name__) if newline is not None and sys.version_info < (3, 10): raise TypeError("write_text() got an unexpected keyword argument 'newline'") with fake_open( self.filesystem, self.skip_names, self._path(), mode="w", encoding=encoding, errors=errors, newline=newline, ) as f: return f.write(data) @classmethod def home(cls): """Return a new path pointing to the user's home directory (as returned by os.path.expanduser('~')). """ home = os.path.expanduser("~") if cls.filesystem.is_windows_fs != (os.name == "nt"): username = os.path.split(home)[1] if cls.filesystem.is_windows_fs: home = os.path.join("C:", "Users", username) else: home = os.path.join("home", username) if not cls.filesystem.exists(home): cls.filesystem.create_dir(home) return cls(home.replace(os.sep, cls.filesystem.path_separator)) def samefile(self, other_path): """Return whether other_path is the same or not as this file (as returned by os.path.samefile()). Args: other_path: A path object or string of the file object to be compared with Raises: OSError: if the filesystem object doesn't exist. """ st = self.stat() try: other_st = other_path.stat() except AttributeError: other_st = self.filesystem.stat(other_path) return st.st_ino == other_st.st_ino and st.st_dev == other_st.st_dev def expanduser(self): """Return a new path with expanded ~ and ~user constructs (as returned by os.path.expanduser) """ return FakePath( os.path.expanduser(self._path()).replace( os.path.sep, self.filesystem.path_separator ) ) def _raise_on_closed(self): if sys.version_info < (3, 9) and self._closed: self._raise_closed() def touch(self, mode=0o666, exist_ok=True): """Create a fake file for the path with the given access mode, if it doesn't exist. Args: mode: the file mode for the file if it does not exist exist_ok: if the file already exists and this is `True`, nothing happens, otherwise `FileExistError` is raised Raises: FileExistsError: if the file exists and ``exits_ok`` is `False`. """ self._raise_on_closed() if self.exists(): if exist_ok: self.filesystem.utime(self._path(), times=None) else: self.filesystem.raise_os_error(errno.EEXIST, self._path()) else: fake_file = self.open("w", encoding="utf8") fake_file.close() self.chmod(mode) if sys.version_info >= (3, 14): def _copy_from_file(self, source, preserve_metadata=False): if sys.platform == "win32": # do not use the optimized version that uses OS functions in Windows if hasattr(self, "_copy_from_file_fallback"): return self._copy_from_file_fallback(source, preserve_metadata) else: return super()._copy_from_file(source, preserve_metadata) # type: ignore[attr-defined] else: # make pathlib think that no system calls are available so that it # will fall back to pure Python calls pathlib_os = pathlib.pathlib_module._os # type: ignore[attr-defined] old_fcopyfile = pathlib_os._fcopyfile pathlib_os._fcopyfile = None old_copy_file_range = pathlib_os._copy_file_range # type: ignore[attr-defined] pathlib_os._copy_file_range = None old_ficlone = pathlib_os._ficlone # type: ignore[attr-defined] pathlib_os._ficlone = None old_sendfile = pathlib_os._sendfile # type: ignore[attr-defined] pathlib_os._sendfile = None try: return super()._copy_from_file(source, preserve_metadata) # type: ignore[attr-defined] finally: pathlib_os._fcopyfile = old_fcopyfile pathlib_os._copy_file_range = old_copy_file_range pathlib_os._ficlone = old_ficlone pathlib_os._sendfile = old_sendfile def _warn_is_reserved_deprecated(): if sys.version_info >= (3, 13): warnings.warn( "pathlib.PurePath.is_reserved() is deprecated and scheduled " "for removal in Python 3.15. Use os.path.isreserved() to detect " "reserved paths on Windows.", DeprecationWarning, ) class FakePathlibModule: """Uses FakeFilesystem to provide a fake pathlib module replacement. Automatically created if using `fake_filesystem_unittest.TestCase`, the `fs` fixture, the `patchfs` decorator, or directly the `Patcher`. """ def __init__(self, filesystem, from_patcher=False): """ Initializes the module with the given filesystem. Args: filesystem: FakeFilesystem used to provide file system information from_patcher: If `False` (the default), `pathlib.os` will be manually patched using `FakeOsModule`. This allows to instantiate the class manually for a test. Will be set to `True` if instantiated from `Patcher`. """ init_module(filesystem) self.pathlib_module = pathlib self._os = None self._os_patcher = None if not from_patcher: self.patch_os_module() def patch_os_module(self): if sys.version_info >= (3, 11) and not isinstance(os, FakeOsModule): self._os = FakeOsModule(FakePath.filesystem) pathlib_os = ( "pathlib._local.os" if sys.version_info[:2] == (3, 13) else "pathlib.os" ) self._os_patcher = mock.patch(pathlib_os, self._os) self._os_patcher.start() def __del__(self): if self._os_patcher is not None: self._os_patcher.stop() class PurePosixPath(PurePath): """A subclass of PurePath that represents non-Windows filesystem paths""" __slots__ = () if sys.version_info >= (3, 12): def is_reserved(self): _warn_is_reserved_deprecated() return False def is_absolute(self): with os.path.filesystem.use_fs_type(FSType.POSIX): # type: ignore[module-attr] return os.path.isabs(self) def joinpath(self, pathsegments): with os.path.filesystem.use_fs_type(FSType.POSIX): # type: ignore[module-attr] return super().joinpath(pathsegments) class PureWindowsPath(PurePath): """A subclass of PurePath that represents Windows filesystem paths""" __slots__ = () if sys.version_info >= (3, 12): """These are reimplemented because the PurePath implementation checks the flavour against ntpath/posixpath. """ def is_reserved(self): _warn_is_reserved_deprecated() if sys.version_info < (3, 13): if not self._tail or self._tail[0].startswith("\\\\"): # UNC paths are never reserved. return False name = ( self._tail[-1].partition(".")[0].partition(":")[0].rstrip(" ") ) return name.upper() in _WIN_RESERVED_NAMES with os.path.filesystem.use_fs_type(FSType.WINDOWS): # type: ignore[module-attr] return os.path.isreserved(self) def is_absolute(self): with os.path.filesystem.use_fs_type(FSType.WINDOWS): return bool(self.drive and self.root) class WindowsPath(FakePath, PureWindowsPath): """A subclass of Path and PureWindowsPath that represents concrete Windows filesystem paths. """ __slots__ = () def owner(self): raise NotImplementedError("Path.owner() is unsupported on this system") def group(self): raise NotImplementedError("Path.group() is unsupported on this system") def is_mount(self): raise NotImplementedError("Path.is_mount() is unsupported on this system") class PosixPath(FakePath, PurePosixPath): """A subclass of Path and PurePosixPath that represents concrete non-Windows filesystem paths. """ __slots__ = () def owner(self): """Return the username of the file owner. It is assumed that `st_uid` is related to a real user, otherwise `KeyError` is raised. """ import pwd return pwd.getpwuid(self.stat().st_uid).pw_name def group(self): """Return the group name of the file group. It is assumed that `st_gid` is related to a real group, otherwise `KeyError` is raised. """ import grp return grp.getgrgid(self.stat().st_gid).gr_name if sys.version_info >= (3, 14): # in Python 3.14, case-sensitivity is checked using an is-check # (self.parser is posixpath) if not given, which we cannot fake # therefore we already provide the case sensitivity under Posix def glob(self, pattern, , case_sensitive=None, recurse_symlinks=False): if case_sensitive is None: case_sensitive = True return super().glob( # pytype: disable=wrong-keyword-args pattern, case_sensitive=case_sensitive, recurse_symlinks=recurse_symlinks, ) Path = FakePath def __getattr__(self, name): """Forwards any unfaked calls to the standard pathlib module.""" return getattr(self.pathlib_module, name) class FakePathlibPathModule: """Patches `pathlib.Path` by passing all calls to FakePathlibModule.""" fake_pathlib = None def __init__(self, filesystem=None, from_patcher=False): if self.fake_pathlib is None: self.__class__.fake_pathlib = FakePathlibModule(filesystem, from_patcher) @property def skip_names(self): return [] # not used, here to allow a setter @skip_names.setter def skip_names(self, value): # this is set from the patcher and passed to the fake Path class self.fake_pathlib.Path.skip_names = value def __call__(self, args, kwargs): return self.fake_pathlib.Path(args, *kwargs) def __getattr__(self, name): return getattr(self.fake_pathlib.Path, name) @classmethod def __instancecheck__(cls, instance): # fake the inheritance to pass isinstance checks - see #666 return isinstance(instance, PurePath) class RealPath(pathlib.Path): """Replacement for `pathlib.Path` if it shall not be faked. Needed because `Path` in `pathlib` is always faked, even if `pathlib` itself is not. """ if sys.version_info < (3, 12): _flavour = ( pathlib._WindowsFlavour() # type:ignore if os.name == "nt" else pathlib._PosixFlavour() # type:ignore ) # type:ignore elif sys.version_info < (3, 13): _flavour = ntpath if os.name == "nt" else posixpath else: parser = ntpath if os.name == "nt" else posixpath def __new__(cls, args, *kwargs): """Creates the correct subclass based on OS.""" if cls is RealPathlibModule.Path: cls = ( RealPathlibModule.WindowsPath # pytype: disable=attribute-error if os.name == "nt" else RealPathlibModule.PosixPath # pytype: disable=attribute-error ) if sys.version_info < (3, 12): return cls._from_parts(args) # pytype: disable=attribute-error else: return object.__new__(cls) if sys.version_info > (3, 10): def with_original_os(f: Callable) -> Callable: """Decorator used for real pathlib Path methods to ensure that real os functions instead of faked ones are used.""" @functools.wraps(f) def wrapped(args, *kwargs): with use_original_os(): return f(args, *kwargs) return wrapped for fct_name, fn in inspect.getmembers(RealPath, inspect.isfunction): if not fct_name.startswith("__"): setattr(RealPath, fct_name, with_original_os(fn)) class RealPathlibPathModule: """Patches `pathlib.Path` by passing all calls to RealPathlibModule.""" real_pathlib = None @classmethod def __instancecheck__(cls, instance): # as we cannot derive from pathlib.Path, we fake # the inheritance to pass isinstance checks - see #666 return isinstance(instance, PurePath) def __init__(self): if self.real_pathlib is None: self.__class__.real_pathlib = RealPathlibModule() def __call__(self, args, *kwargs): return RealPath(args, **kwargs) def __getattr__(self, name): return getattr(self.real_pathlib.Path, name) class RealPathlibModule: """Used to replace `pathlib` for skipped modules. As the original `pathlib` is always patched to use the fake path, we need to provide a version which does not do this. """ def __init__(self): self._pathlib_module = pathlib class PurePosixPath(PurePath): """A subclass of PurePath, that represents Posix filesystem paths""" __slots__ = () class PureWindowsPath(PurePath): """A subclass of PurePath, that represents Windows filesystem paths""" __slots__ = () if sys.platform == "win32": class WindowsPath(RealPath, PureWindowsPath): """A subclass of Path and PureWindowsPath that represents concrete Windows filesystem paths. """ __slots__ = () else: class PosixPath(RealPath, PurePosixPath): """A subclass of Path and PurePosixPath that represents concrete non-Windows filesystem paths. """ __slots__ = () Path = RealPath def __getattr__(self, name): """Forwards any unfaked calls to the standard pathlib module.""" return getattr(self._pathlib_module, name)	41.79 KB	2026-01-20 12:58
📄 fake_scandir.py # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """A fake implementation for the `scandir` function working with FakeFilesystem. Works with both the function integrated into the `os` module since Python 3.5 and the standalone function available in the standalone `scandir` python package. """ import os import sys from pyfakefs.helpers import to_string, make_string_path class DirEntry(os.PathLike): """Emulates os.DirEntry. Note that we did not enforce keyword only arguments.""" def __init__(self, filesystem): """Initialize the dir entry with unset values. Args: filesystem: the fake filesystem used for implementation. """ self._filesystem = filesystem self.name = "" self.path = "" self._abspath = "" self._inode = None self._islink = False self._isdir = False self._statresult = None self._statresult_symlink = None def inode(self): """Return the inode number of the entry.""" if self._inode is None: self.stat(follow_symlinks=False) return self._inode def is_dir(self, follow_symlinks=True): """Return `True` if this entry is a directory entry. Args: follow_symlinks: If `True`, also return `True` if this entry is a symlink pointing to a directory. Returns: `True` if this entry is an existing directory entry, or if ``follow_symlinks`` is set, and this entry points to an existing directory entry. """ return self._isdir and (follow_symlinks or not self._islink) def is_file(self, follow_symlinks=True): """Return `True` if this entry is a regular file entry. Args: follow_symlinks: If `True`, also return `True` if this entry is a symlink pointing to a regular file. Returns: `True` if this entry is an existing file entry, or if ``follow_symlinks`` is set, and this entry points to an existing file entry. """ return not self._isdir and (follow_symlinks or not self._islink) def is_symlink(self): """Return `True` if this entry is a symbolic link (even if broken).""" return self._islink def stat(self, follow_symlinks=True): """Return a `stat_result` object for this entry. Args: follow_symlinks: If `False` and the entry is a symlink, return the result for the symlink, otherwise for the object it points to. """ if follow_symlinks: if self._statresult_symlink is None: file_object = self._filesystem.resolve(self._abspath) self._statresult_symlink = file_object.stat_result.copy() if self._filesystem.is_windows_fs: self._statresult_symlink.st_nlink = 0 return self._statresult_symlink if self._statresult is None: file_object = self._filesystem.lresolve(self._abspath) self._inode = file_object.st_ino self._statresult = file_object.stat_result.copy() if self._filesystem.is_windows_fs: self._statresult.st_nlink = 0 return self._statresult def __fspath__(self): return self.path if sys.version_info >= (3, 12): def is_junction(self) -> bool: """Return `True` if this entry is a junction. Junctions are not a part of posix semantic.""" if not self._filesystem.is_windows_fs: return False file_object = self._filesystem.resolve(self._abspath) return file_object.is_junction class ScanDirIter: """Iterator for DirEntry objects returned from `scandir()` function.""" def __init__(self, filesystem, path): self.filesystem = filesystem if isinstance(path, int): if self.filesystem.is_windows_fs: raise NotImplementedError( "scandir does not support file descriptor path argument" ) self.abspath = self.filesystem.absnormpath( self.filesystem.get_open_file(path).get_object().path ) self.path = "" self.entry_iter = iter(tuple()) else: if path is None: path = "." path = make_string_path(path) self.abspath = self.filesystem.absnormpath(path) self.path = to_string(path) entries = self.filesystem.confirmdir(self.abspath, check_exe_perm=False).entries self.entry_iter = iter(tuple(entries)) def __iter__(self): return self def __next__(self): entry = self.entry_iter.__next__() dir_entry = DirEntry(self.filesystem) dir_entry.name = entry dir_entry.path = self.filesystem.joinpaths(self.path, dir_entry.name) dir_entry._abspath = self.filesystem.joinpaths(self.abspath, dir_entry.name) dir_entry._isdir = self.filesystem.isdir(dir_entry._abspath) dir_entry._islink = self.filesystem.islink(dir_entry._abspath) return dir_entry def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def close(self): pass def scandir(filesystem, path=""): """Return an iterator of DirEntry objects corresponding to the entries in the directory given by path. Args: filesystem: The fake filesystem used for implementation path: Path to the target directory within the fake filesystem. Returns: an iterator to an unsorted list of os.DirEntry objects for each entry in path. Raises: OSError: if the target is not a directory. """ return ScanDirIter(filesystem, path) def _classify_directory_contents(filesystem, root): """Classify contents of a directory as files/directories. Args: filesystem: The fake filesystem used for implementation root: (str) Directory to examine. Returns: (tuple) A tuple consisting of three values: the directory examined, a list containing all of the directory entries, and a list containing all of the non-directory entries. (This is the same format as returned by the `os.walk` generator.) Raises: Nothing on its own, but be ready to catch exceptions generated by underlying mechanisms like `os.listdir`. """ dirs = [] files = [] for entry in filesystem.listdir(root): if filesystem.isdir(filesystem.joinpaths(root, entry)): dirs.append(entry) else: files.append(entry) return root, dirs, files def walk(filesystem, top, topdown=True, onerror=None, followlinks=False): """Perform an os.walk operation over the fake filesystem. Args: filesystem: The fake filesystem used for implementation top: The root directory from which to begin walk. topdown: Determines whether to return the tuples with the root as the first entry (`True`) or as the last, after all the child directory tuples (`False`). onerror: If not `None`, function which will be called to handle the `os.error` instance provided when `os.listdir()` fails. followlinks: If `True`, symbolic links are followed. Yields: (path, directories, nondirectories) for top and each of its subdirectories. See the documentation for the builtin os module for further details. """ def do_walk(top_dir, top_most=False): if not top_most and not followlinks and filesystem.islink(top_dir): return try: top_contents = _classify_directory_contents(filesystem, top_dir) except OSError as exc: top_contents = None if onerror is not None: onerror(exc) if top_contents is not None: if topdown: yield top_contents for directory in top_contents[1]: path = filesystem.joinpaths(top_dir, directory) if not followlinks and filesystem.islink(path): continue yield from do_walk(path) if not topdown: yield top_contents return do_walk(make_string_path(to_string(top)), top_most=True)	8.77 KB	2026-01-20 12:58
📄 helpers.py # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Helper classes used for the fake file system implementation and may be useful for ``pyfakefs`` users.""" import ctypes import importlib import io import locale import os import platform import stat import sys import sysconfig import time import traceback from collections import namedtuple from copy import copy from dataclasses import dataclass from enum import Enum from stat import S_IFLNK from typing import Union, Optional, Any, AnyStr, overload, cast AnyString = Union[str, bytes] AnyPath = Union[AnyStr, os.PathLike] IS_PYPY = platform.python_implementation() == "PyPy" IS_WIN = sys.platform == "win32" IN_DOCKER = os.path.exists("/.dockerenv") PERM_READ = 0o400 # Read permission bit. PERM_WRITE = 0o200 # Write permission bit. PERM_EXE = 0o100 # Execute permission bit. PERM_DEF = 0o777 # Default permission bits. PERM_DEF_FILE = 0o666 # Default permission bits (regular file) PERM_ALL = 0o7777 # All permission bits. STDLIB_PATH = os.path.realpath(sysconfig.get_path("stdlib")) PYFAKEFS_PATH = os.path.dirname(__file__) PYFAKEFS_TEST_PATHS = [ os.path.join(PYFAKEFS_PATH, "tests"), os.path.join(PYFAKEFS_PATH, "pytest_tests"), ] _OpenModes = namedtuple( "_OpenModes", "must_exist can_read can_write truncate append must_not_exist", ) if sys.platform == "win32": fake_id = 0 if ctypes.windll.shell32.IsUserAnAdmin() else 1 USER_ID = fake_id GROUP_ID = fake_id else: USER_ID = os.getuid() GROUP_ID = os.getgid() def get_uid() -> int: """Get the global user id. Same as ``os.getuid()``""" return USER_ID def set_uid(uid: int) -> None: """Set the global user id. This is used as st_uid for new files and to differentiate between a normal user and the root user (uid 0). For the root user, some permission restrictions are ignored. Args: uid: (int) the user ID of the user calling the file system functions. """ global USER_ID USER_ID = uid def get_gid() -> int: """Get the global group id. Same as ``os.getgid()``""" return GROUP_ID def set_gid(gid: int) -> None: """Set the global group id. This is only used to set st_gid for new files, no permission checks are performed. Args: gid: (int) the group ID of the user calling the file system functions. """ global GROUP_ID GROUP_ID = gid def reset_ids() -> None: """Set the global user ID and group ID back to default values.""" if sys.platform == "win32": reset_id = 0 if ctypes.windll.shell32.IsUserAnAdmin() else 1 set_uid(reset_id) set_gid(reset_id) else: set_uid(os.getuid()) set_gid(os.getgid()) def is_root() -> bool: """Return `True` if the current user is the root user.""" return USER_ID == 0 def is_int_type(val: Any) -> bool: """Return `True` if `val` is of integer type.""" return isinstance(val, int) def is_byte_string(val: Any) -> bool: """Return `True` if ``val`` is a bytes-like object, `False` for a Unicode string.""" return not hasattr(val, "encode") def is_unicode_string(val: Any) -> bool: """Return `True` if ``val`` is a Unicode string, `False` for a bytes-like object.""" return hasattr(val, "encode") def get_locale_encoding(): if sys.version_info >= (3, 11): return locale.getencoding() return locale.getpreferredencoding(False) @overload def make_string_path(dir_name: AnyStr) -> AnyStr: ... @overload def make_string_path(dir_name: os.PathLike) -> str: ... def make_string_path(dir_name: AnyPath) -> AnyStr: # type: ignore[type-var] return cast(AnyStr, os.fspath(dir_name)) # pytype: disable=invalid-annotation def to_string(path: Union[AnyStr, Union[str, bytes]]) -> str: """Return the string representation of a byte string using the preferred encoding, or the string itself if path is a str.""" if isinstance(path, bytes): return path.decode(get_locale_encoding()) return path def to_bytes(path: Union[AnyStr, Union[str, bytes]]) -> bytes: """Return the bytes representation of a string using the preferred encoding, or the byte string itself if path is a byte string.""" if isinstance(path, str): return bytes(path, get_locale_encoding()) return path def join_strings(s1: AnyStr, s2: AnyStr) -> AnyStr: """This is a bit of a hack to satisfy mypy - may be refactored.""" return s1 + s2 def real_encoding(encoding: Optional[str]) -> Optional[str]: """Since Python 3.10, the new function ``io.text_encoding`` returns "locale" as the encoding if none is defined. This will be handled as no encoding in pyfakefs.""" if sys.version_info >= (3, 10): return encoding if encoding != "locale" else None return encoding def now(): return time.time() @overload def matching_string(matched: bytes, string: AnyStr) -> bytes: ... @overload def matching_string(matched: str, string: AnyStr) -> str: ... @overload def matching_string(matched: AnyStr, string: None) -> None: ... def matching_string( # type: ignore[misc] matched: AnyStr, string: Optional[AnyStr] ) -> Optional[AnyString]: """Return the string as byte or unicode depending on the type of matched, assuming string is an ASCII string. """ if string is None: return string if isinstance(matched, bytes) and isinstance(string, str): return string.encode(get_locale_encoding()) return string # pytype: disable=bad-return-type @dataclass class FSProperties: sep: str altsep: Optional[str] pathsep: str linesep: str devnull: str # pure POSIX file system properties, for use with PosixPath POSIX_PROPERTIES = FSProperties( sep="/", altsep=None, pathsep=":", linesep="\n", devnull="/dev/null", ) # pure Windows file system properties, for use with WindowsPath WINDOWS_PROPERTIES = FSProperties( sep="\\", altsep="/", pathsep=";", linesep="\r\n", devnull="NUL", ) class FSType(Enum): """Defines which file system properties to use.""" DEFAULT = 0 # use current OS file system + modifications in fake file system POSIX = 1 # pure POSIX properties, for use in PosixPath WINDOWS = 2 # pure Windows properties, for use in WindowsPath class FakeStatResult: """Mimics os.stat_result for use as return type of `stat()` and similar. This is needed as `os.stat_result` has no possibility to set nanosecond times directly. """ def __init__( self, is_windows: bool, user_id: int, group_id: int, initial_time: Optional[float] = None, ): self.st_mode: int = 0 self.st_ino: Optional[int] = None self.st_dev: int = 0 self.st_nlink: int = 0 self.st_uid: int = user_id self.st_gid: int = group_id self._st_size: int = 0 self.is_windows: bool = is_windows self._st_atime_ns: int = int((initial_time or 0) * 1e9) self._st_mtime_ns: int = self._st_atime_ns self._st_ctime_ns: int = self._st_atime_ns def __eq__(self, other: Any) -> bool: return ( isinstance(other, FakeStatResult) and self._st_atime_ns == other._st_atime_ns and self._st_ctime_ns == other._st_ctime_ns and self._st_mtime_ns == other._st_mtime_ns and self.st_size == other.st_size and self.st_gid == other.st_gid and self.st_uid == other.st_uid and self.st_nlink == other.st_nlink and self.st_dev == other.st_dev and self.st_ino == other.st_ino and self.st_mode == other.st_mode ) def __ne__(self, other: Any) -> bool: return not self == other def copy(self) -> "FakeStatResult": """Return a copy where the float usage is hard-coded to mimic the behavior of the real os.stat_result. """ stat_result = copy(self) return stat_result def set_from_stat_result(self, stat_result: os.stat_result) -> None: """Set values from a real os.stat_result. Note: values that are controlled by the fake filesystem are not set. This includes st_ino, st_dev and st_nlink. """ self.st_mode = stat_result.st_mode self.st_uid = stat_result.st_uid self.st_gid = stat_result.st_gid self._st_size = stat_result.st_size self._st_atime_ns = stat_result.st_atime_ns self._st_mtime_ns = stat_result.st_mtime_ns self._st_ctime_ns = stat_result.st_ctime_ns @property def st_ctime(self) -> Union[int, float]: """Return the creation time in seconds.""" return self._st_ctime_ns / 1e9 @st_ctime.setter def st_ctime(self, val: Union[int, float]) -> None: """Set the creation time in seconds.""" self._st_ctime_ns = int(val * 1e9) @property def st_atime(self) -> Union[int, float]: """Return the access time in seconds.""" return self._st_atime_ns / 1e9 @st_atime.setter def st_atime(self, val: Union[int, float]) -> None: """Set the access time in seconds.""" self._st_atime_ns = int(val * 1e9) @property def st_mtime(self) -> Union[int, float]: """Return the modification time in seconds.""" return self._st_mtime_ns / 1e9 @st_mtime.setter def st_mtime(self, val: Union[int, float]) -> None: """Set the modification time in seconds.""" self._st_mtime_ns = int(val * 1e9) @property def st_size(self) -> int: if self.st_mode & S_IFLNK == S_IFLNK and self.is_windows: return 0 return self._st_size @st_size.setter def st_size(self, val: int) -> None: self._st_size = val @property def st_blocks(self) -> int: """Return the number of 512-byte blocks allocated for the file. Assumes a page size of 4096 (matches most systems). Ignores that this may not be available under some systems, and that the result may differ if the file has holes. """ if self.is_windows: raise AttributeError("'os.stat_result' object has no attribute 'st_blocks'") page_size = 4096 blocks_in_page = page_size // 512 pages = self._st_size // page_size if self._st_size % page_size: pages += 1 return pages * blocks_in_page @property def st_file_attributes(self) -> int: if not self.is_windows: raise AttributeError( "module 'os.stat_result' has no attribute 'st_file_attributes'" ) mode = 0 st_mode = self.st_mode if st_mode & stat.S_IFDIR: mode \|= stat.FILE_ATTRIBUTE_DIRECTORY # type:ignore[attr-defined] if st_mode & stat.S_IFREG: mode \|= stat.FILE_ATTRIBUTE_NORMAL # type:ignore[attr-defined] if st_mode & (stat.S_IFCHR \| stat.S_IFBLK): mode \|= stat.FILE_ATTRIBUTE_DEVICE # type:ignore[attr-defined] if st_mode & stat.S_IFLNK: mode \|= stat.FILE_ATTRIBUTE_REPARSE_POINT # type:ignore return mode @property def st_reparse_tag(self) -> int: if not self.is_windows or sys.version_info < (3, 8): raise AttributeError( "module 'os.stat_result' has no attribute 'st_reparse_tag'" ) if self.st_mode & stat.S_IFLNK: return stat.IO_REPARSE_TAG_SYMLINK # type: ignore[attr-defined] return 0 def __getitem__(self, item: int) -> Optional[int]: """Implement item access to mimic `os.stat_result` behavior.""" import stat if item == stat.ST_MODE: return self.st_mode if item == stat.ST_INO: return self.st_ino if item == stat.ST_DEV: return self.st_dev if item == stat.ST_NLINK: return self.st_nlink if item == stat.ST_UID: return self.st_uid if item == stat.ST_GID: return self.st_gid if item == stat.ST_SIZE: return self.st_size if item == stat.ST_ATIME: # item access always returns int for backward compatibility return int(self.st_atime) if item == stat.ST_MTIME: return int(self.st_mtime) if item == stat.ST_CTIME: return int(self.st_ctime) raise ValueError("Invalid item") @property def st_atime_ns(self) -> int: """Return the access time in nanoseconds.""" return self._st_atime_ns @st_atime_ns.setter def st_atime_ns(self, val: int) -> None: """Set the access time in nanoseconds.""" self._st_atime_ns = val @property def st_mtime_ns(self) -> int: """Return the modification time in nanoseconds.""" return self._st_mtime_ns @st_mtime_ns.setter def st_mtime_ns(self, val: int) -> None: """Set the modification time of the fake file in nanoseconds.""" self._st_mtime_ns = val @property def st_ctime_ns(self) -> int: """Return the creation time in nanoseconds.""" return self._st_ctime_ns @st_ctime_ns.setter def st_ctime_ns(self, val: int) -> None: """Set the creation time of the fake file in nanoseconds.""" self._st_ctime_ns = val class BinaryBufferIO(io.BytesIO): """Stream class that handles byte contents for files.""" def __init__(self, contents: Optional[bytes]): super().__init__(contents or b"") def putvalue(self, value: bytes) -> None: self.write(value) class TextBufferIO(io.TextIOWrapper): """Stream class that handles Python string contents for files.""" def __init__( self, contents: Optional[bytes] = None, newline: Optional[str] = None, encoding: Optional[str] = None, errors: str = "strict", ): self._bytestream = io.BytesIO(contents or b"") super().__init__( self._bytestream, encoding=encoding, errors=errors, newline=newline ) def getvalue(self) -> bytes: return self._bytestream.getvalue() def putvalue(self, value: bytes) -> None: self._bytestream.write(value) def is_called_from_skipped_module( skip_names: list, case_sensitive: bool, check_open_code: bool = False ) -> bool: def starts_with(path, string): if case_sensitive: return path.startswith(string) return path.lower().startswith(string.lower()) # in most cases we don't have skip names and won't need the overhead # of analyzing the traceback, except when checking for open_code if not skip_names and not check_open_code: return False stack = traceback.extract_stack() # handle the case that we try to call the original `open_code` # (since Python 3.12) # The stack in this case is: # -1: helpers.is_called_from_skipped_module: 'stack = traceback.extract_stack()' # -2: fake_open.fake_open: 'if is_called_from_skipped_module(' # -3: fake_io.open: 'return fake_open(' # -4: fake_io.open_code : 'return self._io_module.open_code(path)' if ( check_open_code and stack[-4].name == "open_code" and stack[-4].line == "return self._io_module.open_code(path)" ): return True if not skip_names: return False caller_filename = next( ( frame.filename for frame in stack[::-1] if not frame.filename.startswith("<frozen ") and not starts_with(frame.filename, STDLIB_PATH) and ( not starts_with(frame.filename, PYFAKEFS_PATH) or any( starts_with(frame.filename, test_path) for test_path in PYFAKEFS_TEST_PATHS ) ) ), None, ) if caller_filename: caller_module_name = os.path.splitext(caller_filename)[0] caller_module_name = caller_module_name.replace(os.sep, ".") if any( [ caller_module_name == sn or caller_module_name.endswith("." + sn) for sn in skip_names ] ): return True return False def reload_cleanup_handler(name): """Cleanup handler that reloads the module with the given name. Maybe needed in cases where a module is imported locally. """ if name in sys.modules: importlib.reload(sys.modules[name]) return True	16.72 KB	2026-01-20 12:58
📄 legacy_packages.py # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Imports external packages that replace or emulate internal packages. These packages are not needed with any current Python version, and their support in pyfakefs will be removed in a an upcoming release. """ try: import pathlib2 except ImportError: pathlib2 = None try: import scandir as scandir except ImportError: scandir = None	898 B	2026-01-20 12:58
📄 mox3_stubout.py # Copyright 2008 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ This is a fork of the pymox library intended to work with Python 3. The file was modified by quermit@gmail.com and dawid.fatyga@gmail.com Previously, pyfakefs used just this file from the mox3 library. However, mox3 will soon be decommissioned, yet standard mock cannot be used because of the problem described in pyfakefs #182 and mock issue 250 (https://github.com/testing-cabal/mock/issues/250). Therefore just this file was forked from mox3 and incorporated into pyfakefs. """ import inspect class StubOutForTesting: """Sample Usage: You want os.path.exists() to always return true during testing. stubs = StubOutForTesting() stubs.Set(os.path, 'exists', lambda x: 1) ... stubs.UnsetAll() The above changes os.path.exists into a lambda that returns 1. Once the ... part of the code finishes, the UnsetAll() looks up the old value of os.path.exists and restores it. """ def __init__(self): self.cache = [] self.stubs = [] def __del__(self): self.smart_unset_all() self.unset_all() def smart_set(self, obj, attr_name, new_attr): """Replace obj.attr_name with new_attr. This method is smart and works at the module, class, and instance level while preserving proper inheritance. It will not stub out C types however unless that has been explicitly allowed by the type. This method supports the case where attr_name is a staticmethod or a classmethod of obj. If obj is an instance, then it is its class that will actually be stubbed. Note that the method Set() does not do that: if obj is an instance, it (and not its class) will be stubbed. Raises AttributeError if the attribute cannot be found. """ if inspect.ismodule(obj) or ( not inspect.isclass(obj) and attr_name in obj.__dict__ ): orig_obj = obj if attr_name in obj.__dict__: orig_attr = obj.__dict__[attr_name] else: orig_attr = None else: if not inspect.isclass(obj): mro = list(inspect.getmro(obj.__class__)) else: mro = list(inspect.getmro(obj)) mro.reverse() orig_attr = None for cls in mro: try: orig_obj = cls orig_attr = obj.__dict__[attr_name] except KeyError: continue if orig_attr is None: raise AttributeError("Attribute not found.") self.stubs.append((orig_obj, attr_name, orig_attr)) orig_obj.__dict__[attr_name] = new_attr def smart_unset_all(self): """Reverses all the SmartSet() calls. Restores things to their original definition. Its okay to call SmartUnsetAll() repeatedly, as later calls have no effect if no SmartSet() calls have been made. """ self.stubs.reverse() for obj, attr_name, old_attr in self.stubs: obj.__dict__[attr_name] = old_attr self.stubs = [] def set(self, parent, child_name, new_child): """Replace child_name's old definition with new_child. Replace definition in the context of the given parent. The parent could be a module when the child is a function at module scope. Or the parent could be a class when a class' method is being replaced. The named child is set to new_child, while the prior definition is saved away for later, when unset_all() is called. This method supports the case where child_name is a staticmethod or a classmethod of parent. """ old_child = getattr(parent, child_name) old_attribute = parent.__dict__.get(child_name) if old_attribute is not None: if isinstance(old_attribute, staticmethod): old_child = staticmethod(old_child) elif isinstance(old_attribute, classmethod): old_child = classmethod(old_child.__func__) self.cache.append((parent, old_child, child_name)) parent.__dict__[child_name] = new_child def unset_all(self): """Reverses all the Set() calls. Restores things to their original definition. Its okay to call unset_all() repeatedly, as later calls have no effect if no Set() calls have been made. """ # Undo calls to set() in reverse order, in case set() was called on the # same arguments repeatedly (want the original call to be last one # undone) self.cache.reverse() for parent, old_child, child_name in self.cache: parent.__dict__[child_name] = old_child self.cache = []	5.27 KB	2026-01-20 12:58
📄 patched_packages.py # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ Provides patches for some commonly used modules that enable them to work with pyfakefs. """ import sys from importlib import reload try: import pandas as pd try: import pandas.io.parsers as parsers except ImportError: parsers = None except ImportError: pd = None parsers = None try: import xlrd except ImportError: xlrd = None try: import django try: from django.core.files import locks except ImportError: locks = None except ImportError: django = None locks = None # From pandas v 1.2 onwards the python fs functions are used even when the engine # selected is "c". This means that we don't explicitly have to change the engine. patch_pandas = parsers is not None and [ int(v) for v in pd.__version__.split(".")[:2] ] < [ 1, 2, ] def get_modules_to_patch(): modules_to_patch = {} if xlrd is not None: modules_to_patch["xlrd"] = XLRDModule if locks is not None: modules_to_patch["django.core.files.locks"] = FakeLocks return modules_to_patch def get_classes_to_patch(): classes_to_patch = {} if patch_pandas: classes_to_patch["TextFileReader"] = ["pandas.io.parsers"] return classes_to_patch def reload_handler(name): if name in sys.modules: reload(sys.modules[name]) return True def get_cleanup_handlers(): handlers = {} if pd is not None: handlers["pandas.core.arrays.arrow.extension_types"] = ( handle_extension_type_cleanup ) if django is not None: for module_name in django_view_modules(): handlers[module_name] = lambda name=module_name: reload_handler(name) return handlers def get_fake_module_classes(): fake_module_classes = {} if patch_pandas: fake_module_classes["TextFileReader"] = FakeTextFileReader return fake_module_classes if xlrd is not None: class XLRDModule: """Patches the xlrd module, which is used as the default Excel file reader by pandas. Disables using memory mapped files, which are implemented platform-specific on OS level.""" def __init__(self, _): self._xlrd_module = xlrd def open_workbook( self, filename=None, logfile=sys.stdout, verbosity=0, use_mmap=False, file_contents=None, encoding_override=None, formatting_info=False, on_demand=False, ragged_rows=False, ): return self._xlrd_module.open_workbook( filename, logfile, verbosity, False, file_contents, encoding_override, formatting_info, on_demand, ragged_rows, ) def __getattr__(self, name): """Forwards any unfaked calls to the standard xlrd module.""" return getattr(self._xlrd_module, name) if patch_pandas: # we currently need to add fake modules for both the parser module and # the contained text reader - maybe this can be simplified class FakeTextFileReader: fake_parsers = None def __init__(self, filesystem): if self.fake_parsers is None: self.__class__.fake_parsers = ParsersModule(filesystem) def __call__(self, args, kwargs): return self.fake_parsers.TextFileReader(args, *kwargs) def __getattr__(self, name): return getattr(self.fake_parsers.TextFileReader, name) class ParsersModule: def __init__(self, _): self._parsers_module = parsers class TextFileReader(parsers.TextFileReader): def __init__(self, args, *kwargs): kwargs["engine"] = "python" super().__init__(args, **kwargs) def __getattr__(self, name): """Forwards any unfaked calls to the standard xlrd module.""" return getattr(self._parsers_module, name) if pd is not None: def handle_extension_type_cleanup(_name): # the module registers two extension types on load # on reload it raises if the extensions have not been unregistered before try: import pyarrow # the code to register these types has been in the module # since it was created (in pandas 1.5) pyarrow.unregister_extension_type("pandas.interval") pyarrow.unregister_extension_type("pandas.period") except ImportError: pass return False if locks is not None: class FakeLocks: """django.core.files.locks uses low level OS functions, fake it.""" _locks_module = locks def __init__(self, _): pass @staticmethod def lock(f, flags): return True @staticmethod def unlock(f): return True def __getattr__(self, name): return getattr(self._locks_module, name) if django is not None: def get_all_view_modules(urlpatterns, modules=None): if modules is None: modules = set() for pattern in urlpatterns: if hasattr(pattern, "url_patterns"): get_all_view_modules(pattern.url_patterns, modules=modules) else: if hasattr(pattern.callback, "cls"): view = pattern.callback.cls elif hasattr(pattern.callback, "view_class"): view = pattern.callback.view_class else: view = pattern.callback modules.add(view.__module__) return modules def django_view_modules(): try: all_urlpatterns = __import__( django.conf.settings.ROOT_URLCONF ).urls.urlpatterns return get_all_view_modules(all_urlpatterns) except Exception: return set()	6.44 KB	2026-01-20 12:58
📄 py.typed # Marker file for PEP 561. The pyfakefs package uses inline types.	68 B	2026-01-20 12:58
📄 pytest_plugin.py """A pytest plugin for using pyfakefs as a fixture When pyfakefs is installed, the "fs" fixture becomes available. :Usage: def my_fakefs_test(fs): fs.create_file('/var/data/xx1.txt') assert os.path.exists('/var/data/xx1.txt') """ import contextlib import py import pytest from _pytest import capture from pyfakefs.fake_filesystem_unittest import Patcher, Pause try: from _pytest import pathlib except ImportError: pathlib = None # type:ignore[assignment] Patcher.SKIPMODULES.add(py) Patcher.SKIPMODULES.add(pytest) Patcher.SKIPMODULES.add(capture) if pathlib is not None: Patcher.SKIPMODULES.add(pathlib) @pytest.fixture def fs(request): """Fake filesystem.""" if hasattr(request, "param"): # pass optional parameters via @pytest.mark.parametrize patcher = Patcher(request.param) else: patcher = Patcher() patcher.setUp() yield patcher.fs patcher.tearDown() @pytest.fixture(scope="class") def fs_class(request): """Class-scoped fake filesystem fixture.""" if hasattr(request, "param"): patcher = Patcher(request.param) else: patcher = Patcher() patcher.setUp() yield patcher.fs patcher.tearDown() @pytest.fixture(scope="module") def fs_module(request): """Module-scoped fake filesystem fixture.""" if hasattr(request, "param"): patcher = Patcher(request.param) else: patcher = Patcher() patcher.setUp() yield patcher.fs patcher.tearDown() @pytest.fixture(scope="session") def fs_session(request): """Session-scoped fake filesystem fixture.""" if hasattr(request, "param"): patcher = Patcher(request.param) else: patcher = Patcher() patcher.setUp() yield patcher.fs patcher.tearDown() @pytest.hookimpl(tryfirst=True) def pytest_sessionfinish(session, exitstatus): """Make sure that the cache is cleared before the final test shutdown.""" Patcher.clear_fs_cache() @pytest.hookimpl(hookwrapper=True, tryfirst=True) def pytest_runtest_logreport(report): """Make sure that patching is not active during reporting.""" pause = Patcher.PATCHER is not None and Patcher.PATCHER.is_patching context_mgr = Pause(Patcher.PATCHER) if pause else contextlib.nullcontext() with context_mgr: yield if pause and report.when == "teardown": # if we get here, we are not in a function scope fixture # in this case, we still want to pause patching between the tests Patcher.PATCHER.pause() @pytest.hookimpl(tryfirst=True) def pytest_runtest_setup(item): # resume patcher if not in a function scope if Patcher.PATCHER is not None: Patcher.PATCHER.resume()	2.66 KB	2026-01-20 12:58
📁 pytest_tests	-	2026-03-21 15:45
📁 tests	-	2026-03-21 15:45

Edit File:

Terminal