WebSocket Plugins

AsyncWebSocketPlugin

AsyncWebSocketPlugin

AsyncWebSocketPlugin(verifier)

Bases: StateMachinePlugin

Async WebSocket interception plugin.

Patches websockets.connect at the module level. Uses reference counting so nested sandboxes work correctly.

States: connecting -> open -> closed

Source code in src/bigfoot/plugins/websocket_plugin.py

def __init__(self, verifier: StrictVerifier) -> None:
    super().__init__(verifier)
    self._connect_sentinel = _StepSentinel(_ASYNC_SOURCE_CONNECT)
    self._send_sentinel = _StepSentinel(_ASYNC_SOURCE_SEND)
    self._recv_sentinel = _StepSentinel(_ASYNC_SOURCE_RECV)
    self._close_sentinel = _StepSentinel(_ASYNC_SOURCE_CLOSE)

activate

activate()

Reference-counted class-level patch installation.

Source code in src/bigfoot/plugins/websocket_plugin.py

def activate(self) -> None:
    """Reference-counted class-level patch installation."""
    if not _WEBSOCKETS_AVAILABLE:
        raise ImportError(
            "Install bigfoot[websockets] to use AsyncWebSocketPlugin: "
            "pip install bigfoot[websockets]"
        )
    with AsyncWebSocketPlugin._install_lock:
        if AsyncWebSocketPlugin._install_count == 0:
            self._install_patches()
        AsyncWebSocketPlugin._install_count += 1

matches

matches(interaction, expected)

Field-by-field comparison with dirty-equals support.

Source code in src/bigfoot/plugins/websocket_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

assertable_fields(interaction)

Return assertable fields for each step type.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
    """Return assertable fields for each step type."""
    if interaction.source_id == _ASYNC_SOURCE_CLOSE:
        return frozenset()
    return frozenset(interaction.details.keys())

assert_connect

assert_connect(*, uri)

Assert the next async websocket connect interaction.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assert_connect(self, *, uri: str) -> None:
    """Assert the next async websocket connect interaction."""
    from bigfoot._context import _get_test_verifier_or_raise  # noqa: PLC0415

    _get_test_verifier_or_raise().assert_interaction(self._connect_sentinel, uri=uri)

assert_send

assert_send(*, message)

Assert the next async websocket send interaction.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assert_send(self, *, message: Any) -> None:  # noqa: ANN401
    """Assert the next async websocket send interaction."""
    from bigfoot._context import _get_test_verifier_or_raise  # noqa: PLC0415

    _get_test_verifier_or_raise().assert_interaction(self._send_sentinel, message=message)

assert_recv

assert_recv(*, message)

Assert the next async websocket recv interaction.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assert_recv(self, *, message: Any) -> None:  # noqa: ANN401
    """Assert the next async websocket recv interaction."""
    from bigfoot._context import _get_test_verifier_or_raise  # noqa: PLC0415

    _get_test_verifier_or_raise().assert_interaction(self._recv_sentinel, message=message)

assert_close

assert_close()

Assert the next async websocket close interaction.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assert_close(self) -> None:
    """Assert the next async websocket close interaction."""
    from bigfoot._context import _get_test_verifier_or_raise  # noqa: PLC0415

    _get_test_verifier_or_raise().assert_interaction(self._close_sentinel)

SyncWebSocketPlugin

SyncWebSocketPlugin

SyncWebSocketPlugin(verifier)

Bases: StateMachinePlugin

Sync WebSocket interception plugin (websocket-client library).

Patches websocket.create_connection at the module level. Uses reference counting so nested sandboxes work correctly.

States: connecting -> open -> closed

Source code in src/bigfoot/plugins/websocket_plugin.py

def __init__(self, verifier: StrictVerifier) -> None:
    super().__init__(verifier)
    self._connect_sentinel = _StepSentinel(_SYNC_SOURCE_CONNECT)
    self._send_sentinel = _StepSentinel(_SYNC_SOURCE_SEND)
    self._recv_sentinel = _StepSentinel(_SYNC_SOURCE_RECV)
    self._close_sentinel = _StepSentinel(_SYNC_SOURCE_CLOSE)

activate

activate()

Reference-counted class-level patch installation.

Source code in src/bigfoot/plugins/websocket_plugin.py

def activate(self) -> None:
    """Reference-counted class-level patch installation."""
    if not _WEBSOCKET_CLIENT_AVAILABLE:
        raise ImportError(
            "Install bigfoot[websocket-client] to use SyncWebSocketPlugin: "
            "pip install bigfoot[websocket-client]"
        )
    with SyncWebSocketPlugin._install_lock:
        if SyncWebSocketPlugin._install_count == 0:
            self._install_patches()
        SyncWebSocketPlugin._install_count += 1

matches

matches(interaction, expected)

Field-by-field comparison with dirty-equals support.

Source code in src/bigfoot/plugins/websocket_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

assertable_fields(interaction)

Return assertable fields for each step type.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
    """Return assertable fields for each step type."""
    if interaction.source_id == _SYNC_SOURCE_CLOSE:
        return frozenset()
    return frozenset(interaction.details.keys())

assert_connect

assert_connect(*, uri)

Assert the next sync websocket connect interaction.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assert_connect(self, *, uri: str) -> None:
    """Assert the next sync websocket connect interaction."""
    from bigfoot._context import _get_test_verifier_or_raise  # noqa: PLC0415

    _get_test_verifier_or_raise().assert_interaction(self._connect_sentinel, uri=uri)

assert_send

assert_send(*, message)

Assert the next sync websocket send interaction.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assert_send(self, *, message: Any) -> None:  # noqa: ANN401
    """Assert the next sync websocket send interaction."""
    from bigfoot._context import _get_test_verifier_or_raise  # noqa: PLC0415

    _get_test_verifier_or_raise().assert_interaction(self._send_sentinel, message=message)

assert_recv

assert_recv(*, message)

Assert the next sync websocket recv interaction.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assert_recv(self, *, message: Any) -> None:  # noqa: ANN401
    """Assert the next sync websocket recv interaction."""
    from bigfoot._context import _get_test_verifier_or_raise  # noqa: PLC0415

    _get_test_verifier_or_raise().assert_interaction(self._recv_sentinel, message=message)

assert_close

assert_close()

Assert the next sync websocket close interaction.

Source code in src/bigfoot/plugins/websocket_plugin.py

def assert_close(self) -> None:
    """Assert the next sync websocket close interaction."""
    from bigfoot._context import _get_test_verifier_or_raise  # noqa: PLC0415

    _get_test_verifier_or_raise().assert_interaction(self._close_sentinel)