test(prism): capture requests through respx instead of appending to a list and swapping the client transport

tests/test_litellm/llms/openai_like/test_prism_provider.py

@@ -1,19 +1,12 @@
 import json
 from pathlib import Path
+from typing import Final
 
-import httpx
 import pytest
+import respx
 
 import litellm
-from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
-
-
-def _prism_client(requests: list[httpx.Request], response_body: dict[str, object]) -> HTTPHandler:
-    def respond(request: httpx.Request) -> httpx.Response:
-        requests.append(request)
-        return httpx.Response(200, json=response_body)
-
-    return HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond)))
+from litellm.caching.llm_caching_handler import LLMClientCache
 
 
 def test_prism_provider_resolution(monkeypatch: pytest.MonkeyPatch):
@@ -126,47 +119,36 @@ def test_prism_supported_endpoints():
 
 
 def test_prism_responses_request():
-    requests: list[httpx.Request] = []
-    client = _prism_client(
-        requests,
-        {
-            "id": "resp_prism",
-            "object": "response",
-            "created_at": 1_789_550_000,
-            "model": "deepseek-v4-flash",
-            "status": "completed",
-            "output": [
-                {
-                    "id": "msg_prism",
-                    "type": "message",
-                    "role": "assistant",
-                    "status": "completed",
-                    "content": [
-                        {
-                            "type": "output_text",
-                            "text": "Hello from Prism",
-                            "annotations": [],
-                        }
-                    ],
-                }
-            ],
-            "usage": {
-                "input_tokens": 4,
-                "output_tokens": 3,
-                "total_tokens": 7,
+    with respx.mock() as upstream:
+        route: Final = upstream.post("https://api.prisminference.com/v1/responses").respond(
+            200,
+            json={
+                "id": "resp_prism",
+                "object": "response",
+                "created_at": 1_789_550_000,
+                "model": "deepseek-v4-flash",
+                "status": "completed",
+                "output": [
+                    {
+                        "id": "msg_prism",
+                        "type": "message",
+                        "role": "assistant",
+                        "status": "completed",
+                        "content": [{"type": "output_text", "text": "Hello from Prism", "annotations": []}],
+                    }
+                ],
+                "usage": {"input_tokens": 4, "output_tokens": 3, "total_tokens": 7},
             },
-        },
-    )
+        )
+        response: Final = litellm.responses(
+            model="prism/deepseek-v4-flash",
+            input="Say hello",
+            api_key="prism-test-key",
+        )
 
-    response = litellm.responses(
-        model="prism/deepseek-v4-flash",
-        input="Say hello",
-        api_key="prism-test-key",
-        client=client,
-    )
-
-    request = requests[0]
-    body = json.loads(request.content)
+    request: Final = route.calls.last.request
+    body: Final = json.loads(request.content)
+    assert route.call_count == 1
     assert str(request.url) == "https://api.prisminference.com/v1/responses"
     assert request.headers["authorization"] == "Bearer prism-test-key"
     assert body["model"] == "deepseek-v4-flash"
@@ -175,37 +157,33 @@ def test_prism_responses_request():
 
 
 @pytest.mark.asyncio
-async def test_prism_anthropic_messages_request():
-    requests: list[httpx.Request] = []
-    response_body: dict[str, object] = {
-        "id": "msg_prism",
-        "type": "message",
-        "role": "assistant",
-        "model": "deepseek-v4-flash",
-        "content": [{"type": "text", "text": "Hello from Prism"}],
-        "stop_reason": "end_turn",
-        "stop_sequence": None,
-        "usage": {"input_tokens": 4, "output_tokens": 3},
-    }
+async def test_prism_anthropic_messages_request(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "in_memory_llm_clients_cache", LLMClientCache())
+    with respx.mock() as upstream:
+        route: Final = upstream.post("https://api.prisminference.com/v1/messages").respond(
+            200,
+            json={
+                "id": "msg_prism",
+                "type": "message",
+                "role": "assistant",
+                "model": "deepseek-v4-flash",
+                "content": [{"type": "text", "text": "Hello from Prism"}],
+                "stop_reason": "end_turn",
+                "stop_sequence": None,
+                "usage": {"input_tokens": 4, "output_tokens": 3},
+            },
+        )
+        response: Final = await litellm.anthropic.messages.acreate(
+            model="prism/deepseek-v4-flash",
+            messages=[{"role": "user", "content": "Say hello"}],
+            max_tokens=32,
+            api_key="prism-test-key",
+        )
 
-    def respond(request: httpx.Request) -> httpx.Response:
-        requests.append(request)
-        return httpx.Response(200, json=response_body)
-
-    client = AsyncHTTPHandler()
-    await client.close()
-    client.client = httpx.AsyncClient(transport=httpx.MockTransport(respond))
-
-    response = await litellm.anthropic.messages.acreate(
-        model="prism/deepseek-v4-flash",
-        messages=[{"role": "user", "content": "Say hello"}],
-        max_tokens=32,
-        api_key="prism-test-key",
-        client=client,
-    )
-
-    request = requests[0]
-    body = json.loads(request.content)
+    request: Final = route.calls.last.request
+    body: Final = json.loads(request.content)
+    assert route.call_count == 1
     assert str(request.url) == "https://api.prisminference.com/v1/messages"
     assert request.headers["authorization"] == "Bearer prism-test-key"
     assert request.headers["anthropic-version"] == "2023-06-01"