()
| 354 | |
| 355 | @pytest.mark.asyncio |
| 356 | async def test_clear_stream_queue(): |
| 357 | mock_queue = MagicMock() |
| 358 | # 2 items then Empty |
| 359 | mock_queue.get_nowait.side_effect = ["item1", "item2", queue.Empty] |
| 360 | |
| 361 | with ( |
| 362 | patch.object(state, "STREAM_QUEUE", mock_queue), |
| 363 | patch.object(state, "logger") as mock_logger, |
| 364 | patch("asyncio.to_thread", side_effect=mock_queue.get_nowait), |
| 365 | ): |
| 366 | await clear_stream_queue() |
| 367 | |
| 368 | # Should have called get_nowait 3 times via to_thread |
| 369 | # Verify debug log for queue cleared |
| 370 | info_calls = [str(c) for c in mock_logger.info.call_args_list] |
| 371 | assert any("Stream queue cleared" in c for c in info_calls) |
| 372 | |
| 373 | |
| 374 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected