fix(tests): keep foreign asyncio task noise out of the budget limiter caplog asserts

The two gc.collect()-under-caplog tests in test_budget_limiter_hotpath.py asserted
on every captured record, so a LoggingWorker task another test had parked on the
session fixture loop got destroyed inside their window and asyncio's "Task was
destroyed but it is pending!" error failed them. They now assert only the limiter
logger's records and park a pending task on a closed loop in the window themselves,
so the noise is exercised on every run

The edenai and sail spend_capture fixtures no longer start the worker from the
session loop: the test's own call and its flush bind the worker to the test loop

tests/unit/llms/edenai/conftest.py

@@ -2,7 +2,6 @@ import asyncio
 import uuid
 
 import pytest
-import pytest_asyncio
 
 import litellm
 from litellm.integrations.custom_logger import CustomLogger
@@ -40,9 +39,8 @@ class SpendCapture(CustomLogger):
         await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10.0)
 
 
-@pytest_asyncio.fixture
-async def spend_capture(monkeypatch) -> SpendCapture:
-    GLOBAL_LOGGING_WORKER.start()  # rebinds the worker's queue to this test's event loop
+@pytest.fixture
+def spend_capture(monkeypatch) -> SpendCapture:
     capture = SpendCapture(call_id=f"eden-{uuid.uuid4()}")
     monkeypatch.setattr(litellm, "callbacks", [capture])
     return capture

tests/unit/llms/sail/conftest.py

@@ -4,11 +4,9 @@ from typing import Final
 
 import httpx
 import pytest
-import pytest_asyncio
 import respx
 
 import litellm
-from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from tests.unit.llms.sail.helpers import SAIL_API_BASE, SpendCapture, chat_completion_body
 
 
@@ -26,9 +24,8 @@ def sail_env(local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch) -> Ite
     litellm.in_memory_llm_clients_cache.flush_cache()
 
 
-@pytest_asyncio.fixture
-async def spend_capture(monkeypatch: pytest.MonkeyPatch) -> SpendCapture:
-    GLOBAL_LOGGING_WORKER.start()
+@pytest.fixture
+def spend_capture(monkeypatch: pytest.MonkeyPatch) -> SpendCapture:
     capture: Final = SpendCapture(call_id=f"sail-{uuid.uuid4()}")
     monkeypatch.setattr(litellm, "callbacks", [capture])
     return capture

tests/unit/router_strategy/test_budget_limiter_hotpath.py

@@ -1,6 +1,7 @@
 import asyncio
 import gc
 import logging
+import threading
 from types import SimpleNamespace
 from typing import Final
 from unittest.mock import AsyncMock, MagicMock
@@ -8,6 +9,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 import litellm
+from litellm._logging import verbose_router_logger
 from litellm.caching.caching import DualCache
 from litellm.caching.redis_cache import RedisCache, RedisCircuitBreakerOpenError
 from litellm.router_strategy.budget_limiter import RouterBudgetLimiting
@@ -300,6 +302,27 @@ def test_router_add_deployment_registers_deployment_budget(disable_budget_sync,
     assert config.max_budget == 0.000000000001
 
 
+def _leave_a_task_pending_on_a_closed_loop() -> None:
+    def park_a_task_then_close_the_loop() -> None:
+        loop: Final = asyncio.new_event_loop()
+
+        async def park() -> None:
+            await loop.create_future()
+
+        loop.create_task(park())
+        loop.call_soon(loop.stop)
+        loop.run_forever()
+        loop.close()
+
+    parker: Final = threading.Thread(target=park_a_task_then_close_the_loop)
+    parker.start()
+    parker.join()
+
+
+def _limiter_error_lines(caplog: pytest.LogCaptureFixture) -> list[str]:
+    return [record.getMessage() for record in caplog.records if record.name == verbose_router_logger.name]
+
+
 @pytest.mark.asyncio
 async def test_sync_refused_by_the_open_circuit_breaker_is_quiet_and_leaks_no_task(disable_budget_sync, caplog):
     """The budget sync runs every second, so an open breaker must not add an error line or an unretrieved task exception per cycle."""
@@ -321,11 +344,12 @@ async def test_sync_refused_by_the_open_circuit_breaker_is_quiet_and_leaks_no_ta
         with caplog.at_level(logging.ERROR):
             await limiter._sync_in_memory_spend_with_redis()
             await asyncio.sleep(0)
+            _leave_a_task_pending_on_a_closed_loop()
             gc.collect()
     finally:
         loop.set_exception_handler(None)
 
-    assert caplog.records == []
+    assert _limiter_error_lines(caplog) == []
     unretrieved.assert_not_called()
     assert limiter.redis_increment_operation_queue == [
         {"key": "provider_spend:openai:1d", "increment_value": 0.5, "ttl": 60}
@@ -381,11 +405,12 @@ async def test_push_task_failure_is_logged_once_and_not_leaked(disable_budget_sy
         with caplog.at_level(logging.ERROR):
             await limiter._push_in_memory_increments_to_redis()
             await asyncio.sleep(0)
+            _leave_a_task_pending_on_a_closed_loop()
             gc.collect()
     finally:
         loop.set_exception_handler(None)
 
-    assert [record.getMessage() for record in caplog.records] == [
+    assert _limiter_error_lines(caplog) == [
         "Error syncing in-memory cache with Redis: Error 61 connecting to 127.0.0.1:6379"
     ]
     unretrieved.assert_not_called()