Function test_handle_connect_intercept

tests/api_utils/test_stream.py:75–112

(proxy_server)

@pytest.mark.asyncio
async def test_handle_connect_intercept(proxy_server):
    reader = AsyncMock()
    writer = AsyncMock()
    target = "api.test.com:443"

    # Mock cert manager
    proxy_server.cert_manager.get_domain_cert = MagicMock(
        return_value=("cert_path", "key_path")
    )

    # Mock loop and start_tls
    mock_loop = MagicMock()
    mock_transport = MagicMock()
    mock_loop.start_tls = AsyncMock(return_value=mock_transport)

    # Mock ssl context
    mock_ssl_ctx = MagicMock()

    with (
        patch("asyncio.get_running_loop", return_value=mock_loop),
        patch("ssl.create_default_context", return_value=mock_ssl_ctx),
        patch.object(
            proxy_server.proxy_connector, "create_connection", AsyncMock()
        ) as mock_create_conn,
        patch.object(
            proxy_server, "_forward_data_with_interception", AsyncMock()
        ) as mock_forward,
        patch("asyncio.StreamWriter", MagicMock()),
    ):
        mock_create_conn.return_value = (AsyncMock(), AsyncMock())

        await proxy_server._handle_connect(reader, writer, target)

        proxy_server.cert_manager.get_domain_cert.assert_called_with("api.test.com")
        mock_ssl_ctx.load_cert_chain.assert_called()
        writer.write.assert_called()
        mock_create_conn.assert_called()
        mock_forward.assert_called()


@pytest.mark.asyncio