Function test_start_stream_proxy_success

tests/api_utils/test_app.py:310–338

Test _start_stream_proxy success path.

()

@pytest.mark.asyncio
async def test_start_stream_proxy_success():
    """Test _start_stream_proxy success path."""
    state.STREAM_QUEUE = None
    state.STREAM_PROCESS = None

    with (
        patch("api_utils.app.get_environment_variable") as mock_get_env,
        patch("multiprocessing.Queue") as mock_queue_cls,
        patch("multiprocessing.Process") as mock_process_cls,
    ):
        mock_get_env.side_effect = lambda key, default=None: {
            "STREAM_PORT": "3120",
            "UNIFIED_PROXY_CONFIG": "http://upstream:8080",
        }.get(key, default)

        mock_queue = MagicMock()
        mock_queue.get.return_value = "READY"
        mock_queue_cls.return_value = mock_queue

        mock_process = MagicMock()
        mock_process_cls.return_value = mock_process

        await _start_stream_proxy()

        assert state.STREAM_QUEUE is not None
        assert state.STREAM_PROCESS is not None
        mock_process.start.assert_called_once()
        # Verify queue.get was called (via asyncio.to_thread, so we check the mock)
        mock_queue.get.assert_called()


@pytest.mark.asyncio