Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
("sec-websocket-version", "13"),
],
)
await server.reader.send(h2_client.data_to_send()) # type: ignore
events = h2_client.receive_data(await server.writer.receive()) # type: ignore
await server.reader.send(h2_client.data_to_send()) # type: ignore
events = h2_client.receive_data(await server.writer.receive()) # type: ignore
events = h2_client.receive_data(await server.writer.receive()) # type: ignore
assert isinstance(events[0], h2.events.ResponseReceived)
assert events[0].headers == [
(b":status", b"200"),
(b"date", b"Thu, 01 Jan 1970 01:23:20 GMT"),
(b"server", b"hypercorn-h2"),
]
client = wsproto.connection.Connection(wsproto.ConnectionType.CLIENT)
h2_client.send_data(stream_id, client.send(wsproto.events.BytesMessage(data=SANITY_BODY)))
await server.reader.send(h2_client.data_to_send()) # type: ignore
events = h2_client.receive_data(await server.writer.receive()) # type: ignore
client.receive_data(events[0].data)
assert list(client.events()) == [wsproto.events.TextMessage(data="Hello & Goodbye")]
h2_client.send_data(stream_id, client.send(wsproto.events.CloseConnection(code=1000)))
await server.reader.send(h2_client.data_to_send()) # type: ignore
events = h2_client.receive_data(await server.writer.receive()) # type: ignore
client.receive_data(events[0].data)
assert list(client.events()) == [wsproto.events.CloseConnection(code=1000, reason="")]
await server.reader.send(b"") # type: ignore
def test_send_message(client_sends: bool, final: bool) -> None:
client = Connection(CLIENT)
server = Connection(SERVER)
if client_sends:
local = client
remote = server
else:
local = server
remote = client
data = b"x" * 23
remote.receive_data(local.send(BytesMessage(data=data, message_finished=final)))
event = next(remote.events())
assert isinstance(event, BytesMessage)
assert event.data == data
assert event.message_finished is final
BytesMessage(data=b"abc", frame_finished=False, message_finished=True),
),
(
BytesMessage(data=b"abc", frame_finished=False, message_finished=True),
TextMessage(data="abc", frame_finished=False, message_finished=True),
),
],
)
def test_buffer_mixed_types(data: list) -> None:
buffer_ = WebsocketBuffer(10)
buffer_.extend(data[0])
with pytest.raises(TypeError):
buffer_.extend(data[1])
self.scope, {"status": status_code, "headers": []}, time() - self.start_time
)
elif (
message["type"] == "websocket.http.response.start"
and self.state == ASGIWebsocketState.HANDSHAKE
):
self.response = message
elif message["type"] == "websocket.http.response.body" and self.state in {
ASGIWebsocketState.HANDSHAKE,
ASGIWebsocketState.RESPONSE,
}:
await self._send_rejection(message)
elif message["type"] == "websocket.send" and self.state == ASGIWebsocketState.CONNECTED:
event: WSProtoEvent
if message.get("bytes") is not None:
event = BytesMessage(data=bytes(message["bytes"]))
elif not isinstance(message["text"], str):
raise TypeError(f"{message['text']} should be a str")
else:
event = TextMessage(data=message["text"])
await self._send_wsproto_event(event)
elif (
message["type"] == "websocket.close" and self.state == ASGIWebsocketState.HANDSHAKE
):
self.state = ASGIWebsocketState.HTTPCLOSED
await self._send_error_response(403)
elif message["type"] == "websocket.close":
self.state = ASGIWebsocketState.CLOSED
await self._send_wsproto_event(
CloseConnection(code=int(message.get("code", CloseReason.NORMAL_CLOSURE)))
)
await self.send(EndData(stream_id=self.stream_id))
def handle_events(self):
for event in self.conn.events():
if isinstance(event, events.Request):
self.handle_connect(event)
elif isinstance(event, events.TextMessage):
self.handle_text(event)
elif isinstance(event, events.BytesMessage):
self.handle_bytes(event)
elif isinstance(event, events.RejectConnection):
self.handle_no_connect(event)
elif isinstance(event, events.RejectData):
self.handle_no_connect(event)
elif isinstance(event, events.CloseConnection):
self.handle_close(event)
elif isinstance(event, events.Ping):
self.handle_ping(event)
async def _reader_task(self):
handlers = {
AcceptConnection: self._handle_accept_connection_event,
BytesMessage: self._handle_message_event,
CloseConnection: self._handle_close_connection_event,
Ping: self._handle_ping_event,
Pong: self._handle_pong_event,
RejectConnection: self._handle_reject_connection_event,
RejectData: self._handle_reject_data_event,
# Request: lambda event: None, # We won't handle server-side events.
TextMessage: self._handle_message_event,
}
# We need to initiate the opening handshake.
await self._do_handshake()
while self._reader_running:
for event in self._wsproto.events():
event_type = type(event)
try:
Raises
------
:exc:`anysocks.exceptions.ConnectionClosed`
If the connection is already closed.
:exc:`ValueError`
If the type of ``message`` isn't ``str`` or ``bytes``.
"""
if self._closed:
raise ConnectionClosed(self._close_code, self._close_reason)
if isinstance(message, str):
event = TextMessage(data=message)
elif isinstance(message, bytes):
event = BytesMessage(data=message)
else:
raise ValueError('Message must be bytes or string')
await self._send(event)