test(litellm_utils_tests): give aiohttp transport tests real assertions

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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()