Test _start_stream_proxy timeout waiting for READY.
()
| 340 | |
| 341 | @pytest.mark.asyncio |
| 342 | async def test_start_stream_proxy_timeout(): |
| 343 | """Test _start_stream_proxy timeout waiting for READY.""" |
| 344 | with ( |
| 345 | patch("api_utils.app.get_environment_variable", return_value="3120"), |
| 346 | patch("multiprocessing.Queue") as mock_queue_cls, |
| 347 | patch("multiprocessing.Process"), |
| 348 | patch("server.logger"), |
| 349 | patch("asyncio.to_thread", side_effect=queue.Empty), |
| 350 | ): |
| 351 | mock_queue = MagicMock() |
| 352 | mock_queue_cls.return_value = mock_queue |
| 353 | |
| 354 | with pytest.raises(RuntimeError, match="STREAM proxy failed to start in time"): |
| 355 | await _start_stream_proxy() |
| 356 | |
| 357 | |
| 358 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected