scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 174

test(anthropic): type the success logger and use a fixed token expiry in the bridge tests

Browse source
This commit is contained in:
mateo-berri 2026-11-07 08:55:37 -08:00
parent 5a6475d112
commit c72cb031d8
2 changed files with 13 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py
View file

@ -1,5 +1,4 @@
import json
import time
from datetime import datetime
from pathlib import Path
from typing import Final
@ -381,9 +380,12 @@ def _two_reasoning_tool_turn_stream() -> httpx.Response:
return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=body)
_FAR_FUTURE_EXPIRY: Final = 4_102_444_800
@pytest.fixture
def _chatgpt_backend(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter) -> None:
auth: Final = {"access_token": "chatgpt-test-token", "expires_at": time.time() + 3600, "account_id": "acct_test"}
auth: Final = {"access_token": "chatgpt-test-token", "expires_at": _FAR_FUTURE_EXPIRY, "account_id": "acct_test"}
(tmp_path / "auth.json").write_text(json.dumps(auth))
monkeypatch.setenv("CHATGPT_TOKEN_DIR", str(tmp_path))
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)

13
tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_handler.py
View file

@ -3,7 +3,7 @@ import datetime
import json
import os
import sys
from collections.abc import AsyncIterator, Mapping
from collections.abc import AsyncIterator, Mapping, Sequence
from contextlib import nullcontext
from typing import Final, cast
from unittest.mock import AsyncMock, patch
@ -28,6 +28,7 @@ from litellm.llms.anthropic.pass_through.responses_adapters.handler import (
)
from litellm.router import Router
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS, converted_stream_requested
from litellm.types.utils import StandardLoggingPayload
MESSAGES = [{"role": "user", "content": "hello"}]
CLAUDE_CODE_USER_ID = json.dumps({"device_id": "d" * 64, "account_uuid": "", "session_id": "session-abc"})
@ -916,9 +917,13 @@ class _SuccessLog(CustomLogger):
self.logged_reasoning_item_ids: tuple[tuple[object, ...], ...] = ()
self.logged: Final = asyncio.Event()
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:
logged: Final = kwargs["standard_logging_object"]["response"]
items: Final = logged["choices"][0]["message"].get("reasoning_items") or ()
async def async_log_success_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
payload: Final = cast(StandardLoggingPayload, kwargs["standard_logging_object"])
logged: Final = cast(Mapping[str, Sequence[Mapping[str, Mapping[str, object]]]], payload["response"])
message: Final = logged["choices"][0]["message"]
items: Final = cast(Sequence[Mapping[str, object]], message.get("reasoning_items") or ())
self.logged_reasoning_item_ids = (*self.logged_reasoning_item_ids, tuple(item["id"] for item in items))
self.logged.set()