scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-12 03:48:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 57

test(unit): flush queued logging tasks after each test so systemone callback checks stay isolated

Browse source 
Ports the pytest_runtest_call hook from tests/unit/conftest.py on main (#44146). Without it the asystemone call type test picks up adecisions callbacks queued by earlier tests in the same worker

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
(cherry picked from commit 23fec69be2)
This commit is contained in:
mateo 2026-10-11 00:59:15 +00:00
parent acf7455911
commit cd141c6d36
1 changed files with 26 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
tests/unit/conftest.py
View file

@ -47,6 +47,7 @@ import litellm.utils as litellm_utils_module # noqa: E402 # same import-time d
from litellm._logging import ALL_LOGGERS # noqa: E402 # same import-time dependency
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER # noqa: E402 # same import-time dependency
from litellm.anthropic_beta_headers_manager import reload_beta_headers_config # noqa: E402 # same import-time dependency
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER # noqa: E402 # same import-time dependency
from litellm.litellm_core_utils.prompt_templates import factory as prompt_factory_module # noqa: E402 # same import-time dependency
from litellm.litellm_core_utils.prompt_templates import ( # noqa: E402 # same import-time dependency
image_handling as image_handling_module,
@ -159,6 +160,31 @@ def _run_coroutine_if_needed(result: object) -> None:
loop.create_task(coroutine)
async def _complete_test_logging() -> None:
await GLOBAL_LOGGING_WORKER.flush()
await GLOBAL_LOGGING_WORKER.stop()
def _flush_completed_test_logging() -> None:
queue: Final = GLOBAL_LOGGING_WORKER._queue
if queue is None or queue._unfinished_tasks == 0:
return
worker_loop: Final = GLOBAL_LOGGING_WORKER._bound_loop
if worker_loop is None or worker_loop.is_closed():
_run_coroutine_if_needed(_complete_test_logging())
return
if worker_loop.is_running():
asyncio.run_coroutine_threadsafe(_complete_test_logging(), worker_loop).result()
return
worker_loop.run_until_complete(_complete_test_logging())
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call() -> Iterator[None]:
yield
_flush_completed_test_logging()
def _close_handler_if_needed(handler: object) -> None:
close: Final = getattr(handler, "close", None)
if not callable(close):