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 parameters of the failure-recording test loggers

Browse source
This commit is contained in:
mateo-berri 2026-11-07 08:38:17 -08:00
parent ae89db72d2
commit 3d53f0047d
2 changed files with 14 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py
View file

@ -5925,17 +5925,21 @@ class _FailureLog(CustomLogger):
super().__init__()
self.rejections: tuple[str, ...] = ()
def _record(self, kwargs) -> None:
def _record(self, kwargs: Mapping[str, object]) -> None:
error: Final = str(kwargs.get("exception"))
self.rejections = (
*self.rejections,
next((phrase for phrase in _UPSTREAM_REJECTION_PHRASES if phrase in error), error),
)
def log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:
def log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:
async def async_log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)

10
tests/unit/llms/anthropic/messages/responses_adapters/test_responses_adapters_handler.py
View file

@ -1233,17 +1233,21 @@ class _FailureLog(CustomLogger):
super().__init__()
self.rejections: tuple[str, ...] = ()
def _record(self, kwargs) -> None:
def _record(self, kwargs: Mapping[str, object]) -> None:
error: Final = str(kwargs.get("exception"))
self.rejections = (
*self.rejections,
next((phrase for phrase in ("encrypted content", "organization") if phrase in error), error),
)
def log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:
def log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:
async def async_log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)