Pending notifications are flushed when WS reconnects.
(self)
| 62 | |
| 63 | @pytest.mark.asyncio |
| 64 | async def test_drain_pending_flushes(self): |
| 65 | """Pending notifications are flushed when WS reconnects.""" |
| 66 | bridge = _make_bridge() |
| 67 | await bridge.push_tool_result("result1") |
| 68 | await bridge.push_tool_result("result2") |
| 69 | assert len(bridge._pending_notifications) == 2 |
| 70 | |
| 71 | ws = FakeWs() |
| 72 | bridge.set_ws(ws) |
| 73 | await bridge.drain_pending() |
| 74 | |
| 75 | assert len(ws.sent) == 2 |
| 76 | assert len(bridge._pending_notifications) == 0 |
| 77 | |
| 78 | @pytest.mark.asyncio |
| 79 | async def test_duplicate_launch_closes_previous(self): |
nothing calls this directly
no test coverage detected