mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-13 03:59:12 +00:00
04d97abffb
* test: make 77 legacy live tests offline in litellm_utils, router_unit and responses dirs * test: restore lost coverage in litellm_utils offline replacements Bound live health-check tasks to the concurrency limit so eager task creation behind a semaphore fails, route the Langfuse trace-id check through completion -> Logging.get_trace_id with all four metadata combinations, run the callback dedup checks through acompletion, pass the dynamic key via litellm_params, cover the env-inferred default model list, and add the per-provider audio transcription config lookup * test: restore router coverage lost in the offline move Add the missing non-stream LIT-3058 header/count-once test, pin the UTC minute on every usage-counter test, assert router-level client reuse for transcription, and replace the private selector-attribute checks with routing outcomes per strategy. Assert outbound bodies for speech, rerank, image, assistants and moderation, add timeouts to event waits, and drop the router_unit_tests husk files that no longer hold tests * test: restore sync stream, router sync stream, error event and field-type coverage for migrated responses tests Add the sync streaming logging and sync Router.responses streaming cases the offline replacements dropped, port the legacy per-event and response field-type validation, assert raw headers, the float created_at conversion, MCP tool headers, the search_context_size input, and add an offline replacement for the in-stream context-window error event. Remove helpers left dead in the legacy file. * test: run numpydoc-backed unit tests in GHA and drop a test that pinned a crash * test: restore sequence number, item id and content part checks in the responses stream validator * test: match router logging events to the test's own deployment so late events from other tests are ignored * test: restore the s3 cold storage history test to its original assertions --------- Co-authored-by: yuneng <yuneng@berri.ai>
83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
import base64
|
|
from dataclasses import dataclass
|
|
from typing import Final
|
|
|
|
import litellm
|
|
import pytest
|
|
import respx
|
|
|
|
from litellm.secret_managers.google_secret_manager import GoogleSecretManager
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _CachedVertexCredentials:
|
|
token: str
|
|
quota_project_id: str | None
|
|
expired: bool = False
|
|
|
|
def refresh(self, request: object) -> None:
|
|
return None
|
|
|
|
|
|
def _google_secret_manager(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
project_id: str,
|
|
) -> GoogleSecretManager:
|
|
monkeypatch.setenv("GOOGLE_SECRET_MANAGER_PROJECT_ID", project_id)
|
|
monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True)
|
|
secret_manager: Final = GoogleSecretManager()
|
|
credentials: Final = _CachedVertexCredentials(
|
|
token="test-gsm-token",
|
|
quota_project_id=project_id,
|
|
)
|
|
vertex_chat_completion: Final = litellm.vertex_chat_completion
|
|
monkeypatch.setitem(
|
|
vertex_chat_completion._credentials_project_mapping,
|
|
(None, project_id),
|
|
(credentials, project_id),
|
|
)
|
|
return secret_manager
|
|
|
|
|
|
def test_google_secret_manager_decodes_secret_and_requests_latest_version(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
project_id: Final = "test-secret-project"
|
|
secret_manager: Final = _google_secret_manager(monkeypatch, project_id)
|
|
secret_url: Final = (
|
|
f"https://secretmanager.googleapis.com/v1/projects/{project_id}/secrets/OPENAI_API_KEY/versions/latest:access"
|
|
)
|
|
encoded_secret: Final = base64.b64encode(b"anything").decode("ascii")
|
|
upstream: Final[respx.MockRouter]
|
|
|
|
with respx.mock(assert_all_called=True) as upstream:
|
|
secret_route: Final = upstream.get(secret_url).respond(
|
|
200,
|
|
json={"payload": {"data": encoded_secret}},
|
|
)
|
|
|
|
result: Final = secret_manager.get_secret_from_google_secret_manager("OPENAI_API_KEY")
|
|
|
|
assert result == "anything"
|
|
assert secret_route.called
|
|
assert len(upstream.calls) == 1
|
|
assert str(upstream.calls.last.request.url) == secret_url
|
|
assert upstream.calls.last.request.headers["Authorization"] == "Bearer test-gsm-token"
|
|
|
|
|
|
def test_google_secret_manager_returns_cached_values_without_http(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
project_id: Final = "test-secret-project"
|
|
secret_manager: Final = _google_secret_manager(monkeypatch, project_id)
|
|
secret_manager.cache.set_cache("cached-none", None)
|
|
secret_manager.cache.set_cache("cached-string", "lite-llm")
|
|
upstream: Final[respx.MockRouter]
|
|
|
|
with respx.mock() as upstream:
|
|
missing_value: Final = secret_manager.get_secret_from_google_secret_manager("cached-none")
|
|
cached_value: Final = secret_manager.get_secret_from_google_secret_manager("cached-string")
|
|
|
|
assert missing_value is None
|
|
assert cached_value == "lite-llm"
|
|
assert upstream.calls.call_count == 0
|