Test _shutdown_resources cleans up everything.
()
| 477 | |
| 478 | @pytest.mark.asyncio |
| 479 | async def test_shutdown_resources(): |
| 480 | """Test _shutdown_resources cleans up everything.""" |
| 481 | |
| 482 | # Create a dummy task in the current loop |
| 483 | async def dummy_coro(): |
| 484 | await asyncio.sleep(0.1) |
| 485 | |
| 486 | real_task = asyncio.create_task(dummy_coro()) |
| 487 | |
| 488 | # Set up state with mocked resources |
| 489 | mock_stream_process = MagicMock() |
| 490 | state.STREAM_PROCESS = mock_stream_process |
| 491 | state.worker_task = real_task |
| 492 | state.page_instance = MagicMock() |
| 493 | |
| 494 | mock_browser = MagicMock() |
| 495 | mock_browser.is_connected.return_value = True |
| 496 | mock_browser.close = AsyncMock() |
| 497 | state.browser_instance = mock_browser |
| 498 | |
| 499 | mock_pw = MagicMock() |
| 500 | mock_pw.stop = AsyncMock() |
| 501 | state.playwright_manager = mock_pw |
| 502 | |
| 503 | with patch( |
| 504 | "api_utils.app._close_page_logic", new_callable=AsyncMock |
| 505 | ) as mock_close_page: |
| 506 | await _shutdown_resources() |
| 507 | |
| 508 | mock_stream_process.terminate.assert_called_once() |
| 509 | # The task should be cancelled |
| 510 | assert real_task.cancelled() or real_task.done() |
| 511 | mock_close_page.assert_called_once() |
| 512 | mock_browser.close.assert_called_once() |
| 513 | mock_pw.stop.assert_called_once() |
| 514 | |
| 515 | |
| 516 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected