Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,13 @@ repos:
- "--py310-plus"

- repo: https://github.com/astral-sh/ruff-pre-commit
rev: "v0.15.11"
rev: "v0.16.3"
hooks:
- id: ruff
args: ["--fix"]
- id: ruff-format
- repo: https://github.com/codespell-project/codespell
rev: v2.4.2
rev: v2.4.3
hooks:
- id: codespell
args:
Expand All @@ -44,7 +44,7 @@ repos:
language: python
files: \.py$
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.20.1
rev: v2.3.1
hooks:
- id: mypy
exclude: (docs|pyfakefs/tests)
Empty file modified docs/conf.py
100644 → 100755
Empty file.
Empty file modified pyfakefs/__init__.py
100755 → 100644
Empty file.
55 changes: 28 additions & 27 deletions pyfakefs/fake_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,38 +22,37 @@
import sys
import traceback
import weakref
from collections.abc import Callable, Iterator
from stat import (
S_IFREG,
S_IFDIR,
S_IFREG,
)
from types import TracebackType
from typing import (
Union,
TYPE_CHECKING,
Any,
cast,
AnyStr,
NoReturn,
TextIO,
TYPE_CHECKING,
Union,
cast,
)

from collections.abc import Callable, Iterator

from pyfakefs import helpers
from pyfakefs.helpers import (
FakeStatResult,
AnyPath,
AnyString,
BinaryBufferIO,
FakeStatResult,
TextBufferIO,
_OpenModes,
get_locale_encoding,
is_int_type,
is_root,
is_unicode_string,
to_string,
matching_string,
real_encoding,
AnyPath,
AnyString,
get_locale_encoding,
_OpenModes,
is_root,
to_string,
)

if TYPE_CHECKING:
Expand All @@ -79,8 +78,7 @@ class FakeLargeFileIoException(Exception):

def __init__(self, file_path: str) -> None:
super().__init__(
"Read and write operations not supported for "
"fake large file: %s" % file_path
f"Read and write operations not supported for fake large file: {file_path}"
)


Expand Down Expand Up @@ -635,7 +633,7 @@ def remove_entry(self, pathname_name: str, recursive: bool = True) -> None:

if recursive and isinstance(entry, FakeDirectory):
while entry.entries:
entry.remove_entry(list(entry.entries)[0])
entry.remove_entry(next(iter(entry.entries)))
elif entry.st_nlink == 1:
self.filesystem.change_disk_usage(-entry.size, pathname_name, entry.st_dev)

Expand Down Expand Up @@ -836,7 +834,7 @@ def filesystem(self) -> FakeFilesystem:
assert fs is not None
return fs

def __enter__(self) -> FakeFileWrapper:
def __enter__(self) -> FakeFileWrapper: # noqa:PYI034
"""To support usage of this fake file with the 'with' statement."""
return self

Expand Down Expand Up @@ -1255,19 +1253,22 @@ def __getattr__(self, name: str) -> Any:
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 ""
if args and args[0] == 0 and 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
if (
self.raw_io
and 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
Expand Down Expand Up @@ -1352,7 +1353,7 @@ def close_fd(self, fd: int | None) -> None:
def is_stream(self) -> bool:
return True

def __enter__(self) -> StandardStreamWrapper:
def __enter__(self) -> StandardStreamWrapper: # noqa:PYI034
"""To support usage of this standard stream with the 'with' statement."""
return self

Expand Down Expand Up @@ -1410,7 +1411,7 @@ def write(self, contents: bytes) -> int:
self.file_object.write(contents)
return len(contents)

def __enter__(self) -> FakeDirWrapper:
def __enter__(self) -> FakeDirWrapper: # noqa:PYI034
"""To support usage of this fake directory with the 'with' statement."""
return self

Expand Down Expand Up @@ -1443,9 +1444,9 @@ def __init__(
self.filedes: int | None = None
self.real_file = None
if mode:
self.real_file = open(fd, mode)
self.real_file = open(fd, mode) # noqa: SIM115

def __enter__(self) -> FakePipeWrapper:
def __enter__(self) -> FakePipeWrapper: # noqa:PYI034
"""To support usage of this fake pipe with the 'with' statement."""
return self

Expand Down
118 changes: 56 additions & 62 deletions pyfakefs/fake_filesystem.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,42 +92,40 @@
import sys
import tempfile
import weakref
from collections import namedtuple, OrderedDict
from collections import OrderedDict, namedtuple
from collections.abc import Callable
from doctest import TestResults
from enum import Enum

from stat import (
S_IFREG,
S_IFDIR,
S_ISLNK,
S_IFLNK,
S_IFMT,
S_IFREG,
S_ISDIR,
S_IFLNK,
S_ISLNK,
S_ISREG,
)
from typing import (
TYPE_CHECKING,
Any,
cast,
AnyStr,
overload,
NoReturn,
TYPE_CHECKING,
cast,
overload,
)

from collections.abc import Callable

from pyfakefs import fake_file, fake_path, fake_io, fake_os, helpers, fake_open
from pyfakefs.fake_file import AnyFileWrapper, AnyFile
from pyfakefs import fake_file, fake_io, fake_open, fake_os, fake_path, helpers
from pyfakefs.fake_file import AnyFile, AnyFileWrapper
from pyfakefs.helpers import (
is_int_type,
make_string_path,
to_string,
matching_string,
POSIX_PROPERTIES,
WINDOWS_PROPERTIES,
AnyPath,
AnyString,
WINDOWS_PROPERTIES,
POSIX_PROPERTIES,
FSType,
is_int_type,
make_string_path,
matching_string,
to_string,
)

if TYPE_CHECKING:
Expand Down Expand Up @@ -745,9 +743,11 @@ def change_disk_usage(
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)
if (
total_size is not None
and 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):
Expand Down Expand Up @@ -1208,25 +1208,24 @@ def splitdrive(self, path: AnyStr) -> tuple[AnyStr, AnyStr]:
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:]
if self.is_windows_fs and 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):
Expand Down Expand Up @@ -1423,10 +1422,8 @@ def starts_with_drive_letter(self, file_path: AnyStr) -> bool:
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
]
entry.upper() == file_path.upper()
for entry in self.root_dir.entries
)
self.get_object_from_normpath(file_path)
return True
Expand Down Expand Up @@ -1779,9 +1776,8 @@ def get_object_from_normpath(
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 S_ISLNK(target.st_mode) and 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)
Expand Down Expand Up @@ -2070,10 +2066,9 @@ def _do_rename(self, old_dir_object, old_name, new_dir_object, new_name):

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)
if self.islink(path) and 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
Expand Down Expand Up @@ -2108,8 +2103,7 @@ def _rename_to_existing_path(
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)
self.raise_os_error(errno.ENOTDIR, old_file_path)
return None # Nothing to do here

if old_object == new_object:
Expand Down Expand Up @@ -2145,13 +2139,12 @@ def _handle_rename_error_for_dir_or_link(
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 new_object.entries and (
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)

Expand Down Expand Up @@ -2567,7 +2560,7 @@ def add_package_metadata(self, package_name: str) -> None:
Raises:
PackageNotFoundError: if the package with the given name is not found
"""
from importlib.metadata import distribution, PackageNotFoundError
from importlib.metadata import PackageNotFoundError, distribution

# we have to pause patching to get the distribution
# from the real filesystem if we are in patch mode
Expand Down Expand Up @@ -2898,7 +2891,7 @@ def makedir(self, dir_path: AnyPath, mode: int = helpers.PERM_DEF) -> None:
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)
base_dir, _, _ = 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
Expand Down Expand Up @@ -3291,6 +3284,7 @@ def _create_temp_dir(self):

def _run_doctest() -> TestResults:
import doctest

import pyfakefs

return doctest.testmod(pyfakefs.fake_filesystem)
Expand Down
6 changes: 3 additions & 3 deletions pyfakefs/fake_filesystem_shutil.py
100755 → 100644
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,9 @@
import os
import shutil
import sys
from threading import RLock
from collections.abc import Callable
from typing import TYPE_CHECKING
from threading import RLock
from typing import TYPE_CHECKING, ClassVar

if TYPE_CHECKING:
from pyfakefs.fake_filesystem import FakeFilesystem
Expand All @@ -58,7 +58,7 @@ class FakeShutilModule:
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"]
functions_to_patch: ClassVar[list[str]] = ["copy", "copyfile", "rmtree"]
if sys.version_info < (3, 12) or sys.platform != "win32":
functions_to_patch.extend(["copy2", "copytree", "move"])

Expand Down
Loading