diff --git a/tests/test_litellm/llms/openai_like/test_prism_provider.py b/tests/test_litellm/llms/openai_like/test_prism_provider.py index 58140aa5115..c1775c63dc9 100644 --- a/tests/test_litellm/llms/openai_like/test_prism_provider.py +++ b/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"