(
out: WebSocketDataQueue, parser: PatchableWebSocketReader
)
| 371 | |
| 372 | |
| 373 | def test_large_message( |
| 374 | out: WebSocketDataQueue, parser: PatchableWebSocketReader |
| 375 | ) -> None: |
| 376 | large_payload = b"b" * 131072 |
| 377 | data = build_frame(large_payload, WSMsgType.BINARY) |
| 378 | parser._feed_data(data) |
| 379 | |
| 380 | res = out._buffer[0] |
| 381 | assert res == ((WSMsgType.BINARY, large_payload, ""), 131072) |
| 382 | |
| 383 | |
| 384 | def test_large_masked_message( |
nothing calls this directly
no test coverage detected