Bases: BasePlugin
File I/O interception plugin.
Patches builtins.open, pathlib.Path read/write, os file ops, and shutil
copy/remove at the module/class level. Uses reference counting so nested
sandboxes work correctly.
Each operation+path_pattern combination has its own FIFO deque of
FileIoMockConfig objects.
NOT default enabled: requires explicit enabled_plugins = ["file_io"].
Source code in src/tripwire/plugins/file_io_plugin.py
| def __init__(self, verifier: StrictVerifier) -> None:
super().__init__(verifier)
self._queues: dict[str, deque[FileIoMockConfig]] = {}
self._registry_lock: threading.Lock = threading.Lock()
|
mock_operation(operation, path_pattern, *, returns=None, raises=None, required=True)
Register a mock for a single file I/O operation invocation.
Args:
operation: The file operation name (e.g., "open", "read_text", "remove").
path_pattern: The path pattern to match against.
returns: Value to return when this mock is consumed.
raises: If provided, this exception is raised instead of returning.
required: If False, the mock is not reported as unused at teardown.
Source code in src/tripwire/plugins/file_io_plugin.py
| def mock_operation(
self,
operation: str,
path_pattern: str,
*,
returns: Any = None, # noqa: ANN401
raises: BaseException | None = None,
required: bool = True,
) -> None:
"""Register a mock for a single file I/O operation invocation.
Args:
operation: The file operation name (e.g., "open", "read_text", "remove").
path_pattern: The path pattern to match against.
returns: Value to return when this mock is consumed.
raises: If provided, this exception is raised instead of returning.
required: If False, the mock is not reported as unused at teardown.
"""
normalized_path = os.path.normpath(path_pattern)
config = FileIoMockConfig(
operation=operation,
path_pattern=normalized_path,
returns=returns,
raises=raises,
required=required,
)
queue_key = f"{operation}:{normalized_path}"
with self._registry_lock:
if queue_key not in self._queues:
self._queues[queue_key] = deque()
self._queues[queue_key].append(config)
|
Verify builtins.open has not been patched by a third party.
Source code in src/tripwire/plugins/file_io_plugin.py
| def check_conflicts(self) -> None:
"""Verify builtins.open has not been patched by a third party."""
current_open = builtins.open
if hasattr(current_open, "__module__") and current_open.__module__ not in (
"builtins",
"_io",
"io",
None,
):
mod = current_open.__module__
if "unittest.mock" in mod:
patcher = "unittest.mock"
elif "pytest_mock" in mod:
patcher = "pytest-mock"
else:
patcher = "an unknown library"
raise ConflictError(target="builtins.open", patcher=patcher)
|
Install file I/O interceptors.
Source code in src/tripwire/plugins/file_io_plugin.py
| def install_patches(self) -> None:
"""Install file I/O interceptors."""
# Save originals
FileIoPlugin._original_open = builtins.open
FileIoPlugin._original_read_text = pathlib.Path.read_text
FileIoPlugin._original_read_bytes = pathlib.Path.read_bytes
FileIoPlugin._original_write_text = pathlib.Path.write_text
FileIoPlugin._original_write_bytes = pathlib.Path.write_bytes
FileIoPlugin._original_remove = os.remove
FileIoPlugin._original_unlink = os.unlink
FileIoPlugin._original_rename = os.rename
FileIoPlugin._original_replace = os.replace
FileIoPlugin._original_makedirs = os.makedirs
FileIoPlugin._original_mkdir = os.mkdir
FileIoPlugin._original_copy = shutil.copy
FileIoPlugin._original_copy2 = shutil.copy2
FileIoPlugin._original_copytree = shutil.copytree
FileIoPlugin._original_rmtree = shutil.rmtree
# Install interceptors
builtins.open = _intercepted_open
setattr(pathlib.Path, "read_text", _intercepted_read_text)
setattr(pathlib.Path, "read_bytes", _intercepted_read_bytes)
setattr(pathlib.Path, "write_text", _intercepted_write_text)
setattr(pathlib.Path, "write_bytes", _intercepted_write_bytes)
os.remove = _intercepted_remove
os.unlink = _intercepted_unlink
os.rename = _intercepted_rename
os.replace = _intercepted_replace
os.makedirs = _intercepted_makedirs
os.mkdir = _intercepted_mkdir
shutil.copy = _intercepted_copy
shutil.copy2 = _intercepted_copy2
shutil.copytree = _intercepted_copytree
setattr(shutil, "rmtree", _intercepted_rmtree)
|
Restore original file I/O functions.
Source code in src/tripwire/plugins/file_io_plugin.py
| def restore_patches(self) -> None:
"""Restore original file I/O functions."""
if FileIoPlugin._original_open is not None:
builtins.open = FileIoPlugin._original_open
FileIoPlugin._original_open = None
if FileIoPlugin._original_read_text is not None:
setattr(pathlib.Path, "read_text", FileIoPlugin._original_read_text)
FileIoPlugin._original_read_text = None
if FileIoPlugin._original_read_bytes is not None:
setattr(pathlib.Path, "read_bytes", FileIoPlugin._original_read_bytes)
FileIoPlugin._original_read_bytes = None
if FileIoPlugin._original_write_text is not None:
setattr(pathlib.Path, "write_text", FileIoPlugin._original_write_text)
FileIoPlugin._original_write_text = None
if FileIoPlugin._original_write_bytes is not None:
setattr(pathlib.Path, "write_bytes", FileIoPlugin._original_write_bytes)
FileIoPlugin._original_write_bytes = None
if FileIoPlugin._original_remove is not None:
os.remove = FileIoPlugin._original_remove
FileIoPlugin._original_remove = None
if FileIoPlugin._original_unlink is not None:
os.unlink = FileIoPlugin._original_unlink
FileIoPlugin._original_unlink = None
if FileIoPlugin._original_rename is not None:
os.rename = FileIoPlugin._original_rename
FileIoPlugin._original_rename = None
if FileIoPlugin._original_replace is not None:
os.replace = FileIoPlugin._original_replace
FileIoPlugin._original_replace = None
if FileIoPlugin._original_makedirs is not None:
os.makedirs = FileIoPlugin._original_makedirs
FileIoPlugin._original_makedirs = None
if FileIoPlugin._original_mkdir is not None:
os.mkdir = FileIoPlugin._original_mkdir
FileIoPlugin._original_mkdir = None
if FileIoPlugin._original_copy is not None:
shutil.copy = FileIoPlugin._original_copy
FileIoPlugin._original_copy = None
if FileIoPlugin._original_copy2 is not None:
shutil.copy2 = FileIoPlugin._original_copy2
FileIoPlugin._original_copy2 = None
if FileIoPlugin._original_copytree is not None:
shutil.copytree = FileIoPlugin._original_copytree
FileIoPlugin._original_copytree = None
if FileIoPlugin._original_rmtree is not None:
setattr(shutil, "rmtree", FileIoPlugin._original_rmtree)
FileIoPlugin._original_rmtree = None
|
matches(interaction, expected)
Field-by-field comparison with dirty-equals support.
Source code in src/tripwire/plugins/file_io_plugin.py
| def matches(self, interaction: Interaction, expected: dict[str, Any]) -> bool:
"""Field-by-field comparison with dirty-equals support."""
try:
for key, expected_val in expected.items():
actual_val = interaction.details.get(key)
if expected_val != actual_val:
return False
return True
except Exception:
return False
|
assertable_fields(interaction)
All detail fields are required in assert_interaction().
Source code in src/tripwire/plugins/file_io_plugin.py
| def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
"""All detail fields are required in assert_interaction()."""
return frozenset(interaction.details.keys())
|
Return all FileIoMockConfig with required=True still in any queue.
Source code in src/tripwire/plugins/file_io_plugin.py
| def get_unused_mocks(self) -> list[FileIoMockConfig]:
"""Return all FileIoMockConfig with required=True still in any queue."""
unused: list[FileIoMockConfig] = []
with self._registry_lock:
for queue in self._queues.values():
for config in queue:
if config.required:
unused.append(config)
return unused
|
Typed helper: assert the next file_io:open interaction.
All three fields (path, mode, encoding) are required by
assertable_fields, but this helper accepts **kwargs so that
the verifier can enforce completeness via MissingAssertionFieldsError.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_open(
self,
**expected: Any, # noqa: ANN401
) -> None:
"""Typed helper: assert the next file_io:open interaction.
All three fields (path, mode, encoding) are required by
assertable_fields, but this helper accepts **kwargs so that
the verifier can enforce completeness via MissingAssertionFieldsError.
"""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = "file_io:open"
sentinel = _FileIoSentinel(source_id)
if "path" in expected:
expected["path"] = os.path.normpath(expected["path"])
_get_test_verifier_or_raise().assert_interaction(
sentinel,
**expected,
)
|
Typed helper: assert the next file_io:read_text interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_read_text(self, path: str) -> None:
"""Typed helper: assert the next file_io:read_text interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = "file_io:read_text"
sentinel = _FileIoSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
path=os.path.normpath(path),
)
|
Typed helper: assert the next file_io:read_bytes interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_read_bytes(self, path: str) -> None:
"""Typed helper: assert the next file_io:read_bytes interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = "file_io:read_bytes"
sentinel = _FileIoSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
path=os.path.normpath(path),
)
|
assert_write_text(path, data)
Typed helper: assert the next file_io:write_text interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_write_text(self, path: str, data: str) -> None:
"""Typed helper: assert the next file_io:write_text interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = "file_io:write_text"
sentinel = _FileIoSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
path=os.path.normpath(path),
data=data,
)
|
assert_write_bytes(path, data)
Typed helper: assert the next file_io:write_bytes interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_write_bytes(self, path: str, data: bytes) -> None:
"""Typed helper: assert the next file_io:write_bytes interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = "file_io:write_bytes"
sentinel = _FileIoSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
path=os.path.normpath(path),
data=data,
)
|
Typed helper: assert the next file_io:remove or file_io:unlink interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_remove(self, path: str) -> None:
"""Typed helper: assert the next file_io:remove or file_io:unlink interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
verifier = _get_test_verifier_or_raise()
# Try both remove and unlink source_ids
timeline = verifier._timeline
for interaction in timeline.all_unasserted():
if interaction.source_id in ("file_io:remove", "file_io:unlink"):
sentinel = _FileIoSentinel(interaction.source_id)
verifier.assert_interaction(
sentinel, path=os.path.normpath(path),
)
return
# Fall back to remove
sentinel = _FileIoSentinel("file_io:remove")
verifier.assert_interaction(sentinel, path=os.path.normpath(path))
|
Typed helper: assert the next file_io:rename or file_io:replace interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_rename(self, src: str, dst: str) -> None:
"""Typed helper: assert the next file_io:rename or file_io:replace interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
verifier = _get_test_verifier_or_raise()
nsrc, ndst = os.path.normpath(src), os.path.normpath(dst)
# Try both rename and replace source_ids
timeline = verifier._timeline
for interaction in timeline.all_unasserted():
if interaction.source_id in ("file_io:rename", "file_io:replace"):
sentinel = _FileIoSentinel(interaction.source_id)
verifier.assert_interaction(sentinel, src=nsrc, dst=ndst)
return
# Fall back to rename
sentinel = _FileIoSentinel("file_io:rename")
verifier.assert_interaction(sentinel, src=nsrc, dst=ndst)
|
assert_makedirs(path, exist_ok)
Typed helper: assert the next file_io:makedirs interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_makedirs(self, path: str, exist_ok: bool) -> None:
"""Typed helper: assert the next file_io:makedirs interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
sentinel = _FileIoSentinel("file_io:makedirs")
_get_test_verifier_or_raise().assert_interaction(
sentinel, path=os.path.normpath(path), exist_ok=exist_ok,
)
|
Typed helper: assert the next file_io:mkdir interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_mkdir(self, path: str) -> None:
"""Typed helper: assert the next file_io:mkdir interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
sentinel = _FileIoSentinel("file_io:mkdir")
_get_test_verifier_or_raise().assert_interaction(
sentinel, path=os.path.normpath(path),
)
|
Typed helper: assert the next file_io:copy or file_io:copy2 interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_copy(self, src: str, dst: str) -> None:
"""Typed helper: assert the next file_io:copy or file_io:copy2 interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
verifier = _get_test_verifier_or_raise()
nsrc, ndst = os.path.normpath(src), os.path.normpath(dst)
timeline = verifier._timeline
for interaction in timeline.all_unasserted():
if interaction.source_id in ("file_io:copy", "file_io:copy2"):
sentinel = _FileIoSentinel(interaction.source_id)
verifier.assert_interaction(sentinel, src=nsrc, dst=ndst)
return
sentinel = _FileIoSentinel("file_io:copy")
verifier.assert_interaction(sentinel, src=nsrc, dst=ndst)
|
assert_copytree(src, dst)
Typed helper: assert the next file_io:copytree interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_copytree(self, src: str, dst: str) -> None:
"""Typed helper: assert the next file_io:copytree interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = "file_io:copytree"
sentinel = _FileIoSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
src=os.path.normpath(src),
dst=os.path.normpath(dst),
)
|
Typed helper: assert the next file_io:rmtree interaction.
Source code in src/tripwire/plugins/file_io_plugin.py
| def assert_rmtree(self, path: str) -> None:
"""Typed helper: assert the next file_io:rmtree interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = "file_io:rmtree"
sentinel = _FileIoSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
path=os.path.normpath(path),
)
|