(
out: WebSocketDataQueue, parser: PatchableWebSocketReader
)
| 416 | |
| 417 | |
| 418 | def test_continuation( |
| 419 | out: WebSocketDataQueue, parser: PatchableWebSocketReader |
| 420 | ) -> None: |
| 421 | data1 = build_frame(b"line1", WSMsgType.TEXT, is_fin=False) |
| 422 | parser._feed_data(data1) |
| 423 | |
| 424 | data2 = build_frame(b"line2", WSMsgType.CONTINUATION) |
| 425 | parser._feed_data(data2) |
| 426 | |
| 427 | res = out._buffer[0] |
| 428 | assert res == (WSMessage(WSMsgType.TEXT, "line1line2", ""), 10) |
| 429 | |
| 430 | |
| 431 | def test_continuation_with_ping( |
nothing calls this directly
no test coverage detected