mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-08 03:08:45 +00:00
Merge pull request #38066 from BerriAI/litellm_vacuous_20260824
test(litellm_utils_tests): give the aiohttp transport tests teeth
This commit is contained in:
Mateo Wang
GitHub
commit
0f1145160c
2 changed files with 60 additions and 122 deletions
|
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"TQ001": {
|
||||
"limit": 736
|
||||
"limit": 733
|
||||
},
|
||||
"TQ002": {
|
||||
"limit": 742
|
||||
|
|
|
|||
|
|
@ -1,129 +1,67 @@
|
|||
import asyncio
|
||||
import copy
|
||||
import time
|
||||
from datetime import datetime
|
||||
from unittest import mock
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from litellm.types.utils import StandardCallbackDynamicParams
|
||||
|
||||
load_dotenv()
|
||||
import socket
|
||||
from typing import Final
|
||||
|
||||
import aiohttp
|
||||
import httpx
|
||||
import pytest
|
||||
from aiohttp import ClientSession
|
||||
|
||||
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"""
|
||||
def _closed_local_port() -> int:
|
||||
with socket.socket() as probe:
|
||||
probe.bind(("127.0.0.1", 0))
|
||||
return probe.getsockname()[1]
|
||||
|
||||
|
||||
async def test_client_session_helper() -> None:
|
||||
transport: Final = AsyncHTTPHandler._create_aiohttp_transport()
|
||||
assert isinstance(transport, LiteLLMAiohttpTransport)
|
||||
session1: Final = transport._get_valid_client_session()
|
||||
assert isinstance(session1, ClientSession)
|
||||
assert session1.closed is False
|
||||
assert getattr(session1, "_loop") is asyncio.get_running_loop()
|
||||
session2: Final = transport._get_valid_client_session()
|
||||
assert session2 is session1
|
||||
await session1.close()
|
||||
|
||||
|
||||
async def test_event_loop_robustness() -> None:
|
||||
transport: Final = AsyncHTTPHandler._create_aiohttp_transport()
|
||||
session: Final = transport._get_valid_client_session()
|
||||
assert isinstance(session, ClientSession)
|
||||
await session.close()
|
||||
session_after_close: Final = 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()
|
||||
session_after_factory: Final = 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
|
||||
await session_after_close.close()
|
||||
await session_after_factory.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("ssl_verify", "expected_ssl"), [(False, False), (None, True)])
|
||||
async def test_refused_connection_maps_to_httpx_connect_error(
|
||||
ssl_verify: bool | None, expected_ssl: bool, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("NO_PROXY", "127.0.0.1")
|
||||
transport: Final = AsyncHTTPHandler._create_aiohttp_transport(ssl_verify=ssl_verify)
|
||||
port: Final = _closed_local_port()
|
||||
request: Final = httpx.Request("GET", f"https://127.0.0.1:{port}/")
|
||||
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")
|
||||
|
||||
# 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__}")
|
||||
|
||||
# Call it again to test reuse
|
||||
session2 = transport._get_valid_client_session() # type: ignore
|
||||
print(f"✅ Second session call: {type(session2).__name__}")
|
||||
|
||||
# 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
|
||||
|
||||
|
||||
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()
|
||||
|
||||
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}")
|
||||
|
||||
# Test 2: Force recreation by setting client to a callable
|
||||
from aiohttp import ClientSession
|
||||
|
||||
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}")
|
||||
|
||||
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
|
||||
|
||||
|
||||
async def test_httpx_request_simulation():
|
||||
"""Test that the transport can handle a simulated HTTP request"""
|
||||
try:
|
||||
transport = AsyncHTTPHandler._create_aiohttp_transport()
|
||||
|
||||
if transport is not None:
|
||||
print("✅ Transport created for request simulation")
|
||||
|
||||
# Create a simple httpx request to test with
|
||||
import httpx
|
||||
|
||||
request = httpx.Request("GET", "https://httpbin.org/headers")
|
||||
|
||||
# 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}")
|
||||
|
||||
# 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")
|
||||
with pytest.raises(httpx.ConnectError) as raised:
|
||||
await transport.handle_async_request(request)
|
||||
finally:
|
||||
await transport._get_valid_client_session().close()
|
||||
cause: Final = raised.value.__cause__
|
||||
assert isinstance(cause, aiohttp.ClientConnectorError)
|
||||
assert cause.ssl is expected_ssl
|
||||
assert (cause.host, cause.port) == ("127.0.0.1", port)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue