FileIoPlugin Guide¶
FileIoPlugin intercepts file system operations across builtins.open, pathlib.Path read/write methods, os file operations, and shutil copy/remove operations. Each operation+path combination has its own independent FIFO queue. The plugin uses a ContextVar-based reentrancy guard to prevent self-interference with tripwire's own file I/O.
Important: FileIoPlugin is always available (no extra install required) but is NOT default enabled. You must explicitly enable it via enabled_plugins = ["file_io"] in your tripwire config, or access it through the tripwire.file_io proxy.
Setup¶
In pytest, access FileIoPlugin through the tripwire.file_io proxy. It auto-creates the plugin for the current test on first use:
import tripwire
def test_read_config():
tripwire.file_io.mock_operation(
"read_text", "/etc/myapp/config.yaml",
returns="database:\n host: localhost\n port: 5432",
)
with tripwire:
from pathlib import Path
config = Path("/etc/myapp/config.yaml").read_text()
assert "localhost" in config
tripwire.file_io.assert_read_text(path="/etc/myapp/config.yaml")
For manual use outside pytest, construct FileIoPlugin explicitly:
from tripwire import StrictVerifier
from tripwire.plugins.file_io_plugin import FileIoPlugin
verifier = StrictVerifier()
file_io = FileIoPlugin(verifier)
Each verifier may have at most one FileIoPlugin. A second FileIoPlugin(verifier) raises ValueError.
Registering mocks¶
Use tripwire.file_io.mock_operation(operation, path_pattern, *, returns, ...) to register a mock before entering the sandbox:
tripwire.file_io.mock_operation("open", "/tmp/data.csv", returns="id,name\n1,Alice")
tripwire.file_io.mock_operation("remove", "/tmp/data.csv", returns=None)
Parameters¶
| Parameter | Type | Default | Description |
|---|---|---|---|
operation |
str |
required | File operation name (see supported operations below) |
path_pattern |
str |
required | Path to match against |
returns |
Any |
None |
Value to return when this mock is consumed |
raises |
BaseException \| None |
None |
Exception to raise instead of returning |
required |
bool |
True |
Whether an unused mock causes UnusedMocksError at teardown |
Supported operations¶
| Operation | Intercepts | Details fields |
|---|---|---|
open |
builtins.open(...) |
path, mode, encoding |
read_text |
Path.read_text(...) |
path |
read_bytes |
Path.read_bytes(...) |
path |
write_text |
Path.write_text(...) |
path, data |
write_bytes |
Path.write_bytes(...) |
path, data |
remove |
os.remove(...) |
path |
unlink |
os.unlink(...) |
path |
rename |
os.rename(...) |
src, dst |
replace |
os.replace(...) |
src, dst |
makedirs |
os.makedirs(...) |
path, exist_ok |
mkdir |
os.mkdir(...) |
path |
copy |
shutil.copy(...) |
src, dst |
copy2 |
shutil.copy2(...) |
src, dst |
copytree |
shutil.copytree(...) |
src, dst |
rmtree |
shutil.rmtree(...) |
path |
FIFO queues¶
Each operation+path combination has its own independent FIFO queue. Multiple mocks for the same operation and path are consumed in registration order:
def test_multiple_reads():
tripwire.file_io.mock_operation("read_text", "/etc/myapp/config.yaml", returns="v1")
tripwire.file_io.mock_operation("read_text", "/etc/myapp/config.yaml", returns="v2")
with tripwire:
from pathlib import Path
first = Path("/etc/myapp/config.yaml").read_text()
second = Path("/etc/myapp/config.yaml").read_text()
assert first == "v1"
assert second == "v2"
tripwire.file_io.assert_read_text(path="/etc/myapp/config.yaml")
tripwire.file_io.assert_read_text(path="/etc/myapp/config.yaml")
Asserting interactions¶
Use the typed assertion helpers on tripwire.file_io:
assert_open(**expected)¶
All three fields (path, mode, encoding) are required:
assert_read_text(path)¶
assert_read_bytes(path)¶
assert_write_text(path, data)¶
assert_write_bytes(path, data)¶
assert_remove(path)¶
Matches both os.remove and os.unlink interactions:
assert_rename(src, dst)¶
Matches both os.rename and os.replace interactions:
assert_makedirs(path, exist_ok)¶
assert_mkdir(path)¶
assert_copy(src, dst)¶
Matches both shutil.copy and shutil.copy2 interactions:
assert_copytree(src, dst)¶
assert_rmtree(path)¶
Simulating errors¶
Use the raises parameter to simulate file system errors:
import tripwire
def test_file_not_found():
tripwire.file_io.mock_operation(
"read_text", "/etc/myapp/config.yaml",
raises=FileNotFoundError("[Errno 2] No such file or directory: '/etc/myapp/config.yaml'"),
)
with tripwire:
from pathlib import Path
with pytest.raises(FileNotFoundError):
Path("/etc/myapp/config.yaml").read_text()
tripwire.file_io.assert_read_text(path="/etc/myapp/config.yaml")
Full example¶
Production code (examples/file_processor/app.py):
"""Archive a directory and clean up the source."""
import os
import shutil
from pathlib import Path
def archive_and_clean(source_dir, archive_dir, manifest_path):
"""Copy source to archive, write a manifest, then remove the source."""
os.makedirs(archive_dir, exist_ok=True)
shutil.copytree(source_dir, os.path.join(archive_dir, "latest"))
Path(manifest_path).write_text(f"archived: {source_dir}")
shutil.rmtree(source_dir)
Test (examples/file_processor/test_app.py):
"""Test file archival using tripwire file_io_mock."""
import tripwire
from .app import archive_and_clean
def test_archive_and_clean():
tripwire.file_io.mock_operation("makedirs", "/backups/2024", returns=None)
tripwire.file_io.mock_operation("copytree", "/var/data/reports", returns=None)
tripwire.file_io.mock_operation(
"write_text", "/var/log/manifest.txt", returns=None,
)
tripwire.file_io.mock_operation("rmtree", "/var/data/reports", returns=None)
with tripwire:
archive_and_clean(
"/var/data/reports", "/backups/2024", "/var/log/manifest.txt",
)
tripwire.file_io.assert_makedirs(path="/backups/2024", exist_ok=True)
tripwire.file_io.assert_copytree(
src="/var/data/reports", dst="/backups/2024/latest",
)
tripwire.file_io.assert_write_text(
path="/var/log/manifest.txt", data="archived: /var/data/reports",
)
tripwire.file_io.assert_rmtree(path="/var/data/reports")
open() return values¶
When mocking open(), the return value is automatically wrapped in the appropriate IO class:
strreturn value: wrapped inio.StringIObytesreturn value: wrapped inio.BytesIONonereturn value (write mode): returns emptyio.StringIOorio.BytesIOdepending on mode
def test_open_read():
tripwire.file_io.mock_operation(
"open", "/tmp/data.csv",
returns="id,name\n1,Alice\n2,Bob",
)
with tripwire:
with open("/tmp/data.csv", "r") as f:
lines = f.readlines()
assert len(lines) == 3
tripwire.file_io.assert_open(path="/tmp/data.csv", mode="r", encoding="utf-8")
Optional mocks¶
Mark a mock as optional with required=False:
An optional mock that is never triggered does not cause UnusedMocksError at teardown.
UnmockedInteractionError¶
When code performs a file operation that has no remaining mocks in its queue, tripwire raises UnmockedInteractionError: