Skip to content

test: preserve coalesced WebSocket deadline frames #15

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MikiEEE merged 1 commit into from
Aug 23, 2026
Merged

test: preserve coalesced WebSocket deadline frames #15

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 30 additions & 12 deletions tests/test_websocket.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -871,19 +871,23 @@ def connect(path: str):
response = b""
while b"\r\n\r\n" not in response:
response += stream.recv(4096)
return stream, response
headers, separator, websocket_data = response.partition(b"\r\n\r\n")
return stream, headers + separator, websocket_data

def client_work() -> None:
try:
stream, response = connect("/handshake-timeout")
stream, response, _ = connect("/handshake-timeout")
outcomes["handshake"] = response
stream.close()

stream, response = connect("/idle-timeout")
stream, response, websocket_data = connect("/idle-timeout")
outcomes["idle_handshake"] = response
idle_client = api.Connection(api.ConnectionType.CLIENT)
idle_events = _receive_events(
stream, idle_client, api.CloseConnection
stream,
idle_client,
api.CloseConnection,
initial_data=websocket_data,
)
idle_close = next(
event
Expand All @@ -894,23 +898,37 @@ def client_work() -> None:
stream.sendall(idle_client.send(idle_close.response()))
stream.close()

stream, response = connect("/pong-timeout")
stream, response, websocket_data = connect("/pong-timeout")
outcomes["pong_handshake"] = response
pong_client = api.Connection(api.ConnectionType.CLIENT)
ping_events = _receive_events(stream, pong_client, api.Ping)
ping_events = _receive_events(
stream,
pong_client,
api.Ping,
initial_data=websocket_data,
)
outcomes["ping_payload"] = next(
event.payload
for event in ping_events
if isinstance(event, api.Ping)
)
close_events = _receive_events(
stream, pong_client, api.CloseConnection
)
pong_close = next(
event
for event in close_events
if isinstance(event, api.CloseConnection)
(
event
for event in ping_events
if isinstance(event, api.CloseConnection)
),
None,
)
if pong_close is None:
close_events = _receive_events(
stream, pong_client, api.CloseConnection
)
pong_close = next(
event
for event in close_events
if isinstance(event, api.CloseConnection)
)
outcomes["pong_code"] = pong_close.code
stream.sendall(pong_client.send(pong_close.response()))
stream.close()
Expand Down
Loading