diff --git a/tests/litellm_utils_tests/test_aiohttp_handler.py b/tests/litellm_utils_tests/test_aiohttp_handler.py index 9fdac5ca23d..0257660611f 100644 --- a/tests/litellm_utils_tests/test_aiohttp_handler.py +++ b/tests/litellm_utils_tests/test_aiohttp_handler.py @@ -4,6 +4,8 @@ import time from datetime import datetime from unittest import mock +import httpx +from aiohttp import ClientSession from dotenv import load_dotenv from litellm.types.utils import StandardCallbackDynamicParams @@ -13,117 +15,79 @@ load_dotenv() import pytest import litellm +from litellm.llms.custom_httpx.aiohttp_transport import LiteLLMAiohttpTransport from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler @pytest.mark.asyncio async def test_client_session_helper(): """Test that the client session helper handles event loop changes correctly""" - try: - # Create a transport with the new helper - transport = AsyncHTTPHandler._create_aiohttp_transport() - if transport is not None: - print("✅ Successfully created aiohttp transport with helper") + transport = AsyncHTTPHandler._create_aiohttp_transport() + assert isinstance(transport, LiteLLMAiohttpTransport) - # Test the helper function directly if it's a LiteLLMAiohttpTransport - if hasattr(transport, "_get_valid_client_session"): - session1 = transport._get_valid_client_session() # type: ignore - print(f"✅ First session created: {type(session1).__name__}") + session1 = transport._get_valid_client_session() + assert isinstance(session1, ClientSession) + assert session1.closed is False + assert getattr(session1, "_loop") is asyncio.get_running_loop() - # Call it again to test reuse - session2 = transport._get_valid_client_session() # type: ignore - print(f"✅ Second session call: {type(session2).__name__}") + # Within the same event loop the valid session is reused, not rebuilt + session2 = transport._get_valid_client_session() + assert session2 is session1 - # In the same event loop, should be the same session - print(f"✅ Same session reused: {session1 is session2}") - - return True - else: - print("ℹ️ No aiohttp transport available (probably missing httpx-aiohttp)") - return True - except Exception as e: - print(f"❌ Error: {e}") - import traceback - - traceback.print_exc() - return False + await session1.close() async def test_event_loop_robustness(): """Test behavior when event loops change (simulating CI/CD scenario)""" - try: - # Test session creation in multiple scenarios - transport = AsyncHTTPHandler._create_aiohttp_transport() + transport = AsyncHTTPHandler._create_aiohttp_transport() - if transport and hasattr(transport, "_get_valid_client_session"): - # Test 1: Normal usage - session = transport._get_valid_client_session() # type: ignore - print(f"✅ Normal session creation works: {session is not None}") + session = transport._get_valid_client_session() + assert isinstance(session, ClientSession) - # Test 2: Force recreation by setting client to a callable - from aiohttp import ClientSession + # A closed session must be replaced with a live one bound to this loop + await session.close() + session_after_close = transport._get_valid_client_session() + assert isinstance(session_after_close, ClientSession) + assert session_after_close is not session + assert session_after_close.closed is False - transport.client = lambda: ClientSession() # type: ignore - session2 = transport._get_valid_client_session() # type: ignore - print(f"✅ Session recreation after callable works: {session2 is not None}") + # A client that is a factory rather than a session must also be rebuilt + transport.client = lambda: ClientSession() # type: ignore[assignment] + session_after_factory = transport._get_valid_client_session() + assert isinstance(session_after_factory, ClientSession) + assert session_after_factory is not session_after_close + assert session_after_factory.closed is False + assert transport.client is session_after_factory - return True - else: - print("ℹ️ Transport not available or no helper method") - return True - - except Exception as e: - print(f"❌ Error in event loop robustness test: {e}") - import traceback - - traceback.print_exc() - return False + await session_after_close.close() + await session_after_factory.close() async def test_httpx_request_simulation(): """Test that the transport can handle a simulated HTTP request""" - try: - transport = AsyncHTTPHandler._create_aiohttp_transport() + transport = AsyncHTTPHandler._create_aiohttp_transport(ssl_verify=False) + request = httpx.Request("GET", "https://httpbin.org/headers") - if transport is not None: - print("✅ Transport created for request simulation") + # The per-request SSL override the request path reads must reflect ssl_verify + assert transport._ssl_verify is False - # Create a simple httpx request to test with - import httpx + session = transport._get_valid_client_session() + assert isinstance(session, ClientSession) + assert session.closed is False + assert callable(session.request) + assert session.connector is not None + assert session.connector._ssl is False - request = httpx.Request("GET", "https://httpbin.org/headers") + with mock.patch.object( + transport, "_make_aiohttp_request", new=mock.AsyncMock(side_effect=RuntimeError("boom")) + ) as mocked_request: + with pytest.raises(RuntimeError): + await transport.handle_async_request(request) - # Just test that we can get a valid session for this request context - if hasattr(transport, "_get_valid_client_session"): - session = transport._get_valid_client_session() # type: ignore - print(f"✅ Got valid session for request: {session is not None}") + assert mocked_request.call_count == 1 + call_kwargs = mocked_request.call_args.kwargs + assert call_kwargs["request"] is request + assert call_kwargs["ssl_verify"] is False + assert call_kwargs["client_session"] is session - # Test that session has required aiohttp methods - has_request_method = hasattr(session, "request") - print(f"✅ Session has request method: {has_request_method}") - - return has_request_method - - return True - else: - print("ℹ️ No transport available for request simulation") - return True - - except Exception as e: - print(f"❌ Error in request simulation: {e}") - return False - - -if __name__ == "__main__": - print("Testing client session helper and event loop handling fix...") - - result1 = asyncio.run(test_client_session_helper()) - result2 = asyncio.run(test_event_loop_robustness()) - result3 = asyncio.run(test_httpx_request_simulation()) - - if result1 and result2 and result3: - print( - "🎉 All tests passed! The helper function approach should fix the CI/CD event loop issues." - ) - else: - print("💥 Some tests failed") + await session.close()