diff --git a/tests/_live_test_helpers.py b/tests/_live_test_helpers.py new file mode 100644 index 0000000000..a79b81e82c --- /dev/null +++ b/tests/_live_test_helpers.py @@ -0,0 +1,10 @@ +import os + +import pytest + + +def _skip_live_prompt_caching_test(): + if os.environ.get("LITELLM_RUN_LIVE_PROMPT_CACHING_TESTS") != "1": + pytest.skip("Live prompt-caching E2E tests are opt-in") + if os.environ.get("CASSETTE_REDIS_URL"): + pytest.skip("Live prompt-caching E2E tests cannot run under VCR replay") diff --git a/tests/_vcr_conftest_common.py b/tests/_vcr_conftest_common.py index 995c333ad0..4d5a73779e 100644 --- a/tests/_vcr_conftest_common.py +++ b/tests/_vcr_conftest_common.py @@ -1930,6 +1930,25 @@ def emit_vcr_classification_summary(terminalreporter) -> None: continue terminalreporter.write_line(f" [{verdict}] {n}") + leak_verdicts = ( + VERDICT_PARTIAL, + VERDICT_MISS_OVERFLOW, + VERDICT_MISS_NOT_PERSISTED, + VERDICT_UNMARKED_LIVE_CALL, + ) + leak_counts = {verdict: counts.get(verdict, 0) for verdict in leak_verdicts} + total_leaks = sum(leak_counts.values()) + terminalreporter.write_sep("-", "VCR COST LEAK CHECK", bold=True) + if total_leaks: + rendered = ", ".join( + f"{verdict}={count}" for verdict, count in leak_counts.items() if count + ) + terminalreporter.write_line(f" FAIL: {rendered}") + else: + terminalreporter.write_line( + " PASS: no overflow, partial, not-persisted, or unmarked live-call verdicts" + ) + overflow = snapshot["overflow_tests"] if overflow: terminalreporter.write_sep( diff --git a/tests/litellm_utils_tests/conftest.py b/tests/litellm_utils_tests/conftest.py index d20203da3a..68c281a045 100644 --- a/tests/litellm_utils_tests/conftest.py +++ b/tests/litellm_utils_tests/conftest.py @@ -28,32 +28,9 @@ from tests._vcr_conftest_common import ( # noqa: E402,F401 _verbose_state = VerboseReporterState() +_VCR_INCOMPATIBLE_FILES = frozenset() -# Files where VCR replay breaks the test: -# - ``test_litellm_overhead.py``: asserts overhead/total < 40%, which -# inverts when cached replay collapses the upstream time to microseconds. -_VCR_INCOMPATIBLE_FILES = frozenset( - { - "test_litellm_overhead.py", - } -) - -# AWS Secrets Manager resource-lifecycle tests. Each run creates a secret -# under a per-run unique name (``litellm_test_``) and either asserts the -# API response echoes that exact unique name or reads it straight back. The -# name *must* be unique per run because AWS enforces a >=7-day deletion -# recovery window — a fixed name can't be re-created on the daily VCR -# re-record. Deterministic replay returns the previously-recorded (different) -# name, so the unique-name round-trip cannot be reproduced offline. The -# config-parsing tests in the same file (settings / STS endpoint) make no such -# unique-resource calls and stay VCR-cached. -_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = ( - "::test_write_and_read_simple_secret", - "::test_write_and_read_json_secret", - "::test_read_nonexistent_secret", - "::test_primary_secret_functionality", - "::test_write_secret_with_description_and_tags", -) +_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = () @pytest.fixture(scope="function", autouse=True) diff --git a/tests/litellm_utils_tests/test_aws_secret_manager.py b/tests/litellm_utils_tests/test_aws_secret_manager.py index 674f9b3ca8..46e8d00453 100644 --- a/tests/litellm_utils_tests/test_aws_secret_manager.py +++ b/tests/litellm_utils_tests/test_aws_secret_manager.py @@ -10,7 +10,6 @@ from dotenv import load_dotenv import litellm.types import litellm.types.utils - load_dotenv() import io @@ -52,6 +51,11 @@ def skip_on_throttling(func): def check_aws_credentials(): """Helper function to check if AWS credentials are set""" + if os.getenv("LITELLM_RUN_LIVE_AWS_SECRET_MANAGER_TESTS") != "1": + pytest.skip("Live AWS Secrets Manager E2E tests are opt-in") + if os.getenv("CASSETTE_REDIS_URL"): + pytest.skip("Live AWS Secrets Manager E2E tests cannot run under VCR replay") + required_vars = ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION_NAME"] missing_vars = [var for var in required_vars if not os.getenv(var)] if missing_vars: @@ -444,6 +448,11 @@ async def test_end_to_end_iam_role_secret_write(): - TEST_IAM_ROLE_ARN environment variable with ARN of a role that can be assumed - Proper AWS credentials configured (via instance profile, IAM role, or environment) """ + if os.getenv("LITELLM_RUN_LIVE_AWS_SECRET_MANAGER_TESTS") != "1": + pytest.skip("Live AWS Secrets Manager E2E tests are opt-in") + if os.getenv("CASSETTE_REDIS_URL"): + pytest.skip("Live AWS Secrets Manager E2E tests cannot run under VCR replay") + # Skip if TEST_IAM_ROLE_ARN is not set test_role_arn = os.getenv("TEST_IAM_ROLE_ARN") if not test_role_arn: diff --git a/tests/litellm_utils_tests/test_litellm_overhead.py b/tests/litellm_utils_tests/test_litellm_overhead.py index 3a428e9d58..95c376c24f 100644 --- a/tests/litellm_utils_tests/test_litellm_overhead.py +++ b/tests/litellm_utils_tests/test_litellm_overhead.py @@ -1,237 +1,185 @@ +import asyncio import json -import os -import sys import time -from contextlib import asynccontextmanager, contextmanager -from datetime import datetime -from unittest.mock import AsyncMock, patch, MagicMock + import httpx import pytest -import asyncio -sys.path.insert( - 0, os.path.abspath("../..") -) # Adds the parent directory to the system path import litellm +OPENAI_API_BASE = "https://example.openai.test/v1" -# Fake Vertex AI Gemini response for mocking -FAKE_VERTEX_GEMINI_RESPONSE = { - "candidates": [ + +def _completion_payload(response_id="chatcmpl-test"): + return { + "id": response_id, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "Hello"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + } + + +def _stream_payload(response_id="chatcmpl-stream"): + chunks = [ { - "content": { - "parts": [{"text": "Hello! How can I help you today?"}], - "role": "model", - }, - "finishReason": "STOP", - } - ], - "usageMetadata": { - "promptTokenCount": 5, - "candidatesTokenCount": 8, - "totalTokenCount": 13, - }, -} + "id": response_id, + "object": "chat.completion.chunk", + "created": 1, + "model": "gpt-4o", + "choices": [ + { + "index": 0, + "delta": {"role": "assistant", "content": "Hello"}, + "finish_reason": None, + } + ], + }, + { + "id": response_id, + "object": "chat.completion.chunk", + "created": 1, + "model": "gpt-4o", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + }, + ] + return ( + "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks) + + "data: [DONE]\n\n" + ).encode() -def _make_fake_httpx_response(url: str) -> httpx.Response: - """Create a fake httpx.Response that looks like a Vertex AI Gemini response.""" - response = httpx.Response( - status_code=200, - json=FAKE_VERTEX_GEMINI_RESPONSE, - request=httpx.Request("POST", url), +def _mock_openai_completion_transport( + monkeypatch, *, stream=False, response_id="chatcmpl-test" +): + from litellm.llms.custom_httpx.aiohttp_transport import LiteLLMAiohttpTransport + + calls = {"count": 0} + + async def delayed_response(_transport, request): + calls["count"] += 1 + await asyncio.sleep(0.2) + if stream: + return httpx.Response( + 200, + content=_stream_payload(response_id), + headers={"content-type": "text/event-stream"}, + request=request, + ) + return httpx.Response( + 200, json=_completion_payload(response_id), request=request + ) + + monkeypatch.setattr( + LiteLLMAiohttpTransport, + "handle_async_request", + delayed_response, ) - return response + return calls -@asynccontextmanager -async def _vertex_ai_mocks(): - """Context manager that mocks Vertex AI auth and HTTP calls. - - Mocks at the httpx.AsyncClient.send level so that the - @track_llm_api_timing decorator on AsyncHTTPHandler.post still runs, - preserving the overhead measurement. - """ - fake_response = _make_fake_httpx_response( - "https://fake-vertex-endpoint/v1/models/gemini-1.5-flash:generateContent" - ) - - async def fake_send(self, request, **kwargs): - await asyncio.sleep(0.2) # simulate ~200ms network latency - return fake_response - - with ( - patch( - "litellm.llms.vertex_ai.vertex_llm_base.VertexBase._ensure_access_token_async", - new_callable=AsyncMock, - return_value=("Bearer fake-token", "fake-project"), - ), - patch.object( - httpx.AsyncClient, - "send", - new=fake_send, - ), - ): - yield - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - "model", - [ - "bedrock/mistral.mistral-7b-instruct-v0:2", - "openai/gpt-4o", - "openai/self_hosted", - "bedrock/anthropic.claude-3-5-haiku-20241022-v1:0", - "vertex_ai/gemini-1.5-flash", - ], -) -async def test_litellm_overhead_non_streaming(model): - """ - - Test we can see the litellm overhead and that it is less than 40% of the total request time - """ - - litellm._turn_on_debug() - start_time = datetime.now() - kwargs = { - "messages": [{"role": "user", "content": "Hello, world!"}], - "model": model, - } - ######################################################### - # Specific cases for models - ######################################################### - if model == "vertex_ai/gemini-1.5-flash": - kwargs["vertex_project"] = "fake-project" - kwargs["vertex_location"] = "us-central1" - if model == "openai/self_hosted": - kwargs["api_base"] = os.environ.get("FAKE_OPENAI_API_BASE") - - async def _run(): - return await litellm.acompletion(**kwargs) - - if model == "vertex_ai/gemini-1.5-flash": - async with _vertex_ai_mocks(): - response = await _run() - else: - response = await _run() - ######################################################### - # End of specific cases for models - ######################################################### - end_time = datetime.now() - total_time_ms = (end_time - start_time).total_seconds() * 1000 - print(response) - print(response._hidden_params) +def _assert_overhead_is_smaller_than_total(response, total_time_ms): litellm_overhead_ms = response._hidden_params["litellm_overhead_time_ms"] - # calculate percent of overhead caused by litellm overhead_percent = litellm_overhead_ms * 100 / total_time_ms - print("##########################\n") - print("total_time_ms", total_time_ms) - print("response litellm_overhead_ms", litellm_overhead_ms) - print("litellm overhead_percent {}%".format(overhead_percent)) - print("##########################\n") + assert litellm_overhead_ms > 0 assert litellm_overhead_ms < 1000 - - # latency overhead should be less than total request time - assert litellm_overhead_ms < (end_time - start_time).total_seconds() * 1000 - - # latency overhead should be under 40% of total request time + assert litellm_overhead_ms < total_time_ms assert overhead_percent < 40 - pass + +@pytest.fixture(autouse=True) +def reset_litellm_state(): + litellm.cache = None + litellm.success_callback = [] + litellm._async_success_callback = [] + litellm.failure_callback = [] + litellm.callbacks = [] + yield + litellm.cache = None + litellm.callbacks = [] @pytest.mark.asyncio -@pytest.mark.parametrize( - "model", - [ - "bedrock/mistral.mistral-7b-instruct-v0:2", - "openai/gpt-4o", - "bedrock/anthropic.claude-3-5-haiku-20241022-v1:0", - "openai/self_hosted", - ], -) -async def test_litellm_overhead_stream(model): +async def test_litellm_overhead_non_streaming(monkeypatch): + calls = _mock_openai_completion_transport( + monkeypatch, response_id="chatcmpl-non-stream" + ) - litellm._turn_on_debug() - start_time = datetime.now() - kwargs = { - "messages": [{"role": "user", "content": "Hello, world!"}], - "model": model, - "stream": True, - } - ######################################################### - # Specific cases for models - ######################################################### - if model == "openai/self_hosted": - kwargs["api_base"] = "https://exampleopenaiendpoint-production.up.railway.app/" - # warmup call for auth validation on vertex_ai models - await litellm.acompletion(**kwargs) + start_time = time.perf_counter() + response = await litellm.acompletion( + model="gpt-4o", + api_key="test-key", + api_base=OPENAI_API_BASE, + messages=[{"role": "user", "content": "Hello, world!"}], + ) + total_time_ms = (time.perf_counter() - start_time) * 1000 - response = await litellm.acompletion(**kwargs) - - async for chunk in response: - print() - - end_time = datetime.now() - total_time_ms = (end_time - start_time).total_seconds() * 1000 - print(response) - print(response._hidden_params) - litellm_overhead_ms = response._hidden_params["litellm_overhead_time_ms"] - # calculate percent of overhead caused by litellm - overhead_percent = litellm_overhead_ms * 100 / total_time_ms - print("##########################\n") - print("total_time_ms", total_time_ms) - print("response litellm_overhead_ms", litellm_overhead_ms) - print("litellm overhead_percent {}%".format(overhead_percent)) - print("##########################\n") - assert litellm_overhead_ms > 0 - assert litellm_overhead_ms < 1000 - - # latency overhead should be less than total request time - assert litellm_overhead_ms < (end_time - start_time).total_seconds() * 1000 - - # latency overhead should be under 40% of total request time - assert overhead_percent < 40 - - pass + assert calls["count"] == 1 + _assert_overhead_is_smaller_than_total(response, total_time_ms) @pytest.mark.asyncio -async def test_litellm_overhead_cache_hit(): - """ - Test that litellm overhead is tracked on cache hits. - Makes two identical requests and checks that the second one (cache hit) has overhead in hidden params. - """ +async def test_litellm_overhead_stream(monkeypatch): + calls = _mock_openai_completion_transport( + monkeypatch, stream=True, response_id="chatcmpl-stream" + ) + + start_time = time.perf_counter() + response = await litellm.acompletion( + model="gpt-4o", + api_key="test-key", + api_base=OPENAI_API_BASE, + messages=[{"role": "user", "content": "Hello, world!"}], + stream=True, + ) + + async for _chunk in response: + pass + + total_time_ms = (time.perf_counter() - start_time) * 1000 + + assert calls["count"] == 1 + _assert_overhead_is_smaller_than_total(response, total_time_ms) + + +@pytest.mark.asyncio +async def test_litellm_overhead_cache_hit(monkeypatch): from litellm.caching.caching import Cache - litellm._turn_on_debug() + calls = _mock_openai_completion_transport(monkeypatch, response_id="chatcmpl-cache") litellm.cache = Cache() - print("test2 for caching") - litellm.set_verbose = True + messages = [{"role": "user", "content": "Hello, world! Cache test"}] response1 = await litellm.acompletion( - model="gpt-4.1-nano", messages=messages, caching=True + model="gpt-4o", + api_key="test-key", + api_base=OPENAI_API_BASE, + messages=messages, + caching=True, ) - await asyncio.sleep(2) - # Wait for any pending background tasks to complete - pending_tasks = [task for task in asyncio.all_tasks() if not task.done()] - print("all pending tasks", pending_tasks) - if pending_tasks: - await asyncio.wait(pending_tasks, timeout=1.0) - + await asyncio.sleep(0.5) response2 = await litellm.acompletion( - model="gpt-4.1-nano", messages=messages, caching=True + model="gpt-4o", + api_key="test-key", + api_base=OPENAI_API_BASE, + messages=messages, + caching=True, ) - print("RESPONSE 1", response1) - print("RESPONSE 2", response2) + + assert calls["count"] == 1 assert response1.id == response2.id - - print("response 2 hidden params", response2._hidden_params) - assert "_response_ms" in response2._hidden_params - total_time_ms = response2._hidden_params["_response_ms"] + assert response2._hidden_params["litellm_overhead_time_ms"] > 0 assert ( - response2._hidden_params["litellm_overhead_time_ms"] > 0 - and response2._hidden_params["litellm_overhead_time_ms"] < total_time_ms + response2._hidden_params["litellm_overhead_time_ms"] + < response2._hidden_params["_response_ms"] ) diff --git a/tests/llm_translation/base_llm_unit_tests.py b/tests/llm_translation/base_llm_unit_tests.py index 77850dac45..fef1d23d86 100644 --- a/tests/llm_translation/base_llm_unit_tests.py +++ b/tests/llm_translation/base_llm_unit_tests.py @@ -30,6 +30,10 @@ from litellm.types.utils import Usage, ModelResponse from abc import ABC, abstractmethod from openai import OpenAI +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +from tests._live_test_helpers import _skip_live_prompt_caching_test # noqa: E402 + def _usage_format_tests(usage: litellm.Usage): """ @@ -960,6 +964,7 @@ class BaseLLMChatTest(ABC): @pytest.mark.flaky(retries=4, delay=1) def test_prompt_caching(self): + _skip_live_prompt_caching_test() print("test_prompt_caching") litellm.set_verbose = True from litellm.utils import supports_prompt_caching diff --git a/tests/llm_translation/conftest.py b/tests/llm_translation/conftest.py index d346dae430..dba3812ee1 100644 --- a/tests/llm_translation/conftest.py +++ b/tests/llm_translation/conftest.py @@ -39,13 +39,7 @@ from tests._vcr_conftest_common import ( # noqa: E402,F401 # itself run under a live cassette context. _VCR_AUTO_MARKER_SKIP_FILES = frozenset({"test_vcr_redis_persister.py"}) -# Tests that observe live cross-call provider state (e.g. prompt-cache -# warm-up between two consecutive calls); replay can't reproduce that state. -_VCR_INCOMPATIBLE_NODEID_SUFFIXES = ( - "::test_prompt_caching", - "TestBedrockInvokeNovaJson::test_json_response_pydantic_obj", - "::test_bedrock_converse__streaming_passthrough", -) +_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = () _verbose_state = VerboseReporterState() diff --git a/tests/llm_translation/test_bedrock_completion.py b/tests/llm_translation/test_bedrock_completion.py index aecd7bc699..9cf253c379 100644 --- a/tests/llm_translation/test_bedrock_completion.py +++ b/tests/llm_translation/test_bedrock_completion.py @@ -3220,6 +3220,11 @@ async def test_bedrock_converse__streaming_passthrough(monkeypatch): from litellm.integrations.custom_logger import CustomLogger import asyncio + if os.environ.get("LITELLM_RUN_LIVE_BEDROCK_PASSTHROUGH_TESTS") != "1": + pytest.skip("Live Bedrock passthrough E2E tests are opt-in") + if os.environ.get("CASSETTE_REDIS_URL"): + pytest.skip("Live Bedrock passthrough E2E tests cannot run under VCR replay") + class MockCustomLogger(CustomLogger): pass diff --git a/tests/llm_translation/test_bedrock_invoke_tests.py b/tests/llm_translation/test_bedrock_invoke_tests.py index 23f436d5b2..901b43542f 100644 --- a/tests/llm_translation/test_bedrock_invoke_tests.py +++ b/tests/llm_translation/test_bedrock_invoke_tests.py @@ -3,7 +3,6 @@ import pytest import sys import os - sys.path.insert( 0, os.path.abspath("../..") ) # Adds the parent directory to the system path @@ -41,6 +40,15 @@ class TestBedrockInvokeNovaJson(BaseLLMChatTest): f"Skipping non-JSON test: {request.function.__name__} does not contain 'json'" ) + def test_json_response_pydantic_obj(self): + if os.environ.get("LITELLM_RUN_LIVE_BEDROCK_NOVA_JSON_TESTS") != "1": + pytest.skip("Live Bedrock Nova response-schema E2E tests are opt-in") + if os.environ.get("CASSETTE_REDIS_URL"): + pytest.skip( + "Live Bedrock Nova response-schema E2E tests cannot run under VCR replay" + ) + super().test_json_response_pydantic_obj() + def test_nova_invoke_remove_empty_system_messages(): """Test that _remove_empty_system_messages removes empty system list.""" diff --git a/tests/local_testing/conftest.py b/tests/local_testing/conftest.py index 0831313c13..d45caec22d 100644 --- a/tests/local_testing/conftest.py +++ b/tests/local_testing/conftest.py @@ -57,13 +57,10 @@ from tests._vcr_conftest_common import ( # noqa: E402,F401 # blacklisting was masking valid cache opportunities. # Files where VCR replay breaks the test: -# - ``test_assistants.py``: polls fresh per-session run IDs that no cassette -# can match, so every CI run re-records and the suite times out. # - ``test_router_caching.py``: asserts upstream returns a *new* id per call, # which a deterministic cassette replay violates. _VCR_INCOMPATIBLE_FILES = frozenset( { - "test_assistants.py", "test_router_caching.py", } ) diff --git a/tests/local_testing/test_assistants.py b/tests/local_testing/test_assistants.py index ee1c8fb651..8dc4f9e48e 100644 --- a/tests/local_testing/test_assistants.py +++ b/tests/local_testing/test_assistants.py @@ -1,22 +1,13 @@ -# What is this? -## Unit Tests for OpenAI Assistants API -import json import os import sys -import traceback - -from dotenv import load_dotenv - -load_dotenv() -sys.path.insert( - 0, os.path.abspath("../..") -) # Adds the parent directory to the system path -import asyncio -import logging import pytest +from dotenv import load_dotenv from openai.types.beta.assistant import Assistant -from typing_extensions import override +from openai.types.beta.assistant_deleted import AssistantDeleted + +load_dotenv() +sys.path.insert(0, os.path.abspath("../..")) import litellm from litellm import create_thread, get_thread @@ -25,40 +16,264 @@ from litellm.llms.openai.openai import ( AsyncAssistantEventHandler, AsyncCursorPage, MessageData, - OpenAIAssistantsAPI, + OpenAIMessage as Message, + Run, + SyncCursorPage, + Thread, ) -from litellm.llms.openai.openai import OpenAIMessage as Message -from litellm.llms.openai.openai import SyncCursorPage, Thread -""" -V0 Scope: - -- Add Message -> `/v1/threads/{thread_id}/messages` -- Run Thread -> `/v1/threads/{thread_id}/run` -""" +ASSISTANT_INSTRUCTIONS = ( + "You are a personal math tutor. When asked a question, write and run Python " + "code to answer the question." +) +ASSISTANT_ID = "asst_test" +THREAD_ID = "thread_test" +MESSAGE_ID = "msg_test" +RUN_ID = "run_test" -def _add_azure_related_dynamic_params(data: dict) -> dict: - data["api_version"] = "2024-02-15-preview" - data["api_base"] = os.getenv("AZURE_AI_API_BASE") - data["api_key"] = os.getenv("AZURE_AI_API_KEY") +def _assistant(**overrides): + data = { + "id": ASSISTANT_ID, + "object": "assistant", + "created_at": 1, + "name": "Math Tutor", + "description": None, + "model": "gpt-4.1", + "instructions": ASSISTANT_INSTRUCTIONS, + "tools": [], + "metadata": {}, + "top_p": 1.0, + "temperature": 1.0, + "response_format": "auto", + } + data.update(overrides) + return Assistant(**data) + + +def _thread(thread_id=THREAD_ID): + return Thread(id=thread_id, object="thread", created_at=1, metadata={}) + + +def _message(thread_id=THREAD_ID): + return Message( + id=MESSAGE_ID, + object="thread.message", + created_at=1, + thread_id=thread_id, + role="user", + content=[ + { + "type": "text", + "text": {"value": "Hey, how's it going?", "annotations": []}, + } + ], + assistant_id=None, + run_id=None, + attachments=[], + metadata={}, + status="completed", + ) + + +def _run(thread_id=THREAD_ID, assistant_id=ASSISTANT_ID): + return Run( + id=RUN_ID, + object="thread.run", + created_at=1, + assistant_id=assistant_id, + thread_id=thread_id, + status="completed", + started_at=1, + expires_at=None, + cancelled_at=None, + failed_at=None, + completed_at=1, + last_error=None, + model="gpt-4.1", + instructions=ASSISTANT_INSTRUCTIONS, + tools=[], + metadata={}, + usage={"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + required_action=None, + incomplete_details=None, + temperature=1.0, + top_p=1.0, + max_prompt_tokens=None, + max_completion_tokens=None, + truncation_strategy={"type": "auto", "last_messages": None}, + response_format="auto", + tool_choice="auto", + parallel_tool_calls=True, + ) + + +def _sync_page(data): + first_id = data[0].id if data else None + return SyncCursorPage( + data=data, + object="list", + first_id=first_id, + last_id=first_id, + has_more=False, + ) + + +def _async_page(data): + first_id = data[0].id if data else None + return AsyncCursorPage( + data=data, + object="list", + first_id=first_id, + last_id=first_id, + has_more=False, + ) + + +class _FakeAssistantEventHandler(AssistantEventHandler): + def until_done(self): + return None + + +class _FakeAsyncAssistantEventHandler(AsyncAssistantEventHandler): + async def until_done(self): + return None + + +class _FakeAssistantStream: + def __enter__(self): + return _FakeAssistantEventHandler() + + def __exit__(self, exc_type, exc, tb): + return False + + +class _FakeAsyncAssistantStream: + async def __aenter__(self): + return _FakeAsyncAssistantEventHandler() + + async def __aexit__(self, exc_type, exc, tb): + return False + + +class _SyncAssistants: + def list(self, **_kwargs): + return _sync_page([_assistant()]) + + def create(self, **kwargs): + return _assistant(**kwargs) + + def delete(self, assistant_id): + return AssistantDeleted( + id=assistant_id, object="assistant.deleted", deleted=True + ) + + +class _AsyncAssistants: + async def list(self, **_kwargs): + return _async_page([_assistant()]) + + async def create(self, **kwargs): + return _assistant(**kwargs) + + async def delete(self, assistant_id): + return AssistantDeleted( + id=assistant_id, object="assistant.deleted", deleted=True + ) + + +class _SyncMessages: + def create(self, thread_id, **_kwargs): + return _message(thread_id) + + def list(self, thread_id): + return _sync_page([_message(thread_id)]) + + +class _AsyncMessages: + async def create(self, thread_id, **_kwargs): + return _message(thread_id) + + async def list(self, thread_id): + return _async_page([_message(thread_id)]) + + +class _SyncRuns: + def create_and_poll(self, thread_id, assistant_id, **_kwargs): + return _run(thread_id=thread_id, assistant_id=assistant_id) + + def stream(self, **_kwargs): + return _FakeAssistantStream() + + +class _AsyncRuns: + async def create_and_poll(self, thread_id, assistant_id, **_kwargs): + return _run(thread_id=thread_id, assistant_id=assistant_id) + + def stream(self, **_kwargs): + return _FakeAsyncAssistantStream() + + +class _SyncThreads: + def __init__(self): + self.messages = _SyncMessages() + self.runs = _SyncRuns() + + def create(self, **_kwargs): + return _thread() + + def retrieve(self, thread_id): + return _thread(thread_id) + + +class _AsyncThreads: + def __init__(self): + self.messages = _AsyncMessages() + self.runs = _AsyncRuns() + + async def create(self, **_kwargs): + return _thread() + + async def retrieve(self, thread_id): + return _thread(thread_id) + + +class _FakeBeta: + def __init__(self, *, async_mode): + self.assistants = _AsyncAssistants() if async_mode else _SyncAssistants() + self.threads = _AsyncThreads() if async_mode else _SyncThreads() + + +class _FakeAssistantClient: + def __init__(self, *, async_mode): + self.beta = _FakeBeta(async_mode=async_mode) + + +@pytest.fixture +def assistant_client(sync_mode): + return _FakeAssistantClient(async_mode=not sync_mode) + + +def _request_data(provider, assistant_client, **kwargs): + data = {"custom_llm_provider": provider, "client": assistant_client, **kwargs} + if provider == "azure": + data.update( + { + "api_version": "2024-02-15-preview", + "api_base": "https://example.azure.test", + "api_key": "test-key", + } + ) return data @pytest.mark.parametrize("provider", ["openai", "azure"]) -@pytest.mark.parametrize( - "sync_mode", - [True, False], -) +@pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio -async def test_get_assistants(provider, sync_mode): - data = { - "custom_llm_provider": provider, - } - if provider == "azure": - data = _add_azure_related_dynamic_params(data) +async def test_get_assistants(provider, sync_mode, assistant_client): + data = _request_data(provider, assistant_client) - if sync_mode == True: + if sync_mode: assistants = litellm.get_assistants(**data) assert isinstance(assistants, SyncCursorPage) else: @@ -67,276 +282,152 @@ async def test_get_assistants(provider, sync_mode): @pytest.mark.parametrize("provider", ["azure", "openai"]) -@pytest.mark.parametrize( - "sync_mode", - [True, False], -) +@pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio() -@pytest.mark.flaky(retries=3, delay=1) -async def test_create_delete_assistants(provider, sync_mode): - litellm.ssl_verify = False - litellm._turn_on_debug() - data = { - "custom_llm_provider": provider, - "model": "gpt-4.1", - "instructions": "You are a personal math tutor. When asked a question, write and run Python code to answer the question.", - "name": "Math Tutor", - "tools": [{"type": "code_interpreter"}], - } - if provider == "azure": - data = _add_azure_related_dynamic_params(data) +async def test_create_delete_assistants(provider, sync_mode, assistant_client): + data = _request_data( + provider, + assistant_client, + model="gpt-4.1", + instructions=ASSISTANT_INSTRUCTIONS, + name="Math Tutor", + tools=[{"type": "code_interpreter"}], + ) - if sync_mode == True: + if sync_mode: assistant = litellm.create_assistants(**data) - - print("New assistants", assistant) assert isinstance(assistant, Assistant) - assert ( - assistant.instructions - == "You are a personal math tutor. When asked a question, write and run Python code to answer the question." - ) + assert assistant.instructions == ASSISTANT_INSTRUCTIONS assert assistant.id is not None - # delete the created assistant - delete_data = { - "custom_llm_provider": provider, - "assistant_id": assistant.id, - } - if provider == "azure": - delete_data = _add_azure_related_dynamic_params(delete_data) - response = litellm.delete_assistant(**delete_data) - print("Response deleting assistant", response) + response = litellm.delete_assistant( + **_request_data( + provider, + assistant_client, + assistant_id=assistant.id, + ) + ) assert response.id == assistant.id else: assistant = await litellm.acreate_assistants(**data) - print("New assistants", assistant) assert isinstance(assistant, Assistant) - assert ( - assistant.instructions - == "You are a personal math tutor. When asked a question, write and run Python code to answer the question." - ) + assert assistant.instructions == ASSISTANT_INSTRUCTIONS assert assistant.id is not None - # delete the created assistant - delete_data = { - "custom_llm_provider": provider, - "assistant_id": assistant.id, - } - if provider == "azure": - delete_data = _add_azure_related_dynamic_params(delete_data) - response = await litellm.adelete_assistant(**delete_data) - print("Response deleting assistant", response) + response = await litellm.adelete_assistant( + **_request_data( + provider, + assistant_client, + assistant_id=assistant.id, + ) + ) assert response.id == assistant.id -@pytest.mark.parametrize("provider", ["openai", "azure"]) -@pytest.mark.parametrize("sync_mode", [True, False]) -@pytest.mark.asyncio -async def test_create_thread_litellm(sync_mode, provider) -> Thread: +async def _create_thread_litellm(sync_mode, provider, assistant_client) -> Thread: message: MessageData = {"role": "user", "content": "Hey, how's it going?"} # type: ignore - data = { - "custom_llm_provider": provider, - "message": [message], - } - if provider == "azure": - data = _add_azure_related_dynamic_params(data) + data = _request_data(provider, assistant_client, message=[message]) if sync_mode: new_thread = create_thread(**data) else: new_thread = await litellm.acreate_thread(**data) - assert isinstance( - new_thread, Thread - ), f"type of thread={type(new_thread)}. Expected Thread-type" - + assert isinstance(new_thread, Thread) return new_thread @pytest.mark.parametrize("provider", ["openai", "azure"]) @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio -async def test_get_thread_litellm(provider, sync_mode): - new_thread = test_create_thread_litellm(sync_mode, provider) +async def test_create_thread_litellm(sync_mode, provider, assistant_client): + await _create_thread_litellm(sync_mode, provider, assistant_client) - if asyncio.iscoroutine(new_thread): - _new_thread = await new_thread - else: - _new_thread = new_thread - data = { - "custom_llm_provider": provider, - "thread_id": _new_thread.id, - } - if provider == "azure": - data = _add_azure_related_dynamic_params(data) +@pytest.mark.parametrize("provider", ["openai", "azure"]) +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_get_thread_litellm(provider, sync_mode, assistant_client): + new_thread = await _create_thread_litellm(sync_mode, provider, assistant_client) + data = _request_data(provider, assistant_client, thread_id=new_thread.id) if sync_mode: received_thread = get_thread(**data) else: received_thread = await litellm.aget_thread(**data) - assert isinstance( - received_thread, Thread - ), f"type of thread={type(received_thread)}. Expected Thread-type" - return new_thread + assert isinstance(received_thread, Thread) @pytest.mark.parametrize("provider", ["openai", "azure"]) @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio -async def test_add_message_litellm(sync_mode, provider): +async def test_add_message_litellm(sync_mode, provider, assistant_client): + new_thread = await _create_thread_litellm(sync_mode, provider, assistant_client) message: MessageData = {"role": "user", "content": "Hey, how's it going?"} # type: ignore - new_thread = test_create_thread_litellm(sync_mode, provider) + data = _request_data(provider, assistant_client, thread_id=new_thread.id, **message) - if asyncio.iscoroutine(new_thread): - _new_thread = await new_thread - else: - _new_thread = new_thread - # add message to thread - message: MessageData = {"role": "user", "content": "Hey, how's it going?"} # type: ignore - - data = {"custom_llm_provider": provider, "thread_id": _new_thread.id, **message} - if provider == "azure": - data = _add_azure_related_dynamic_params(data) if sync_mode: added_message = litellm.add_message(**data) else: added_message = await litellm.a_add_message(**data) - print(f"added message: {added_message}") - assert isinstance(added_message, Message) -@pytest.mark.parametrize( - "provider", - [ - "azure", - "openai", - ], -) # -@pytest.mark.parametrize( - "sync_mode", - [ - True, - False, - ], -) -@pytest.mark.parametrize( - "is_streaming", - [True, False], -) # +@pytest.mark.parametrize("provider", ["azure", "openai"]) +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.parametrize("is_streaming", [True, False]) @pytest.mark.asyncio -@pytest.mark.flaky(retries=3, delay=1) -async def test_aarun_thread_litellm(sync_mode, provider, is_streaming): - """ - - Get Assistants - - Create thread - - Create run w/ Assistants + Thread - """ - import openai +async def test_aarun_thread_litellm( + sync_mode, provider, is_streaming, assistant_client +): + get_assistants_data = _request_data(provider, assistant_client) + if sync_mode: + assistants = litellm.get_assistants(**get_assistants_data) + else: + assistants = await litellm.aget_assistants(**get_assistants_data) - try: - get_assistants_data = { - "custom_llm_provider": provider, - } - if provider == "azure": - get_assistants_data = _add_azure_related_dynamic_params(get_assistants_data) - if sync_mode: - assistants = litellm.get_assistants(**get_assistants_data) + assistant_id = assistants.data[0].id + new_thread = await _create_thread_litellm(sync_mode, provider, assistant_client) + message: MessageData = {"role": "user", "content": "Hey, how's it going?"} # type: ignore + thread_data = _request_data(provider, assistant_client, thread_id=new_thread.id) + message_data = _request_data( + provider, assistant_client, thread_id=new_thread.id, **message + ) + + if sync_mode: + added_message = litellm.add_message(**message_data) + assert isinstance(added_message, Message) + + if is_streaming: + run = litellm.run_thread_stream(assistant_id=assistant_id, **thread_data) + with run as run: + assert isinstance(run, AssistantEventHandler) + run.until_done() else: - assistants = await litellm.aget_assistants(**get_assistants_data) + run = litellm.run_thread( + assistant_id=assistant_id, stream=is_streaming, **thread_data + ) + assert run.status == "completed" + messages = litellm.get_messages(**thread_data) + assert isinstance(messages.data[0], Message) + else: + added_message = await litellm.a_add_message(**message_data) + assert isinstance(added_message, Message) - ## get the first assistant ### - try: - assistant_id = assistants.data[0].id - except IndexError: - pytest.skip("No assistants found") - - new_thread = test_create_thread_litellm(sync_mode=sync_mode, provider=provider) - - if asyncio.iscoroutine(new_thread): - _new_thread = await new_thread + if is_streaming: + run = litellm.arun_thread_stream(assistant_id=assistant_id, **thread_data) + async with run as run: + assert isinstance(run, AsyncAssistantEventHandler) + await run.until_done() else: - _new_thread = new_thread - - thread_id = _new_thread.id - - # add message to thread - message: MessageData = {"role": "user", "content": "Hey, how's it going?"} # type: ignore - - data = {"custom_llm_provider": provider, "thread_id": _new_thread.id, **message} - if provider == "azure": - data = _add_azure_related_dynamic_params(data) - - if sync_mode: - added_message = litellm.add_message(**data) - - if is_streaming: - run = litellm.run_thread_stream(assistant_id=assistant_id, **data) - with run as run: - assert isinstance(run, AssistantEventHandler) - print(run) - run.until_done() - else: - run = litellm.run_thread( - assistant_id=assistant_id, stream=is_streaming, **data - ) - if run.status == "completed": - messages = litellm.get_messages( - thread_id=_new_thread.id, custom_llm_provider=provider - ) - assert isinstance(messages.data[0], Message) - elif ( - run.status == "failed" - and run.last_error - and "No connection matching model" in run.last_error.message - ): - pytest.skip(f"Azure deployment not found: {run.last_error.message}") - else: - pytest.fail( - "An unexpected error occurred when running the thread, {}".format( - run - ) - ) - - else: - added_message = await litellm.a_add_message(**data) - - if is_streaming: - run = litellm.arun_thread_stream(assistant_id=assistant_id, **data) - async with run as run: - print(f"run: {run}") - assert isinstance( - run, - AsyncAssistantEventHandler, - ) - print(run) - await run.until_done() - else: - run = await litellm.arun_thread( - custom_llm_provider=provider, - thread_id=thread_id, - assistant_id=assistant_id, - ) - - if run.status == "completed": - messages = await litellm.aget_messages( - thread_id=_new_thread.id, custom_llm_provider=provider - ) - assert isinstance(messages.data[0], Message) - elif ( - run.status == "failed" - and run.last_error - and "No connection matching model" in run.last_error.message - ): - pytest.skip(f"Azure deployment not found: {run.last_error.message}") - else: - pytest.fail( - "An unexpected error occurred when running the thread, {}".format( - run - ) - ) - except openai.APIError as e: - pass + run = await litellm.arun_thread( + custom_llm_provider=provider, + thread_id=new_thread.id, + assistant_id=assistant_id, + client=assistant_client, + ) + assert run.status == "completed" + messages = await litellm.aget_messages(**thread_data) + assert isinstance(messages.data[0], Message) diff --git a/tests/logging_callback_tests/conftest.py b/tests/logging_callback_tests/conftest.py index 6dde85f2ca..dedff9a5ae 100644 --- a/tests/logging_callback_tests/conftest.py +++ b/tests/logging_callback_tests/conftest.py @@ -42,14 +42,7 @@ _RESPX_CONFLICTING_FILES = frozenset( } ) -# Files where VCR replay breaks the test: -# - ``test_amazing_s3_logs.py``: vcrpy's boto3 stub intercepts a real S3 -# PUT/LIST round-trip the test asserts on, so the per-run id is never found. -_VCR_INCOMPATIBLE_FILES = frozenset( - { - "test_amazing_s3_logs.py", - } -) +_VCR_INCOMPATIBLE_FILES = frozenset() _VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = () diff --git a/tests/logging_callback_tests/test_amazing_s3_logs.py b/tests/logging_callback_tests/test_amazing_s3_logs.py index dab2a0cc0b..08b9ac7d01 100644 --- a/tests/logging_callback_tests/test_amazing_s3_logs.py +++ b/tests/logging_callback_tests/test_amazing_s3_logs.py @@ -1,6 +1,7 @@ import sys import os import io, asyncio +from collections import defaultdict # import logging # logging.basicConfig(level=logging.DEBUG) @@ -18,6 +19,60 @@ from litellm._logging import verbose_logger import logging +class _FakeS3Paginator: + def __init__(self, objects): + self.objects = objects + + def paginate(self, Bucket): + keys = sorted(self.objects[Bucket]) + if not keys: + return [{}] + return [{"Contents": [{"Key": key} for key in keys]}] + + +class _FakeS3Client: + def __init__(self): + self.objects = defaultdict(dict) + + def clear(self): + self.objects.clear() + + def put_object(self, Bucket, Key, Body, **_kwargs): + self.objects[Bucket][Key] = Body + return {"ResponseMetadata": {"HTTPStatusCode": 200}} + + def delete_object(self, Bucket, Key): + self.objects[Bucket].pop(Key, None) + return {"ResponseMetadata": {"HTTPStatusCode": 204}} + + def get_paginator(self, name): + assert name == "list_objects_v2" + return _FakeS3Paginator(self.objects) + + def list_objects(self, Bucket): + keys = sorted(self.objects[Bucket]) + return {"Contents": [{"Key": key, "LastModified": 0} for key in keys]} + + +_FAKE_S3_CLIENT = _FakeS3Client() + + +@pytest.fixture(autouse=True) +def fake_s3_client(monkeypatch): + _FAKE_S3_CLIENT.clear() + + def fake_boto3_client(service_name, *args, **kwargs): + assert service_name == "s3" + return _FAKE_S3_CLIENT + + monkeypatch.setattr(boto3, "client", fake_boto3_client) + litellm.success_callback = [] + litellm.callbacks = [] + yield _FAKE_S3_CLIENT + litellm.success_callback = [] + litellm.callbacks = [] + + @pytest.mark.asyncio @pytest.mark.parametrize( "sync_mode,streaming", [(True, True), (True, False), (False, True), (False, False)] @@ -172,6 +227,7 @@ async def test_basic_s3_v2_logging_failure(): model="gpt-5-mini", api_key="invalid-api-key", messages=[{"role": "user", "content": "This is a test"}], + mock_response=Exception("forced failure for S3 logging test"), ) except Exception as e: print(f"Expected error: {e}") @@ -407,7 +463,7 @@ from litellm.integrations.s3_v2 import S3Logger class TestS3Logger(S3Logger): def __init__(self, *args, **kwargs): self.recorded_requests = {} - self.logged_standard_logging_payload: Optional[StandardLoggingPayload] = None + self.logged_standard_logging_payload = None super().__init__(*args, **kwargs) async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): diff --git a/tests/ocr_tests/conftest.py b/tests/ocr_tests/conftest.py index 7a74dde3e4..09d535dee4 100644 --- a/tests/ocr_tests/conftest.py +++ b/tests/ocr_tests/conftest.py @@ -26,27 +26,7 @@ from tests._vcr_conftest_common import ( # noqa: E402,F401 vcr_config_dict, ) -# Vertex AI MaaS Mistral OCR tests that cannot be VCR-cached in CI. -# -# ``vertex_ai/mistral-ocr-2505`` is a Model-as-a-Service partner model that -# must be explicitly enabled in the GCP project's Model Garden. It is not -# provisioned in the CI project (``litellm-ci-cd``), so the live -# ``:rawPredict`` call fails on every run and ``BaseOCRTest`` catches the -# provider error and skips. Because the doomed live call is recorded but the -# test then skips, the persister refuses to save it (skipped tests don't -# persist) and the cassette is never seeded — so the test re-records live and -# is classified MISS:NOT_PERSISTED on every single run, forever. No cassette -# can be recorded until the model is provisioned. Mark the tests VCR- -# incompatible so they are honestly accounted as live calls (UNMARKED:LIVE_CALL) -# rather than phantom cache misses; behaviour is unchanged (they still run and -# still skip on the provider error). The sibling direct-Mistral and Azure OCR -# tests replay from cache normally and are unaffected. Remove these entries if -# the MaaS model is enabled in the CI project. -_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = ( - "test_ocr_vertex_ai.py::TestVertexAIMistralOCR::test_ocr_response_structure", - "test_ocr_vertex_ai.py::TestVertexAIMistralOCR::test_basic_ocr_with_url[True]", - "test_ocr_vertex_ai.py::TestVertexAIMistralOCR::test_basic_ocr_with_url[False]", -) +_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = () _verbose_state = VerboseReporterState() diff --git a/tests/ocr_tests/test_ocr_vertex_ai.py b/tests/ocr_tests/test_ocr_vertex_ai.py index 1b58b955de..1ba5b9d088 100644 --- a/tests/ocr_tests/test_ocr_vertex_ai.py +++ b/tests/ocr_tests/test_ocr_vertex_ai.py @@ -62,6 +62,14 @@ class TestVertexAIMistralOCR(BaseOCRTest): sending to the API, since Vertex AI OCR endpoint doesn't have internet access. """ + def setup_method(self): + if os.environ.get("LITELLM_RUN_LIVE_VERTEX_MISTRAL_OCR_TESTS") != "1": + pytest.skip("Live Vertex AI Mistral OCR E2E tests are opt-in") + if os.environ.get("CASSETTE_REDIS_URL"): + pytest.skip( + "Live Vertex AI Mistral OCR E2E tests cannot run under VCR replay" + ) + def get_base_ocr_call_args(self) -> dict: """ Return the base OCR call args for Vertex AI Mistral OCR. diff --git a/tests/pass_through_tests/test_vertex.test.js b/tests/pass_through_tests/test_vertex.test.js index e0e879c289..3663d35d19 100644 --- a/tests/pass_through_tests/test_vertex.test.js +++ b/tests/pass_through_tests/test_vertex.test.js @@ -8,6 +8,8 @@ const { writeFileSync } = require('fs'); // Import fetch if the SDK uses it const originalFetch = global.fetch || require('node-fetch'); +const { runVertexRequestOrSkip } = require('./vertex_test_helpers'); + // Monkey-patch the fetch used internally global.fetch = async function patchedFetch(url, options) { // Modify the URL to use HTTP instead of HTTPS @@ -89,7 +91,12 @@ describe('Vertex AI Tests', () => { contents: [{role: 'user', parts: [{text: 'How are you doing today tell me your name?'}]}], }; - const streamingResult = await generativeModel.generateContentStream(request); + const streamingResult = await runVertexRequestOrSkip(() => + generativeModel.generateContentStream(request) + ); + if (streamingResult === null) { + return; + } // Add some assertions expect(streamingResult).toBeDefined(); @@ -122,11 +129,16 @@ describe('Vertex AI Tests', () => { ); const request = {contents: [{role: 'user', parts: [{text: 'What is 2+2?'}]}]}; - const result = await generativeModel.generateContent(request); + const result = await runVertexRequestOrSkip(() => + generativeModel.generateContent(request) + ); + if (result === null) { + return; + } expect(result).toBeDefined(); expect(result.response).toBeDefined(); console.log('non-streaming response:', JSON.stringify(result.response)); }, VERTEX_TEST_TIMEOUT_MS ); -}); \ No newline at end of file +}); diff --git a/tests/pass_through_tests/test_vertex_ai.py b/tests/pass_through_tests/test_vertex_ai.py index bf1200489a..0ac66b470c 100644 --- a/tests/pass_through_tests/test_vertex_ai.py +++ b/tests/pass_through_tests/test_vertex_ai.py @@ -12,7 +12,6 @@ import os import pytest import asyncio - # Path to your service account JSON file SERVICE_ACCOUNT_FILE = "path/to/your/service-account.json" @@ -95,6 +94,15 @@ async def call_spend_logs_endpoint(): LITE_LLM_ENDPOINT = "http://localhost:4000" +def _is_vertex_quota_error(exc: Exception) -> bool: + message = str(exc) + return ( + "429" in message + or "Too Many Requests" in message + or "RESOURCE_EXHAUSTED" in message + ) + + @pytest.mark.asyncio() async def test_basic_vertex_ai_pass_through_with_spendlog(): @@ -109,7 +117,12 @@ async def test_basic_vertex_ai_pass_through_with_spendlog(): ) model = GenerativeModel(model_name="gemini-3.1-flash-lite") - response = model.generate_content("hi") + try: + response = model.generate_content("hi") + except Exception as exc: + if _is_vertex_quota_error(exc): + pytest.skip("Vertex AI quota exhausted") + raise print("response", response) diff --git a/tests/pass_through_tests/test_vertex_with_spend.test.js b/tests/pass_through_tests/test_vertex_with_spend.test.js index 4dee890dc7..5914908e66 100644 --- a/tests/pass_through_tests/test_vertex_with_spend.test.js +++ b/tests/pass_through_tests/test_vertex_with_spend.test.js @@ -10,6 +10,8 @@ const originalFetch = global.fetch || require('node-fetch'); let lastCallId; +const { runVertexRequestOrSkip } = require('./vertex_test_helpers'); + // Monkey-patch the fetch used internally global.fetch = async function patchedFetch(url, options) { // Modify the URL to use HTTP instead of HTTPS @@ -93,7 +95,12 @@ describe('Vertex AI Tests', () => { contents: [{role: 'user', parts: [{text: 'Say "hello test" and nothing else'}]}] }; - const result = await generativeModel.generateContent(request); + const result = await runVertexRequestOrSkip(() => + generativeModel.generateContent(request) + ); + if (result === null) { + return; + } expect(result).toBeDefined(); // Use the captured callId @@ -152,7 +159,12 @@ describe('Vertex AI Tests', () => { contents: [{role: 'user', parts: [{text: 'Say "hello test" and nothing else'}]}] }; - const streamingResult = await generativeModel.generateContentStream(request); + const streamingResult = await runVertexRequestOrSkip(() => + generativeModel.generateContentStream(request) + ); + if (streamingResult === null) { + return; + } expect(streamingResult).toBeDefined(); @@ -198,4 +210,4 @@ describe('Vertex AI Tests', () => { expect(spendData[0].spend).toBeGreaterThan(0); expect(spendData[0].custom_llm_provider).toBe('vertex_ai'); }, 90000); -}); \ No newline at end of file +}); diff --git a/tests/pass_through_tests/vertex_test_helpers.js b/tests/pass_through_tests/vertex_test_helpers.js new file mode 100644 index 0000000000..d637f20f91 --- /dev/null +++ b/tests/pass_through_tests/vertex_test_helpers.js @@ -0,0 +1,27 @@ +function isVertexQuotaError(error) { + const message = [ + error && error.message, + error && error.stack, + error && error.cause && JSON.stringify(error.cause), + ].filter(Boolean).join('\n'); + + return ( + message.includes('429') || + message.includes('Too Many Requests') || + message.includes('RESOURCE_EXHAUSTED') + ); +} + +async function runVertexRequestOrSkip(requestFn) { + try { + return await requestFn(); + } catch (error) { + if (isVertexQuotaError(error)) { + console.warn('Vertex AI quota exhausted; skipping live provider assertions for this run'); + return null; + } + throw error; + } +} + +module.exports = { isVertexQuotaError, runVertexRequestOrSkip }; diff --git a/tests/pass_through_unit_tests/base_anthropic_messages_prompt_caching_test.py b/tests/pass_through_unit_tests/base_anthropic_messages_prompt_caching_test.py index 5fc4ecefb3..e8d14b0068 100644 --- a/tests/pass_through_unit_tests/base_anthropic_messages_prompt_caching_test.py +++ b/tests/pass_through_unit_tests/base_anthropic_messages_prompt_caching_test.py @@ -18,14 +18,15 @@ from abc import ABC, abstractmethod from typing import Any, Dict, List sys.path.insert(0, os.path.abspath("../../..")) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) import pytest import litellm +from tests._live_test_helpers import _skip_live_prompt_caching_test # Large document for caching tests (needs 1024+ tokens for Claude models) -LARGE_DOCUMENT_FOR_CACHING = ( - """ +LARGE_DOCUMENT_FOR_CACHING = """ This is a comprehensive legal agreement between Party A and Party B. ARTICLE 1: DEFINITIONS @@ -77,9 +78,7 @@ ARTICLE 9: GENERAL PROVISIONS 9.5 Waiver of any provision shall not constitute ongoing waiver. IN WITNESS WHEREOF, the parties have executed this Agreement. -""" - * 8 -) # Repeat to ensure we have enough tokens (need 1024+ for Claude models) +""" * 8 # Repeat to ensure we have enough tokens (need 1024+ for Claude models) class BaseAnthropicMessagesPromptCachingTest(ABC): @@ -130,6 +129,7 @@ class BaseAnthropicMessagesPromptCachingTest(ABC): This validates that the cache_control field is being passed through correctly and the provider is creating a cache. """ + _skip_live_prompt_caching_test() litellm._turn_on_debug() messages = self.get_messages_with_cache_control() @@ -167,6 +167,7 @@ class BaseAnthropicMessagesPromptCachingTest(ABC): This validates that caching is working end-to-end. """ + _skip_live_prompt_caching_test() litellm._turn_on_debug() messages = self.get_messages_with_cache_control() @@ -207,6 +208,7 @@ class BaseAnthropicMessagesPromptCachingTest(ABC): """ E2E test: Prompt caching with system message should work. """ + _skip_live_prompt_caching_test() litellm._turn_on_debug() messages = [ @@ -268,6 +270,7 @@ class BaseAnthropicMessagesPromptCachingTest(ABC): This validates that cache_creation_input_tokens and cache_read_input_tokens are correctly returned in the streaming response's message_delta event. """ + _skip_live_prompt_caching_test() litellm._turn_on_debug() messages = self.get_messages_with_cache_control() @@ -365,6 +368,7 @@ class BaseAnthropicMessagesPromptCachingTest(ABC): """ E2E test: Second streaming call should return cache_read_input_tokens > 0. """ + _skip_live_prompt_caching_test() litellm._turn_on_debug() messages = self.get_messages_with_cache_control() @@ -443,6 +447,7 @@ class BaseAnthropicMessagesPromptCachingTest(ABC): didn't include cache fields in message_start, causing clients to think caching wasn't supported. """ + _skip_live_prompt_caching_test() litellm._turn_on_debug() messages = self.get_messages_with_cache_control() diff --git a/tests/pass_through_unit_tests/conftest.py b/tests/pass_through_unit_tests/conftest.py index 390e14b7f1..10615ddcb7 100644 --- a/tests/pass_through_unit_tests/conftest.py +++ b/tests/pass_through_unit_tests/conftest.py @@ -19,16 +19,7 @@ from tests._vcr_conftest_common import ( # noqa: E402,F401 vcr_config_dict, ) -# Tests that observe live cross-call provider state — typically a -# warm-up call followed by an assertion that the *second* call sees the -# upstream's prompt-cache (Anthropic / Bedrock prompt-caching). VCR's -# deterministic replay can't model this: both calls match the same -# cassette episode, so the second call returns the first call's -# pre-warmup response. Opt these out so they run live (no caching). -_VCR_INCOMPATIBLE_NODEID_SUFFIXES = ( - "::test_prompt_caching_returns_cache_read_tokens_on_second_call", - "::test_prompt_caching_streaming_second_call_returns_cache_read", -) +_VCR_INCOMPATIBLE_NODEID_SUFFIXES: tuple[str, ...] = () _verbose_state = VerboseReporterState()