Bases: StateMachinePlugin
Socket interception plugin.
Patches socket.socket.connect/send/sendall/recv/close at the class level.
Uses reference counting so nested sandboxes work correctly.
States: disconnected -> connected -> closed
Source code in src/bigfoot/plugins/socket_plugin.py
| def __init__(self, verifier: "StrictVerifier") -> None:
super().__init__(verifier)
self._connect_sentinel = _StepSentinel(_SOURCE_CONNECT)
self._send_sentinel = _StepSentinel(_SOURCE_SEND)
self._sendall_sentinel = _StepSentinel(_SOURCE_SENDALL)
self._recv_sentinel = _StepSentinel(_SOURCE_RECV)
self._close_sentinel = _StepSentinel(_SOURCE_CLOSE)
|
Reference-counted class-level patch installation.
Source code in src/bigfoot/plugins/socket_plugin.py
| def activate(self) -> None:
"""Reference-counted class-level patch installation."""
with SocketPlugin._install_lock:
if SocketPlugin._install_count == 0:
self._install_patches()
SocketPlugin._install_count += 1
|
matches(interaction, expected)
Field-by-field comparison with dirty-equals support.
Source code in src/bigfoot/plugins/socket_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)
Return assertable fields for each step type.
Source code in src/bigfoot/plugins/socket_plugin.py
| def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
"""Return assertable fields for each step type."""
if interaction.source_id == _SOURCE_CONNECT:
return frozenset({"host", "port"})
if interaction.source_id == _SOURCE_SEND:
return frozenset({"data"})
if interaction.source_id == _SOURCE_SENDALL:
return frozenset({"data"})
if interaction.source_id == _SOURCE_RECV:
return frozenset({"size", "data"})
if interaction.source_id == _SOURCE_CLOSE:
return frozenset()
return frozenset(interaction.details.keys())
|
assert_connect(*, host, port)
Assert the next socket connect interaction.
Source code in src/bigfoot/plugins/socket_plugin.py
| def assert_connect(self, *, host: str, port: int) -> None:
"""Assert the next socket connect interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._connect_sentinel, host=host, port=port
)
|
Assert the next socket send interaction.
Source code in src/bigfoot/plugins/socket_plugin.py
| def assert_send(self, *, data: bytes) -> None:
"""Assert the next socket send interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._send_sentinel, data=data
)
|
Assert the next socket sendall interaction.
Source code in src/bigfoot/plugins/socket_plugin.py
| def assert_sendall(self, *, data: bytes) -> None:
"""Assert the next socket sendall interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._sendall_sentinel, data=data
)
|
assert_recv(*, size, data)
Assert the next socket recv interaction.
Source code in src/bigfoot/plugins/socket_plugin.py
| def assert_recv(self, *, size: int, data: bytes) -> None:
"""Assert the next socket recv interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._recv_sentinel, size=size, data=data
)
|
Assert the next socket close interaction.
Source code in src/bigfoot/plugins/socket_plugin.py
| def assert_close(self) -> None:
"""Assert the next socket close interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(self._close_sentinel)
|