Class PatchableWebSocketReader

tests/test_websocket_parser.py:26–46

WebSocketReader subclass that allows for patching parse_frame.

class PatchableWebSocketReader(WebSocketReader):
    """WebSocketReader subclass that allows for patching parse_frame."""

    def parse_frame(
        self, data: bytes
    ) -> list[tuple[bool, int, bytes | bytearray, int]]:
        # This method is overridden to allow for patching in tests.
        frames: list[tuple[bool, int, bytes | bytearray, int]] = []

        def _handle_frame(
            fin: bool,
            opcode: int,
            payload: bytes | bytearray,
            compressed: int,
        ) -> None:
            # This method is overridden to allow for patching in tests.
            frames.append((fin, opcode, payload, compressed))

        with mock.patch.object(self, "_handle_frame", _handle_frame):
            self._feed_data(data)
        return frames


def build_frame(