scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

test(http): annotate finalizer regression locals

Browse source
This commit is contained in:
King Star 2026-09-10 13:29:34 +08:00
parent c895e80c90
commit ca25e3be82
No known key found for this signature in database
2 changed files with 28 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
tests/test_litellm/caching/test_llm_client_cache_e2e.py
View file

@ -139,12 +139,12 @@ async def test_evicted_openai_client_finalizes_aiohttp_session():
from litellm.llms.custom_httpx.aiohttp_transport import LiteLLMAiohttpTransport
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
transport = AsyncHTTPHandler._create_async_transport()
transport: Final = AsyncHTTPHandler._create_async_transport()
assert isinstance(transport, LiteLLMAiohttpTransport)
session_ref = weakref.ref(transport._get_valid_client_session())
http_client = httpx.AsyncClient(transport=transport)
client = AsyncOpenAI(api_key="sk-test", http_client=http_client)
transport_ref = weakref.ref(transport)
session_ref: Final = weakref.ref(transport._get_valid_client_session())
http_client: Final = httpx.AsyncClient(transport=transport)
client: Final = AsyncOpenAI(api_key="sk-test", http_client=http_client)
transport_ref: Final = weakref.ref(transport)
warning: Final[asyncio.Future[str]] = asyncio.get_running_loop().create_future()
def exception_handler(_loop: asyncio.AbstractEventLoop, context: dict[str, object]) -> None:
@ -153,7 +153,7 @@ async def test_evicted_openai_client_finalizes_aiohttp_session():
asyncio.get_running_loop().set_exception_handler(exception_handler)
cache = litellm.in_memory_llm_clients_cache
cache: Final = litellm.in_memory_llm_clients_cache
cache.set_cache("openai-aiohttp-client", client, ttl=600)
cache.set_cache("filler", "x", ttl=600)

44
tests/test_litellm/llms/custom_httpx/test_aiohttp_transport.py
View file

@ -45,8 +45,8 @@ async def test_aclose_closes_owned_session():
@pytest.mark.asyncio
async def test_finalizer_does_not_close_shared_session():
"""Finalization must preserve a session owned by the caller."""
session = aiohttp.ClientSession()
transport = LiteLLMAiohttpTransport(client=session, owns_session=False)
session: Final = aiohttp.ClientSession()
transport: Final = LiteLLMAiohttpTransport(client=session, owns_session=False)
del transport
gc.collect()
@ -896,8 +896,8 @@ def _make_session_on_dead_loop() -> aiohttp.ClientSession:
def test_finalizer_closes_owned_session_after_loop_closes():
"""Finalization must synchronously dispose a session whose loop is already closed."""
session = _make_session_on_dead_loop()
transport = LiteLLMAiohttpTransport(client=session)
session: Final = _make_session_on_dead_loop()
transport: Final = LiteLLMAiohttpTransport(client=session)
try:
del transport
@ -910,8 +910,8 @@ def test_finalizer_closes_owned_session_after_loop_closes():
def test_finalizer_close_failure_falls_back_to_sync_teardown():
"""A failed foreign-loop close future must still dispose the session connector."""
session = _make_session_on_dead_loop()
future: "concurrent.futures.Future[None]" = concurrent.futures.Future()
session: Final = _make_session_on_dead_loop()
future: Final[concurrent.futures.Future[None]] = concurrent.futures.Future()
future.set_exception(RuntimeError("simulated close failure"))
try:
@ -926,9 +926,9 @@ def test_finalizer_watcher_handles_loop_inspection_failure():
"""A loop inspection error must not strand a finalizer session."""
from unittest.mock import Mock
session = _make_session_on_dead_loop()
future: "concurrent.futures.Future[None]" = concurrent.futures.Future()
broken_loop = cast(asyncio.AbstractEventLoop, Mock())
session: Final = _make_session_on_dead_loop()
future: Final[concurrent.futures.Future[None]] = concurrent.futures.Future()
broken_loop: Final = cast(asyncio.AbstractEventLoop, Mock())
broken_loop.is_running.side_effect = RuntimeError("simulated loop inspection failure")
try:
@ -945,12 +945,12 @@ async def test_finalizer_handoff_failure_closes_session_synchronously():
"""A foreign-loop handoff race must close the session when scheduling fails."""
from unittest.mock import Mock, patch
session = aiohttp.ClientSession()
original_loop = session._loop
foreign_loop = cast(asyncio.AbstractEventLoop, Mock())
session: Final = aiohttp.ClientSession()
original_loop: Final = session._loop
foreign_loop: Final = cast(asyncio.AbstractEventLoop, Mock())
foreign_loop.is_running.return_value = True
session._loop = foreign_loop
transport = LiteLLMAiohttpTransport(client=session)
transport: Final = LiteLLMAiohttpTransport(client=session)
try:
with patch( # test-quality-ok: inject handoff failure to verify finalizer fallback
@ -969,9 +969,9 @@ def test_connector_close_failure_is_suppressed():
"""Synchronous connector teardown must remain best effort in a finalizer."""
from unittest.mock import Mock
connector = Mock()
connector: Final = Mock()
connector._close.side_effect = RuntimeError("simulated connector failure")
session = Mock()
session: Final = Mock()
session._connector = connector
LiteLLMAiohttpTransport._mark_connector_closed(cast(aiohttp.ClientSession, session))
@ -1158,13 +1158,13 @@ async def test_finalizer_closes_session_on_foreign_running_loop():
import threading
import time
ready = threading.Event()
blocked = threading.Event()
release = threading.Event()
ready: Final = threading.Event()
blocked: Final = threading.Event()
release: Final = threading.Event()
state: Final[queue.Queue[tuple[asyncio.AbstractEventLoop, aiohttp.ClientSession]]] = queue.Queue()
def worker() -> None:
loop = asyncio.new_event_loop()
loop: Final = asyncio.new_event_loop()
async def make() -> None:
state.put((loop, aiohttp.ClientSession()))
@ -1179,19 +1179,19 @@ async def test_finalizer_closes_session_on_foreign_running_loop():
loop.run_forever()
loop.close()
thread = threading.Thread(target=worker, daemon=True)
thread: Final = threading.Thread(target=worker, daemon=True)
thread.start()
assert ready.wait(5), "worker loop failed to start"
loop, session = state.get(timeout=5)
assert blocked.wait(5), "worker loop failed to enter its blocking callback"
transport = LiteLLMAiohttpTransport(client=session)
transport: Final = LiteLLMAiohttpTransport(client=session)
try:
transport.__del__()
loop.call_soon_threadsafe(loop.stop)
release.set()
deadline = time.monotonic() + 5
deadline: Final = time.monotonic() + 5
while (
(not session.closed or LiteLLMAiohttpTransport._finalizer_sessions)
and time.monotonic() < deadline