test(anthropic): use integration coverage for keepalive errors

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

tests/unit/proxy/anthropic_endpoints/test_answered_status.py

@@ -1,3 +1,5 @@
+from typing import Final
+
 import openai
 import pytest
 
@@ -18,9 +20,9 @@ def _anthropic_internal_server_error(provider_status_code: int | None) -> litell
 
 @pytest.mark.parametrize("code", [400, 500, 503])
 def test_a_proxy_exception_passes_through_unchanged(code: int) -> None:
-    exc = ProxyException(message="budget service unavailable", type="None", param=None, code=code)
+    exc: Final = ProxyException(message="budget service unavailable", type="None", param=None, code=code)
 
-    answered_exc = with_answered_status(exc)
+    answered_exc: Final = with_answered_status(exc)
 
     assert answered_exc is exc
     assert exc.code == str(code)
@@ -76,9 +78,9 @@ def test_a_proxy_exception_passes_through_unchanged(code: int) -> None:
 def test_a_litellm_exception_answers_the_status_anthropic_gives_its_failure(
     exc: openai.APIStatusError, answered: int, copy: bool
 ) -> None:
-    raised_status = exc.status_code
+    raised_status: Final = exc.status_code
 
-    answered_exc = with_answered_status(exc)
+    answered_exc: Final = with_answered_status(exc)
 
     assert (answered_exc is not exc) is copy
     assert (type(answered_exc), answered_exc.status_code, str(answered_exc)) == (type(exc), answered, str(exc))
@@ -101,14 +103,14 @@ def test_the_answered_copy_keeps_the_raised_exceptions_cause_and_traceback() ->
     with pytest.raises(litellm.InternalServerError) as raised:
         _raise_a_bedrock_overload_caused_by_a_reset()
 
-    answered_exc = with_answered_status(raised.value)
+    answered_exc: Final = with_answered_status(raised.value)
 
     assert isinstance(raised.value.__cause__, ConnectionError)
     assert (answered_exc.__cause__, answered_exc.__traceback__) == (raised.value.__cause__, raised.value.__traceback__)
 
 
 def test_an_exception_without_a_status_passes_through_unchanged() -> None:
-    exc = ValueError("no status")
+    exc: Final = ValueError("no status")
 
     assert with_answered_status(exc) is exc
     assert not hasattr(exc, "status_code")

tests/unit/proxy/test_common_request_processing.py

@@ -10979,53 +10979,6 @@ async def test_base_process_llm_request_audits_a_failure_that_lands_after_its_ke
     assert collected[-1] == b"data: [DONE]\n\n"
 
 
-@pytest.mark.asyncio
-async def test_anthropic_messages_answers_a_late_503_after_its_keepalive_as_an_api_error_event() -> None:
-    unavailable: Final = litellm.ServiceUnavailableError(
-        message='BedrockException - {"message": "Bedrock is unable to process your request."}',
-        llm_provider="bedrock",
-        model="global.anthropic.claude-opus-5-5",
-    )
-    release_failure: Final = asyncio.Event()
-
-    async def fail_after_keepalive(self: ProxyBaseLLMRequestProcessing, **kwargs: object) -> None:
-        await release_failure.wait()
-        raise unavailable
-
-    proxy_logging_obj: Final = MagicMock(spec=ProxyLogging)
-    proxy_logging_obj.post_call_failure_hook = AsyncMock(return_value=None)
-    processor: Final = ProxyBaseLLMRequestProcessing(
-        data={"model": "global.anthropic.claude-opus-5-5", "stream": True}
-    )
-
-    with patch.object(litellm, "sse_keepalive_ping_interval_seconds", 0.05):
-        with patch.object(ProxyBaseLLMRequestProcessing, "_process_llm_request", fail_after_keepalive):
-            response: Final = await processor.base_process_llm_request(
-                request=MagicMock(spec=Request),
-                fastapi_response=Response(),
-                user_api_key_dict=MagicMock(spec=UserAPIKeyAuth),
-                route_type="anthropic_messages",
-                proxy_logging_obj=proxy_logging_obj,
-                general_settings={},
-                proxy_config=MagicMock(spec=ProxyConfig),
-            )
-        body_iterator: Final = response.body_iterator
-        first_chunk: Final = await body_iterator.__anext__()
-        release_failure.set()
-        remaining_chunks: Final = [
-            chunk.encode("utf-8") if isinstance(chunk, str) else bytes(chunk) async for chunk in body_iterator
-        ]
-
-    assert proxy_logging_obj.post_call_failure_hook.await_args.kwargs["original_exception"] is unavailable
-    assert first_chunk == TTFT_PING
-    event_line, data_line = remaining_chunks[-1].decode().strip().split("\n")
-    assert event_line == "event: error"
-    error: Final = json.loads(data_line.removeprefix("data: "))
-    assert error["type"] == "error"
-    assert error["error"]["type"] == "api_error"
-    assert "Bedrock is unable to process your request." in error["error"]["message"]
-
-
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "deployment_keepalive, expect_ping",