diff --git a/tests/test_litellm/caching/test_llm_client_cache_e2e.py b/tests/test_litellm/caching/test_llm_client_cache_e2e.py index a8fb959b0d7..1d69e7e63eb 100644 --- a/tests/test_litellm/caching/test_llm_client_cache_e2e.py +++ b/tests/test_litellm/caching/test_llm_client_cache_e2e.py @@ -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) diff --git a/tests/test_litellm/llms/custom_httpx/test_aiohttp_transport.py b/tests/test_litellm/llms/custom_httpx/test_aiohttp_transport.py index e303100fd4f..641f1688dc5 100644 --- a/tests/test_litellm/llms/custom_httpx/test_aiohttp_transport.py +++ b/tests/test_litellm/llms/custom_httpx/test_aiohttp_transport.py @@ -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