(ws_testdata)
| 257 | |
| 258 | |
| 259 | def test_drop_message(ws_testdata): |
| 260 | tctx, playbook, flow = ws_testdata |
| 261 | assert ( |
| 262 | playbook |
| 263 | << websocket.WebsocketStartHook(flow) |
| 264 | >> reply() |
| 265 | >> DataReceived(tctx.server, b"\x81\x03foo") |
| 266 | << websocket.WebsocketMessageHook(flow) |
| 267 | ) |
| 268 | flow.websocket.messages[-1].drop() |
| 269 | playbook >> reply() |
| 270 | playbook << None |
| 271 | assert playbook |
| 272 | |
| 273 | |
| 274 | def test_fragmented(ws_testdata): |
nothing calls this directly
no test coverage detected
searching dependent graphs…