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 
Backport of 3d53f0047d (#49076) to lts/1.108.x.
This commit is contained in:
mateo-berri 2026-11-07 08:46:14 -08:00
parent a1c02f7d1a
commit 5a6475d112
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/pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py
View file

@ -5589,17 +5589,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/pass_through/responses_adapters/test_responses_adapters_handler.py
View file

@ -804,17 +804,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)