(self)
| 73 | |
| 74 | @pytest.mark.asyncio |
| 75 | async def test_flush_async(self) -> None: |
| 76 | q = SendQueue() |
| 77 | q.enqueue("a") |
| 78 | q.enqueue("b") |
| 79 | |
| 80 | sent: list[str] = [] |
| 81 | |
| 82 | async def async_send(data: str) -> None: |
| 83 | sent.append(data) |
| 84 | |
| 85 | await q.flush_async(async_send) |
| 86 | assert sent == ["a", "b"] |
| 87 | assert len(q) == 0 |
| 88 | |
| 89 | @pytest.mark.asyncio |
| 90 | async def test_flush_async_requeues_on_failure(self) -> None: |
nothing calls this directly
no test coverage detected