(ws_testdata)
| 272 | |
| 273 | |
| 274 | def test_fragmented(ws_testdata): |
| 275 | tctx, playbook, flow = ws_testdata |
| 276 | assert ( |
| 277 | playbook |
| 278 | << websocket.WebsocketStartHook(flow) |
| 279 | >> reply() |
| 280 | >> DataReceived(tctx.server, b"\x01\x03foo") |
| 281 | >> DataReceived(tctx.server, b"\x80\x03bar") |
| 282 | << websocket.WebsocketMessageHook(flow) |
| 283 | >> reply() |
| 284 | << SendData(tctx.client, b"\x01\x03foo") |
| 285 | << SendData(tctx.client, b"\x80\x03bar") |
| 286 | ) |
| 287 | assert flow.websocket.messages[-1].content == b"foobar" |
| 288 | |
| 289 | |
| 290 | def test_unfragmented(ws_testdata): |
nothing calls this directly
no test coverage detected
searching dependent graphs…