From 514bc181d65898b23692b8663084f98e1d123970 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 23 Sep 2026 09:51:56 -0700 Subject: [PATCH] test(integration): regression tests for July cost tracking, budgeting and spend bugs (#42694) * test(integration): streamed Bedrock Messages usage cost equals the recorded spend (Pylon #6667) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): echoed cost-map model info is not persisted as deployment overrides (Pylon #6844) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): reset sweep runs on one pod per tick while replicas share the lease (Pylon #6521) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): bedrock post-call guardrail scans streamed Anthropic Messages tool use without 500 (Pylon #6503) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): realtime cached audio tokens bill at the audio cache-read rate (Pylon #6704) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): legacy GET /spend/logs returns at most the 10000 most recent rows and flags truncation (Pylon #6752) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): itemize Responses API cache write tokens as cache creation cost (Pylon #6454) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): migration entrypoint deploys pending migrations before proxy startup (Pylon #6649) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): opted-in team keys stop at the owner's personal budget (Pylon #6641) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): guardrail information stays in the spend log when the caller sends metadata on /v1/messages (Pylon #6614) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): key model allowlist is enforced on Bedrock passthrough routes (Pylon #6419) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): JWT mapped key backfills a null user email from token claims (Pylon #6266) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): scheduled budget reset recovers from a transient DB transport failure (Pylon #6582) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): team key lists models granted through a team access group (Pylon #6044) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): JWT subject without team claim lands in the configured default team (Pylon #5895) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): end-user spend lands for a key without user_id when the auth cache is Redis (Pylon #6021) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): stale-low redis counter still blocks team member over budget (Pylon #5824) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): prompt-carrying spend rows are written in byte-bounded statements (Pylon #6083) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): logs UI session_total_spend sums every round of a multi-round session (Pylon #5928) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): config.yaml guardrails are served by the guardrail usage detail and overview (Pylon #5813) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): plain chat request skips the object permission lookup (Pylon #5965) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): register july accounting regression contracts Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): isolate cost map override clear on owned proxy Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): make reset lease claim and db relay refusal deterministic Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): poll pg_stat settle, bound unbanned relay refusals, clear reset lease on teardown Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): bound relay refusals so the budget sweep can reconnect Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: kerry Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- tests/integration/_support/database.py | 6 +- tests/integration/_support/database_relay.py | 95 +++++++++ .../test_access_group_model_listing.py | 51 +++++ .../test_bedrock_passthrough_model_access.py | 65 ++++++ .../test_jwt_default_team_provisioning.py | 86 ++++++++ .../test_jwt_mapped_key_email_backfill.py | 112 ++++++++++ .../test_object_permission_lookup.py | 74 +++++++ .../database/test_migration_entrypoint.py | 99 +++++++++ .../test_guardrail_usage_config_guardrail.py | 77 +++++++ .../observability/test_guardrail_effects.py | 192 +++++++++++++++++- .../pricing/test_configured_prices.py | 45 ++++ .../test_realtime_cached_audio_pricing.py | 131 ++++++++++++ .../integration/spend/test_cache_and_quota.py | 106 +++++++++- .../test_end_user_spend_without_proxy_user.py | 29 +++ .../spend/test_legacy_spend_logs_row_cap.py | 46 +++++ .../spend/test_messages_stream_usage_cost.py | 176 ++++++++++++++++ .../test_reset_budget_leader_election.py | 75 +++++++ .../test_responses_cache_write_itemization.py | 98 +++++++++ .../spend/test_session_total_spend.py | 98 +++++++++ .../spend/test_spend_log_write_batching.py | 75 +++++++ .../spend/test_team_member_spend.py | 45 ++++ .../spend/test_user_budget_on_team_keys.py | 54 +++++ 22 files changed, 1830 insertions(+), 5 deletions(-) create mode 100644 tests/integration/_support/database_relay.py create mode 100644 tests/integration/authorization/test_access_group_model_listing.py create mode 100644 tests/integration/authorization/test_bedrock_passthrough_model_access.py create mode 100644 tests/integration/authorization/test_jwt_default_team_provisioning.py create mode 100644 tests/integration/authorization/test_jwt_mapped_key_email_backfill.py create mode 100644 tests/integration/authorization/test_object_permission_lookup.py create mode 100644 tests/integration/database/test_migration_entrypoint.py create mode 100644 tests/integration/management/test_guardrail_usage_config_guardrail.py create mode 100644 tests/integration/pricing/test_realtime_cached_audio_pricing.py create mode 100644 tests/integration/spend/test_end_user_spend_without_proxy_user.py create mode 100644 tests/integration/spend/test_legacy_spend_logs_row_cap.py create mode 100644 tests/integration/spend/test_messages_stream_usage_cost.py create mode 100644 tests/integration/spend/test_reset_budget_leader_election.py create mode 100644 tests/integration/spend/test_responses_cache_write_itemization.py create mode 100644 tests/integration/spend/test_session_total_spend.py create mode 100644 tests/integration/spend/test_spend_log_write_batching.py create mode 100644 tests/integration/spend/test_user_budget_on_team_keys.py diff --git a/tests/integration/_support/database.py b/tests/integration/_support/database.py index 283d26a632e..e7f0ebdf603 100644 --- a/tests/integration/_support/database.py +++ b/tests/integration/_support/database.py @@ -8,7 +8,9 @@ from pydantic import JsonValue, TypeAdapter ROWS: Final = TypeAdapter(list[dict[str, JsonValue]]) -def read_rows(query: str, parameters: tuple[str, ...]) -> list[dict[str, JsonValue]]: - with psycopg.connect(os.environ["DATABASE_URL"], row_factory=dict_row) as connection: +def read_rows( + query: str, parameters: tuple[str, ...], *, database_url: str | None = None +) -> list[dict[str, JsonValue]]: + with psycopg.connect(database_url or os.environ["DATABASE_URL"], row_factory=dict_row) as connection: connection.execute("SET TRANSACTION READ ONLY") return ROWS.validate_python(connection.execute(query, parameters).fetchall()) diff --git a/tests/integration/_support/database_relay.py b/tests/integration/_support/database_relay.py new file mode 100644 index 00000000000..46f3e17af13 --- /dev/null +++ b/tests/integration/_support/database_relay.py @@ -0,0 +1,95 @@ +import asyncio +import socket +import threading +from collections.abc import Generator +from contextlib import contextmanager +from typing import Final +from urllib.parse import urlsplit, urlunsplit + +from pydantic import TypeAdapter + +PORT: Final = TypeAdapter(int) + + +def _free_port() -> int: + with socket.socket() as reserve: + reserve.bind(("127.0.0.1", 0)) + return PORT.validate_python(reserve.getsockname()[1]) + + +class DatabaseRelay: + def __init__(self, upstream_host: str, upstream_port: int, trigger: bytes) -> None: + self.port: Final = _free_port() + self._upstream_host: Final = upstream_host + self._upstream_port: Final = upstream_port + self._trigger: Final = trigger + self._loop: Final = asyncio.new_event_loop() + self._armed: Final = threading.Event() + self.tripped: Final = threading.Event() + self.refused = 0 + self._writers: tuple[asyncio.StreamWriter, ...] = () + self._ready: Final = threading.Event() + self._thread: Final = threading.Thread(target=self._run, daemon=True) + + def arm(self) -> None: + self._armed.set() + + def start(self) -> None: + self._thread.start() + assert self._ready.wait(10), "Database relay did not start" + + def stop(self) -> None: + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join(10) + + def _run(self) -> None: + asyncio.set_event_loop(self._loop) + self._loop.run_until_complete(asyncio.start_server(self._serve, "127.0.0.1", self.port)) + self._ready.set() + self._loop.run_forever() + + def _drop_all(self) -> None: + for writer in self._writers: + writer.close() + self._writers = () + + async def _serve(self, client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter) -> None: + if self.tripped.is_set() and self.refused < 5: + self.refused += 1 + client_writer.close() + return + server_reader, server_writer = await asyncio.open_connection(self._upstream_host, self._upstream_port) + self._writers = (*self._writers, client_writer, server_writer) + + async def forward(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, inspect: bool) -> None: + try: + while chunk := await reader.read(65536): + if inspect and self._armed.is_set() and not self.tripped.is_set() and self._trigger in chunk: + self.tripped.set() + self._drop_all() + return + writer.write(chunk) + await writer.drain() + except (ConnectionError, asyncio.IncompleteReadError): + return + finally: + writer.close() + + await asyncio.gather( + forward(client_reader, server_writer, True), + forward(server_reader, client_writer, False), + ) + + +@contextmanager +def database_relay(database_url: str, trigger: bytes) -> Generator[tuple[DatabaseRelay, str]]: + parts: Final = urlsplit(database_url) + assert parts.hostname is not None and parts.port is not None, database_url + relay: Final = DatabaseRelay(parts.hostname, parts.port, trigger) + relay.start() + credentials: Final = f"{parts.username}:{parts.password}@" if parts.username else "" + relayed: Final = urlunsplit(parts._replace(netloc=f"{credentials}127.0.0.1:{relay.port}")) + try: + yield relay, relayed + finally: + relay.stop() diff --git a/tests/integration/authorization/test_access_group_model_listing.py b/tests/integration/authorization/test_access_group_model_listing.py new file mode 100644 index 00000000000..9b37cc8f232 --- /dev/null +++ b/tests/integration/authorization/test_access_group_model_listing.py @@ -0,0 +1,51 @@ +import uuid +from collections.abc import Iterator +from contextlib import contextmanager +from typing import Final + +import httpx +import pytest + +from tests.integration._support.client import Gateway, eventually, object_value, string_value + + +@contextmanager +def _team_access_group(gateway: Gateway, team_id: str, model: str) -> Iterator[str]: + created: Final = gateway.request( + "POST", + "/v1/access_group", + { + "access_group_name": f"integration-{uuid.uuid4().hex}", + "access_model_names": [model], + "assigned_team_ids": [team_id], + }, + ) + assert created.status_code == 201, created.text + identity: Final = string_value(created.json()["access_group_id"]) + try: + yield identity + finally: + deleted: Final = gateway.request("DELETE", f"/v1/access_group/{identity}") + assert deleted.status_code == 204, deleted.text + + +def _listed_model_ids(response: httpx.Response) -> tuple[str, ...]: + entries: Final = response.json()["data"] + assert isinstance(entries, list), response.text + return tuple(string_value(object_value(entry)["id"]) for entry in entries) + + +@pytest.mark.covers("authorization.access_groups.team_key_lists_models_granted_through_team_access_group") +def test_team_key_lists_models_granted_through_team_access_group(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + model: Final = scenario.model() + team_id: Final = scenario.team(models=["no-default-models"]) + with _team_access_group(gateway, team_id, model): + key: Final = scenario.key(team_id=team_id) + response: Final = eventually( + lambda: gateway.request("GET", "/v1/models", key=key), + lambda value: value.status_code == 200 and _listed_model_ids(value) == (model,), + return_last_on_timeout=True, + ) + assert response.status_code == 200, response.text + assert _listed_model_ids(response) == (model,), response.text diff --git a/tests/integration/authorization/test_bedrock_passthrough_model_access.py b/tests/integration/authorization/test_bedrock_passthrough_model_access.py new file mode 100644 index 00000000000..c1b54fe7e38 --- /dev/null +++ b/tests/integration/authorization/test_bedrock_passthrough_model_access.py @@ -0,0 +1,65 @@ +import json +from typing import Final + +import pytest +from integration._support.client import Gateway +from integration._support.upstream import _aws_event_frame +from integration._support.wire import Reply, Request, wire_server + +_MODEL_ID: Final = "anthropic.claude-sonnet-5-v1:0" +_ACTIONS: Final = ("converse", "invoke", "converse-stream", "invoke-with-response-stream") +_REQUEST_BODY: Final = {"messages": [{"role": "user", "content": [{"text": "synthetic passthrough allowlist"}]}]} +_CONVERSE_RESPONSE: Final = json.dumps( + { + "output": {"message": {"role": "assistant", "content": [{"text": "bedrock allowlist control"}]}}, + "stopReason": "end_turn", + "usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15}, + "metrics": {"latencyMs": 1}, + } +).encode() +_STREAM_BYTES: Final = b"".join( + _aws_event_frame(kind, payload, "sc", "u") + for kind, payload in ( + ("messageStart", {"role": "assistant"}), + ("contentBlockDelta", {"delta": {"text": "bedrock allowlist control"}, "contentBlockIndex": 0}), + ("messageStop", {"stopReason": "end_turn"}), + ("metadata", {"usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15}}), + ) +) + + +def bedrock_peer(request: Request) -> Reply: + assert request.method == "POST", request.target + assert json.loads(request.body)["messages"] == _REQUEST_BODY["messages"], request.body + if request.target.endswith("-stream"): + return Reply(body=_STREAM_BYTES, content_type="application/vnd.amazon.eventstream") + return Reply(body=_CONVERSE_RESPONSE) + + +@pytest.mark.covers("authz.key_models.bedrock_passthrough_route_model_is_enforced") +def test_key_scoped_to_one_model_cannot_call_another_through_bedrock_passthrough_routes(gateway: Gateway) -> None: + with wire_server(bedrock_peer) as wire, gateway.scenario() as scenario: + allowed: Final = scenario.model( + model=f"bedrock/{_MODEL_ID}", + api_base=wire.url, + aws_access_key_id="AKIASCRIPTEDPROVIDER", + aws_secret_access_key="scripted-secret", + aws_region_name="us-east-1", + ) + denied: Final = scenario.model( + model=f"bedrock/{_MODEL_ID}", + api_base=wire.url, + aws_access_key_id="AKIASCRIPTEDPROVIDER", + aws_secret_access_key="scripted-secret", + aws_region_name="us-east-1", + ) + key: Final = scenario.key(models=[allowed]) + for action in _ACTIONS: + response: Final = gateway.request("POST", f"/bedrock/model/{denied}/{action}", _REQUEST_BODY, key=key) + assert response.status_code == 403, f"{action}: {response.status_code} {response.text}" + assert response.json()["error"]["type"] == "key_model_access_denied", f"{action}: {response.text}" + assert wire.drain() == (), f"{action} reached the provider: {response.text}" + for action in _ACTIONS: + served: Final = gateway.request("POST", f"/bedrock/model/{allowed}/{action}", _REQUEST_BODY, key=key) + assert served.status_code == 200, f"{action}: {served.status_code} {served.text}" + assert tuple(request.target for request in wire.drain()) == (f"/model/{_MODEL_ID}/{action}",), served.text diff --git a/tests/integration/authorization/test_jwt_default_team_provisioning.py b/tests/integration/authorization/test_jwt_default_team_provisioning.py new file mode 100644 index 00000000000..04715d128b0 --- /dev/null +++ b/tests/integration/authorization/test_jwt_default_team_provisioning.py @@ -0,0 +1,86 @@ +import json +import time +import uuid +from pathlib import Path +from typing import Final + +import jwt +import pytest +import yaml +from cryptography.hazmat.primitives.asymmetric import rsa + +from tests.integration._support.client import Gateway, eventually +from tests.integration._support.database import read_rows +from tests.integration._support.process import owned_proxy +from tests.integration._support.wire import Reply, Request, wire_server + +KEY_ID: Final = "integration-jwt-signing-key" +TEAM_BUDGET: Final = 25.0 + + +def _jwks_reply(public_jwk: str) -> Reply: + return Reply(body=json.dumps({"keys": [{**json.loads(public_jwk), "kid": KEY_ID}]}).encode()) + + +@pytest.mark.covers("authorization.jwt.new_subject_without_team_claim_joins_default_team") +def test_jwt_subject_without_team_claim_is_provisioned_into_configured_default_team( + gateway: Gateway, tmp_path: Path +) -> None: + private_key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048) + public_jwk: Final = jwt.algorithms.RSAAlgorithm.to_jwk(private_key.public_key()) + + def respond(request: Request) -> Reply: + assert request.method == "GET", request + return _jwks_reply(public_jwk) + + with wire_server(respond) as jwks, gateway.scenario() as scenario: + team: Final = scenario.team() + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["general_settings"] = { + **config["general_settings"], + "enable_jwt_auth": True, + "litellm_jwtauth": {"user_id_jwt_field": "sub", "user_id_upsert": True}, + } + config["litellm_settings"] = { + **config["litellm_settings"], + "default_internal_user_params": { + "user_role": "internal_user", + "teams": [{"team_id": team, "user_role": "user", "max_budget_in_team": TEAM_BUDGET}], + }, + } + path: Final = tmp_path / "jwt_default_team.yaml" + path.write_text(yaml.safe_dump(config)) + subject: Final = f"integration-jwt-{uuid.uuid4().hex}" + token: Final = jwt.encode( + {"sub": subject, "iat": int(time.time()), "exp": int(time.time()) + 300}, + private_key, + algorithm="RS256", + headers={"kid": KEY_ID}, + ) + with owned_proxy(gateway, tmp_path, {"JWT_PUBLIC_KEY_URL": jwks.url}, config=path) as candidate: + model: Final = scenario.model() + scenario.cleanups.callback(scenario.delete_user, subject) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": "default team control"}]}, + key=token, + ) + assert response.status_code == 200, response.text + assert response.json()["choices"][0]["message"]["content"] == ( + "Hello! This is a mock response from the fake OpenAI endpoint." + ), response.text + assert read_rows( + 'SELECT user_id, user_role, teams FROM "LiteLLM_UserTable" WHERE user_id = %s', (subject,) + ) == [{"user_id": subject, "user_role": "internal_user", "teams": [team]}] + memberships: Final = eventually( + lambda: read_rows( + 'SELECT m.team_id, b.max_budget FROM "LiteLLM_TeamMembership" m ' + 'JOIN "LiteLLM_BudgetTable" b ON b.budget_id = m.budget_id WHERE m.user_id = %s', + (subject,), + ), + lambda rows: len(rows) == 1, + ) + assert memberships == [{"team_id": team, "max_budget": TEAM_BUDGET}] + roster: Final = read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,)) + assert {"user_id": subject, "role": "user", "user_email": None} in roster[0]["members_with_roles"], roster diff --git a/tests/integration/authorization/test_jwt_mapped_key_email_backfill.py b/tests/integration/authorization/test_jwt_mapped_key_email_backfill.py new file mode 100644 index 00000000000..47c8ad5f115 --- /dev/null +++ b/tests/integration/authorization/test_jwt_mapped_key_email_backfill.py @@ -0,0 +1,112 @@ +import json +import time +import uuid +from hashlib import sha256 +from pathlib import Path +from typing import Final + +import jwt +import pytest +from cryptography.hazmat.primitives.asymmetric import rsa +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.process import owned_proxy +from integration._support.wire import Reply, Request, wire_server +from jwt.algorithms import RSAAlgorithm + +AUDIENCE: Final = "litellm-integration" +KEY_ID: Final = "integration-signing-key" +CLIENT_CLAIM: Final = "client_id" + + +def _proxy_config(directory: Path, model: str, upstream_url: str) -> Path: + config: Final = directory / "jwt_mapped_key_config.yaml" + config.write_text( + json.dumps( + { + "model_list": [ + { + "model_name": model, + "litellm_params": { + "model": "openai/" + model, + "api_base": upstream_url + "/v1", + "api_key": "sk-upstream", + }, + } + ], + "general_settings": { + "master_key": "os.environ/LITELLM_MASTER_KEY", + "database_url": "os.environ/DATABASE_URL", + "store_model_in_db": True, + "proxy_batch_write_at": 1, + "proxy_batch_polling_interval": 1, + "enable_jwt_auth": True, + "litellm_jwtauth": { + "user_id_jwt_field": "sub", + "user_email_jwt_field": "email", + "virtual_key_claim_field": CLIENT_CLAIM, + }, + }, + "router_settings": {"disable_cooldowns": True}, + } + ) + ) + return config + + +def _signed_token(private_key: rsa.RSAPrivateKey, user_id: str, email: str, client_id: str) -> str: + now: Final = int(time.time()) + return jwt.encode( + {"sub": user_id, "email": email, CLIENT_CLAIM: client_id, "aud": AUDIENCE, "iat": now, "exp": now + 300}, + private_key, + algorithm="RS256", + headers={"kid": KEY_ID}, + ) + + +@pytest.mark.covers("authorization.jwt.mapped_key_backfills_null_user_email_from_claims") +def test_jwt_mapped_key_request_backfills_null_user_email_from_token_claims(gateway: Gateway, tmp_path: Path) -> None: + private_key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048) + public_jwk: Final = json.loads(RSAAlgorithm.to_jwk(private_key.public_key())) + jwks: Final = json.dumps({"keys": [{**public_jwk, "kid": KEY_ID, "use": "sig", "alg": "RS256"}]}).encode() + + def respond(request: Request) -> Reply: + assert request.target == "/jwks", request + return Reply(body=jwks) + + model: Final = "integration-jwt-" + uuid.uuid4().hex + with wire_server(respond) as issuer: + config: Final = _proxy_config(tmp_path, model, gateway.upstream_url) + overrides: Final = {"JWT_PUBLIC_KEY_URL": issuer.url + "/jwks", "JWT_AUDIENCE": AUDIENCE} + with owned_proxy(gateway, tmp_path, overrides, config=config) as candidate, candidate.scenario() as scenario: + user: Final = scenario.user() + key: Final = scenario.key(user_id=user, models=[model]) + client_id: Final = "integration-client-" + uuid.uuid4().hex + mapping: Final = candidate.post( + "/jwt/key/mapping/new", {"jwt_claim_name": CLIENT_CLAIM, "jwt_claim_value": client_id, "key": key} + ) + scenario.cleanups.callback(candidate.post, "/jwt/key/mapping/delete", {"id": mapping["id"]}) + assert read_rows('SELECT user_email FROM "LiteLLM_UserTable" WHERE user_id = %s', (user,)) == [ + {"user_email": None} + ] + email: Final = f"{user}@integration.example" + token: Final = _signed_token(private_key, user, email, client_id) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": "jwt email backfill control"}]}, + key=token, + ) + assert response.status_code == 200, response.text + assert read_rows('SELECT user_email FROM "LiteLLM_UserTable" WHERE user_id = %s', (user,)) == [ + {"user_email": email} + ] + spend_rows: Final = eventually( + lambda: read_rows( + 'SELECT api_key, "user" FROM "LiteLLM_SpendLogs" WHERE request_id = %s', + (str(response.json()["id"]),), + ), + lambda rows: len(rows) == 1, + seconds=70, + ) + assert spend_rows == [{"api_key": sha256(key.encode()).hexdigest(), "user": user}] diff --git a/tests/integration/authorization/test_object_permission_lookup.py b/tests/integration/authorization/test_object_permission_lookup.py new file mode 100644 index 00000000000..21fba30c165 --- /dev/null +++ b/tests/integration/authorization/test_object_permission_lookup.py @@ -0,0 +1,74 @@ +import time +from typing import Final + +import pytest + +from tests.integration._support.client import Gateway, eventually +from tests.integration._support.database import read_rows + +PLAIN_REQUESTS: Final = 10 +STATS_FLUSH_WINDOW_SECONDS: Final = 11.0 + + +def _object_permission_reads() -> int: + rows: Final = read_rows( + "SELECT seq_scan + idx_scan AS reads FROM pg_stat_user_tables WHERE relname = %s", + ("LiteLLM_ObjectPermissionTable",), + ) + reads: Final = rows[0]["reads"] + assert isinstance(reads, int), rows + return reads + + +def _settled_object_permission_reads(previous: int, unchanged_since: float) -> int: + changed: Final = eventually( + lambda: _object_permission_reads() != previous, + lambda drifted: drifted, + seconds=STATS_FLUSH_WINDOW_SECONDS - (time.monotonic() - unchanged_since), + return_last_on_timeout=True, + ) + if not changed: + return previous + return _settled_object_permission_reads(_object_permission_reads(), time.monotonic()) + + +def _assert_plain_chat_served(gateway: Gateway, model: str, key: str) -> None: + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": "no vector stores"}]}, + key=key, + ) + assert response.status_code == 200, response.text + assert response.json()["choices"][0]["message"]["content"] == ( + "Hello! This is a mock response from the fake OpenAI endpoint." + ) + + +def _assert_forbidden_vector_store_denied(gateway: Gateway, model: str, key: str) -> None: + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": "forbidden"}], "vector_store_ids": ["vs_forbidden"]}, + key=key, + ) + assert response.status_code == 401, response.text + assert response.json()["error"]["type"] == "key_vector_store_access_denied", response.text + + +@pytest.mark.covers("authorization.vector_store.plain_request_skips_object_permission_lookup") +def test_chat_request_without_vector_stores_does_not_read_object_permission_table(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + model: Final = scenario.model() + key: Final = scenario.key(models=[model], object_permission={"vector_stores": ["vs_allowed"]}) + _assert_plain_chat_served(gateway, model, key) + before_control: Final = _object_permission_reads() + _assert_forbidden_vector_store_denied(gateway, model, key) + eventually(_object_permission_reads, lambda reads: reads > before_control, seconds=15) + baseline: Final = _settled_object_permission_reads(_object_permission_reads(), time.monotonic()) + for _ in range(PLAIN_REQUESTS): + _assert_plain_chat_served(gateway, model, key) + after: Final = _settled_object_permission_reads(_object_permission_reads(), time.monotonic()) + assert after - baseline < PLAIN_REQUESTS, ( + f"{PLAIN_REQUESTS} plain chat requests added {after - baseline} object permission reads" + ) diff --git a/tests/integration/database/test_migration_entrypoint.py b/tests/integration/database/test_migration_entrypoint.py new file mode 100644 index 00000000000..19c480b9531 --- /dev/null +++ b/tests/integration/database/test_migration_entrypoint.py @@ -0,0 +1,99 @@ +import os +import shutil +import subprocess +import sys +import uuid +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path +from typing import Final +from urllib.parse import urlsplit, urlunsplit + +import psycopg +import pytest +from integration._support.client import Gateway +from integration._support.process import owned_proxy +from psycopg import sql +from psycopg.rows import dict_row + +REPO_ROOT: Final = Path(__file__).resolve().parents[3] +PRISMA_DIR: Final = REPO_ROOT / "litellm-proxy-extras" / "litellm_proxy_extras" +MISSING_MIGRATION: Final = "20260626120000_add_mcp_tool_search_enabled" +SHIPPED_MIGRATIONS: Final = tuple(sorted(path.name for path in (PRISMA_DIR / "migrations").iterdir() if path.is_dir())) + + +@contextmanager +def fresh_database() -> Iterator[str]: + name: Final = f"integration_upgrade_{uuid.uuid4().hex}" + admin_url: Final = os.environ["DATABASE_URL"] + parsed: Final = urlsplit(admin_url) + with psycopg.connect(admin_url, autocommit=True) as admin: + admin.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(name))) + try: + yield urlunsplit(parsed._replace(path=f"/{name}")) + finally: + admin.execute(sql.SQL("DROP DATABASE {} WITH (FORCE)").format(sql.Identifier(name))) + + +def deploy_older_schema(database_url: str, directory: Path) -> None: + older: Final = directory / "older-release" + (older / "migrations").mkdir(parents=True) + shutil.copy(PRISMA_DIR / "schema.prisma", older / "schema.prisma") + shutil.copy(PRISMA_DIR / "migrations" / "migration_lock.toml", older / "migrations" / "migration_lock.toml") + for name in (name for name in SHIPPED_MIGRATIONS if name < MISSING_MIGRATION): + shutil.copytree(PRISMA_DIR / "migrations" / name, older / "migrations" / name) + subprocess.run( + [sys.executable, "-I", "-m", "prisma", "migrate", "deploy", "--schema", str(older / "schema.prisma")], + check=True, + capture_output=True, + text=True, + timeout=300, + env={**os.environ, "DATABASE_URL": database_url}, + ) + + +def applied_migrations(database_url: str) -> tuple[str, ...]: + with psycopg.connect(database_url, row_factory=dict_row) as connection: + rows: Final = connection.execute( + 'SELECT migration_name FROM "_prisma_migrations" ' + "WHERE finished_at IS NOT NULL AND rolled_back_at IS NULL ORDER BY migration_name" + ).fetchall() + return tuple(str(row["migration_name"]) for row in rows) + + +def object_permission_columns(database_url: str) -> tuple[str, ...]: + with psycopg.connect(database_url, row_factory=dict_row) as connection: + rows: Final = connection.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_name = 'LiteLLM_ObjectPermissionTable' AND column_name = 'mcp_tool_search_enabled'" + ).fetchall() + return tuple(str(row["column_name"]) for row in rows) + + +@pytest.mark.covers("other.database.migrations.entrypoint_deploys_pending_migrations_before_startup") +def test_migration_entrypoint_upgrades_an_older_schema_so_the_proxy_serves_mcp_tools( + gateway: Gateway, tmp_path: Path +) -> None: + with fresh_database() as database_url: + deploy_older_schema(database_url, tmp_path) + assert object_permission_columns(database_url) == () + assert applied_migrations(database_url) == tuple( + name for name in SHIPPED_MIGRATIONS if name < MISSING_MIGRATION + ) + entrypoint: Final = subprocess.run( + [sys.executable, "-I", "-m", "litellm.proxy.prisma_migration"], + capture_output=True, + text=True, + timeout=300, + cwd=REPO_ROOT, + env={**os.environ, "DATABASE_URL": database_url}, + ) + assert entrypoint.returncode == 0, entrypoint.stdout + entrypoint.stderr + assert object_permission_columns(database_url) == ("mcp_tool_search_enabled",), entrypoint.stdout + assert applied_migrations(database_url) == SHIPPED_MIGRATIONS, entrypoint.stdout + with owned_proxy( + gateway, tmp_path, {"DATABASE_URL": database_url, "DISABLE_SCHEMA_UPDATE": "true"} + ) as upgraded: + tools: Final = upgraded.request("GET", "/mcp-rest/tools/list") + assert tools.status_code == 200, tools.text + assert tools.json()["tools"] == [], tools.text diff --git a/tests/integration/management/test_guardrail_usage_config_guardrail.py b/tests/integration/management/test_guardrail_usage_config_guardrail.py new file mode 100644 index 00000000000..b9ba795b89b --- /dev/null +++ b/tests/integration/management/test_guardrail_usage_config_guardrail.py @@ -0,0 +1,77 @@ +import uuid +from pathlib import Path +from typing import Final + +import pytest +import yaml +from pydantic import JsonValue + +from tests.integration._support.client import Gateway, object_value, string_value +from tests.integration._support.process import owned_proxy + +WINDOW: Final = {"start_date": "2026-01-01", "end_date": "2026-01-07"} + + +def listed_guardrail(gateway: Gateway, guardrail_name: str) -> dict[str, JsonValue]: + listed: Final = gateway.get("/v2/guardrails/list")["guardrails"] + assert isinstance(listed, list), listed + matches: Final = tuple(object_value(row) for row in listed if object_value(row)["guardrail_name"] == guardrail_name) + assert len(matches) == 1, f"{guardrail_name} appears {len(matches)} times in {listed}" + return matches[0] + + +@pytest.mark.covers("mgmt.guardrails.usage.config_yaml_guardrail_has_detail_and_overview_row") +def test_config_yaml_guardrail_is_served_by_usage_detail_and_overview(gateway: Gateway, tmp_path: Path) -> None: + guardrail_name: Final = "tool-permission-" + uuid.uuid4().hex + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["guardrails"] = [ + { + "guardrail_name": guardrail_name, + "litellm_params": { + "guardrail": "tool_permission", + "mode": "post_call", + "default_on": False, + "rules": [{"id": "deny_delete", "tool_name": "(?i)^.*(delete|drop).*", "decision": "deny"}], + "default_action": "allow", + "on_disallowed_action": "block", + }, + "guardrail_info": {"type": "Tool Permission", "description": "declared in config.yaml"}, + } + ] + path: Final = tmp_path / "guardrail.yaml" + path.write_text(yaml.safe_dump(config)) + with owned_proxy(gateway, tmp_path, {}, config=path) as candidate: + guardrail_id: Final = string_value(listed_guardrail(candidate, guardrail_name)["guardrail_id"]) + detail: Final = candidate.request("GET", f"/guardrails/usage/detail/{guardrail_id}", params=WINDOW) + assert detail.status_code == 200, detail.text + body: Final = object_value(detail.json()) + assert { + "guardrail_id": body["guardrail_id"], + "guardrail_name": body["guardrail_name"], + "provider": body["provider"], + "type": body["type"], + "description": body["description"], + "requestsEvaluated": body["requestsEvaluated"], + "failRate": body["failRate"], + } == { + "guardrail_id": guardrail_id, + "guardrail_name": guardrail_name, + "provider": "tool_permission", + "type": "Tool Permission", + "description": "declared in config.yaml", + "requestsEvaluated": 0, + "failRate": 0.0, + }, detail.text + overview: Final = candidate.request("GET", "/guardrails/usage/overview", params=WINDOW) + assert overview.status_code == 200, overview.text + rows: Final = object_value(overview.json())["rows"] + assert isinstance(rows, list), overview.text + config_rows: Final = tuple(object_value(row) for row in rows if object_value(row)["id"] == guardrail_id) + assert len(config_rows) == 1, overview.text + assert (config_rows[0]["name"], config_rows[0]["provider"], config_rows[0]["requestsEvaluated"]) == ( + guardrail_name, + "tool_permission", + 0, + ), overview.text + missing: Final = candidate.request("GET", f"/guardrails/usage/detail/{uuid.uuid4()}", params=WINDOW) + assert missing.status_code == 404, missing.text diff --git a/tests/integration/observability/test_guardrail_effects.py b/tests/integration/observability/test_guardrail_effects.py index f96cae593dc..4fac42a796d 100644 --- a/tests/integration/observability/test_guardrail_effects.py +++ b/tests/integration/observability/test_guardrail_effects.py @@ -5,7 +5,8 @@ from typing import Final import pytest import yaml -from integration._support.client import Gateway +from integration._support.client import Gateway, eventually, object_value +from integration._support.database import read_rows from integration._support.mcp import mcp_peer, register_mcp, tool_names from integration._support.process import owned_proxy from integration._support.wire import Reply, Request, wire_server @@ -88,6 +89,91 @@ def test_guardrail_rewrites_system_and_user_in_actual_anthropic_request(gateway: assert len(policy.drain()) == len(upstream.drain()) == 1 +@pytest.mark.covers("other.observability.guardrails.anthropic_messages_caller_metadata_keeps_guardrail_spend_log") +def test_anthropic_messages_with_caller_metadata_keeps_guardrail_information_in_spend_log( + gateway: Gateway, tmp_path: Path +) -> None: + identity: Final = "guardrail" + uuid.uuid4().hex + prompt: Final = "synthetic allowed prompt " + identity + caller_metadata: Final = {"user_id": "device-account-session"} + + def guardrail(request: Request) -> Reply: + assert request.target == "/beta/litellm_basic_guardrail_api" + assert json.loads(request.body)["texts"] == [prompt] + return Reply(body=json.dumps({"action": "NONE"}).encode()) + + def provider(request: Request) -> Reply: + assert request.target == "/v1/messages" + body: Final = json.loads(request.body) + assert body["messages"] == [{"role": "user", "content": prompt}] + assert body["metadata"] == caller_metadata + return Reply( + body=json.dumps( + { + "id": identity, + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5-20250929", + "content": [{"type": "text", "text": "permitted response"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 11, "output_tokens": 4}, + } + ).encode() + ) + + with wire_server(guardrail) as policy, wire_server(provider) as upstream: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["guardrails"] = [ + { + "guardrail_name": identity, + "litellm_params": { + "guardrail": "generic_guardrail_api", + "mode": "pre_call", + "default_on": True, + "api_base": policy.url, + "api_key": "synthetic-guardrail-key", + }, + } + ] + path: Final = tmp_path / "caller-metadata.yaml" + path.write_text(yaml.safe_dump(config)) + with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model="anthropic/claude-sonnet-4-5-20250929", api_base=upstream.url, api_key="synthetic-anthropic-key" + ) + response: Final = candidate.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 16, + "messages": [{"role": "user", "content": prompt}], + "metadata": caller_metadata, + }, + ) + assert response.status_code == 200, response.text + assert response.json()["content"] == [{"type": "text", "text": "permitted response"}], response.text + assert response.headers["x-litellm-applied-guardrails"] == identity, dict(response.headers) + assert len(policy.drain()) == len(upstream.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT call_type, metadata FROM "LiteLLM_SpendLogs" WHERE model_group=%s', + (model,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["call_type"] == "anthropic_messages", rows[0] + saved: Final = object_value(rows[0]["metadata"]) + entries: Final = saved["guardrail_information"] + assert isinstance(entries, list) and len(entries) == 1, saved + entry: Final = object_value(entries[0]) + assert entry["guardrail_name"] == identity, saved + assert entry["guardrail_mode"] == "pre_call", saved + assert entry["guardrail_status"] == "success", saved + + @pytest.mark.covers("other.observability.guardrails.denial_prevents_provider_with_allowed_control") def test_guardrail_denial_prevents_provider_and_preserves_allowed_control(gateway: Gateway, tmp_path: Path) -> None: identity: Final = "guardrail" + uuid.uuid4().hex @@ -242,6 +328,110 @@ def test_bedrock_passthrough_converse_guardrail_ignores_denied_term_in_tool_defi assert [json.loads(request.body)["texts"] for request in policy.drain()] == [[allowed], [denied]] +@pytest.mark.covers("other.observability.guardrails.bedrock_post_call_scans_streamed_anthropic_messages_tool_use") +def test_bedrock_guardrail_streams_anthropic_messages_tool_use_instead_of_chunk_builder_500( + gateway: Gateway, tmp_path: Path +) -> None: + identity: Final = "guardrail" + uuid.uuid4().hex + guardrail_id: Final = "synthetic" + uuid.uuid4().hex[:8] + spoken: Final = "Checking the forecast" + frames: Final = ( + 'event: message_start\ndata: {"type": "message_start", "message": {"id": "msg_synthetic", "type": "message", ' + '"role": "assistant", "model": "claude-sonnet-4-5-20250929", "content": [], "stop_reason": null, ' + '"stop_sequence": null, "usage": {"input_tokens": 11, "output_tokens": 1}}}\n\n', + 'event: content_block_start\ndata: {"type": "content_block_start", "index": 0, ' + '"content_block": {"type": "text", "text": ""}}\n\n', + 'event: content_block_delta\ndata: {"type": "content_block_delta", "index": 0, ' + f'"delta": {{"type": "text_delta", "text": "{spoken}"}}}}\n\n', + 'event: content_block_stop\ndata: {"type": "content_block_stop", "index": 0}\n\n', + 'event: content_block_start\ndata: {"type": "content_block_start", "index": 1, ' + '"content_block": {"type": "tool_use", "id": "toolu_synthetic", "name": "lookup_weather", "input": {}}}\n\n', + 'event: content_block_delta\ndata: {"type": "content_block_delta", "index": 1, ' + '"delta": {"type": "input_json_delta", "partial_json": "{\\"city\\": \\"Paris\\"}"}}\n\n', + 'event: content_block_stop\ndata: {"type": "content_block_stop", "index": 1}\n\n', + 'event: message_delta\ndata: {"type": "message_delta", "delta": {"stop_reason": "tool_use", ' + '"stop_sequence": null}, "usage": {"output_tokens": 9}}\n\n', + 'event: message_stop\ndata: {"type": "message_stop"}\n\n', + ) + tools: Final = [ + { + "name": "lookup_weather", + "description": "Look up the forecast for a city", + "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, + } + ] + + def guardrail(request: Request) -> Reply: + assert request.target == f"/guardrail/{guardrail_id}/version/DRAFT/apply", request.target + body: Final = json.loads(request.body) + assert body["source"] == "OUTPUT", body + assert body["content"] == [{"text": {"text": spoken}}], body + return Reply(body=json.dumps({"action": "NONE", "outputs": [], "assessments": []}).encode()) + + def provider(request: Request) -> Reply: + assert request.target == "/v1/messages" + body: Final = json.loads(request.body) + assert body["stream"] is True, body + assert body["tools"] == tools, body + return Reply(content_type="text/event-stream", chunks=tuple(frame.encode() for frame in frames)) + + with wire_server(guardrail) as policy, wire_server(provider) as upstream: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["guardrails"] = [ + { + "guardrail_name": identity, + "litellm_params": { + "guardrail": "bedrock", + "mode": "post_call", + "default_on": True, + "mask_response_content": True, + "guardrailIdentifier": guardrail_id, + "guardrailVersion": "DRAFT", + "aws_region_name": "us-east-1", + "aws_access_key_id": "AKIASYNTHETICGUARDRAIL", + "aws_secret_access_key": "synthetic-secret", + "aws_bedrock_runtime_endpoint": policy.url, + }, + } + ] + path: Final = tmp_path / "bedrock-stream.yaml" + path.write_text(yaml.safe_dump(config)) + with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model="anthropic/claude-sonnet-4-5-20250929", api_base=upstream.url, api_key="synthetic-anthropic-key" + ) + response: Final = candidate.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 64, + "stream": True, + "tools": tools, + "messages": [{"role": "user", "content": f"What is the weather in Paris? {identity}"}], + }, + ) + assert response.status_code == 200, response.text + head, separator, tail = response.text.partition("\n\n") + assert separator == "\n\n", response.text + assert head.startswith("event: message_start\ndata: "), response.text + assert json.loads(head.removeprefix("event: message_start\ndata: ")) == { + "type": "message_start", + "message": { + "id": "msg_synthetic", + "type": "message", + "role": "assistant", + "model": model, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 11, "output_tokens": 1}, + }, + }, response.text + assert tail == "".join(frames[1:]), response.text + assert len(policy.drain()) == len(upstream.drain()) == 1 + + @pytest.mark.covers("other.mcp.guardrails.request_selection_blocks_resolved_tool_without_execution") def test_request_selected_mcp_guardrail_blocks_direct_and_virtual_calls(gateway: Gateway, tmp_path: Path) -> None: guardrail = "mcp-policy-" + uuid.uuid4().hex diff --git a/tests/integration/pricing/test_configured_prices.py b/tests/integration/pricing/test_configured_prices.py index e39833516c2..0e4efea3a15 100644 --- a/tests/integration/pricing/test_configured_prices.py +++ b/tests/integration/pricing/test_configured_prices.py @@ -8,8 +8,10 @@ import pytest import yaml from pydantic import JsonValue +from litellm import get_model_info from tests.integration._support.client import Gateway, eventually, object_value, string_value from tests.integration._support.database import read_rows +from tests.integration._support.process import owned_proxy @pytest.mark.covers("quota_management.spend_tracking.custom_price.matches_input_rates") @@ -169,6 +171,49 @@ def test_saving_echoed_model_info_does_not_freeze_cost_map_price_into_deployment assert {key: value for key, value in stored.items() if key in COST_MAP_DISPLAY_PRICING_KEYS} == {}, stored +def displayed_model_info(gateway: Gateway, model: str) -> dict[str, JsonValue]: + entries: Final = gateway.get("/model/info")["data"] + assert isinstance(entries, list) + target: Final = next(object_value(entry) for entry in entries if object_value(entry)["model_name"] == model) + return object_value(target["model_info"]) + + +@pytest.mark.covers("pricing.model_update.echoed_cost_map_metadata_is_not_persisted_as_override") +def test_saving_echoed_model_info_does_not_persist_cost_map_metadata_as_overrides(gateway: Gateway) -> None: + catalog_entry: Final = get_model_info("openai/gpt-4o-mini") + with gateway.scenario() as scenario: + model: Final = scenario.model() + displayed: Final = displayed_model_info(gateway, model) + identity: Final = string_value(displayed["id"]) + assert displayed["key"] == catalog_entry["key"], displayed + assert displayed["max_input_tokens"] == catalog_entry["max_input_tokens"], displayed + saved: Final = gateway.request( + "PATCH", f"/model/{identity}/update", {"model_info": {**displayed, "description": "echoed ui save"}} + ) + assert saved.status_code == 200, saved.text + stored: Final = persisted_model_info(identity) + assert stored["description"] == "echoed ui save", stored + assert {key: value for key, value in stored.items() if key in catalog_entry} == {}, stored + + +@pytest.mark.covers("pricing.model_update.echoing_cost_map_value_back_clears_stored_override") +def test_saving_the_cost_map_value_back_over_a_stored_override_clears_it(gateway: Gateway, tmp_path: Path) -> None: + catalog_limit: Final = get_model_info("openai/gpt-4o-mini")["max_input_tokens"] + assert isinstance(catalog_limit, int) and catalog_limit != 4321, catalog_limit + with owned_proxy(gateway, tmp_path, {}) as candidate, candidate.scenario() as scenario: + overridden: Final = scenario.model(model_info={"max_input_tokens": 4321}) + displayed: Final = displayed_model_info(candidate, overridden) + identity: Final = string_value(displayed["id"]) + assert displayed["max_input_tokens"] == 4321, displayed + assert persisted_model_info(identity)["max_input_tokens"] == 4321 + saved: Final = candidate.request( + "PATCH", f"/model/{identity}/update", {"model_info": {**displayed, "max_input_tokens": catalog_limit}} + ) + assert saved.status_code == 200, saved.text + stored: Final = persisted_model_info(identity) + assert "max_input_tokens" not in stored, stored + + @pytest.mark.covers("quota_management.spend_tracking.default_prices.loaded_router_preserves_cached_defaults") def test_loaded_router_preserves_cached_defaults_during_real_requests(gateway: Gateway, tmp_path: Path) -> None: from litellm import Router diff --git a/tests/integration/pricing/test_realtime_cached_audio_pricing.py b/tests/integration/pricing/test_realtime_cached_audio_pricing.py new file mode 100644 index 00000000000..4a7598d0cbf --- /dev/null +++ b/tests/integration/pricing/test_realtime_cached_audio_pricing.py @@ -0,0 +1,131 @@ +import asyncio +import json +import os +import uuid +from hashlib import sha256 +from typing import Final + +import pytest +import websockets +from pydantic import BaseModel, ConfigDict, JsonValue + +import litellm +from tests.integration._support.client import JSON_OBJECT, Gateway, eventually +from tests.integration._support.database import read_rows +from tests.integration._support.upstream import delete_scenario, register_scenario +from tests.integration.cost_calculation.cost_tracking_case import RealtimeResponse + + +class RealtimeRates(BaseModel): + model_config = ConfigDict(frozen=True) + + input_cost_per_token: float + input_cost_per_audio_token: float + cache_read_input_token_cost: float + cache_read_input_audio_token_cost: float | None = None + output_cost_per_token: float + output_cost_per_audio_token: float + + +MODEL: Final = "gpt-realtime-2" +RATES: Final = RealtimeRates.model_validate(litellm.get_model_info(MODEL, custom_llm_provider="openai")) +TEXT_RATE: Final = RATES.input_cost_per_token +AUDIO_RATE: Final = RATES.input_cost_per_audio_token +CACHED_TEXT_RATE: Final = RATES.cache_read_input_token_cost +CACHED_AUDIO_RATE: Final = ( + RATES.cache_read_input_token_cost + if RATES.cache_read_input_audio_token_cost is None + else RATES.cache_read_input_audio_token_cost +) +OUTPUT_TEXT_RATE: Final = RATES.output_cost_per_token +OUTPUT_AUDIO_RATE: Final = RATES.output_cost_per_audio_token +INPUT_TEXT_TOKENS: Final = 116 +INPUT_AUDIO_TOKENS: Final = 167 +CACHED_TEXT_TOKENS: Final = 64 +CACHED_AUDIO_TOKENS: Final = 128 +INPUT_TOKENS: Final = INPUT_TEXT_TOKENS + INPUT_AUDIO_TOKENS +OUTPUT_TEXT_TOKENS: Final = 8 +OUTPUT_AUDIO_TOKENS: Final = 12 +OUTPUT_TOKENS: Final = OUTPUT_TEXT_TOKENS + OUTPUT_AUDIO_TOKENS +EXPECTED_INPUT_COST: Final = ( + (INPUT_TEXT_TOKENS - CACHED_TEXT_TOKENS) * TEXT_RATE + + CACHED_TEXT_TOKENS * CACHED_TEXT_RATE + + (INPUT_AUDIO_TOKENS - CACHED_AUDIO_TOKENS) * AUDIO_RATE + + CACHED_AUDIO_TOKENS * CACHED_AUDIO_RATE +) +EXPECTED_OUTPUT_COST: Final = OUTPUT_TEXT_TOKENS * OUTPUT_TEXT_RATE + OUTPUT_AUDIO_TOKENS * OUTPUT_AUDIO_RATE + + +def cached_audio_response_done() -> RealtimeResponse: + return RealtimeResponse( + content_type="application/x-realtime", + events=( + { + "type": "response.done", + "event_id": "evt_$REQUEST_ID", + "response": { + "id": "resp_$REQUEST_ID", + "object": "realtime.response", + "status": "completed", + "output": [], + "usage": { + "total_tokens": INPUT_TOKENS + OUTPUT_TOKENS, + "input_tokens": INPUT_TOKENS, + "output_tokens": OUTPUT_TOKENS, + "input_token_details": { + "text_tokens": INPUT_TEXT_TOKENS, + "audio_tokens": INPUT_AUDIO_TOKENS, + "cached_tokens": CACHED_TEXT_TOKENS + CACHED_AUDIO_TOKENS, + "cached_tokens_details": { + "text_tokens": CACHED_TEXT_TOKENS, + "audio_tokens": CACHED_AUDIO_TOKENS, + }, + }, + "output_token_details": { + "text_tokens": OUTPUT_TEXT_TOKENS, + "audio_tokens": OUTPUT_AUDIO_TOKENS, + }, + }, + }, + }, + ), + ) + + +async def _one_realtime_turn(proxy_url: str, key: str, model: str) -> dict[str, JsonValue]: + async with websockets.connect( + f"{proxy_url.replace('http://', 'ws://').replace('https://', 'wss://')}/v1/realtime?model={model}", + additional_headers={"Authorization": f"Bearer {key}"}, + ) as websocket: + session: Final = JSON_OBJECT.validate_json(await websocket.recv()) + await websocket.send(json.dumps({"type": "response.create"})) + async for message in websocket: + if JSON_OBJECT.validate_json(message).get("type") == "response.done": + return session + raise AssertionError(f"websocket closed before response.done for {model}") + + +@pytest.mark.covers("pricing.realtime.cached_audio_tokens_bill_at_audio_cache_read_rate") +def test_realtime_cached_audio_tokens_bill_at_audio_cache_read_rate_not_full_audio_rate(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + scenario_id: Final = f"realtime-cached-audio-{uuid.uuid4().hex[:12]}" + handle: Final = register_scenario(scenario_id, cached_audio_response_done()) + scenario.cleanups.callback(delete_scenario, handle) + key: Final = scenario.key() + model: Final = scenario.model( + model=f"openai/{MODEL}", api_key=scenario_id, api_base=gateway.upstream_url.rstrip("/") + ) + session: Final = asyncio.run(_one_realtime_turn(os.environ["INTEGRATION_PROXY_URL"].rstrip("/"), key, model)) + assert session.get("type") == "session.created", session + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, prompt_tokens, completion_tokens, call_type FROM "LiteLLM_SpendLogs" WHERE api_key = %s', + (sha256(key.encode()).hexdigest(),), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["call_type"] == "_arealtime", rows + assert rows[0]["prompt_tokens"] == INPUT_TOKENS, rows + assert rows[0]["completion_tokens"] == OUTPUT_TOKENS, rows + assert float(str(rows[0]["spend"])) == pytest.approx(EXPECTED_INPUT_COST + EXPECTED_OUTPUT_COST, rel=1e-6), rows diff --git a/tests/integration/spend/test_cache_and_quota.py b/tests/integration/spend/test_cache_and_quota.py index 97ef785aa15..1c2b5551855 100644 --- a/tests/integration/spend/test_cache_and_quota.py +++ b/tests/integration/spend/test_cache_and_quota.py @@ -1,19 +1,27 @@ import json +import os import threading import uuid +from collections.abc import Generator from concurrent.futures import ThreadPoolExecutor -from contextlib import ExitStack +from contextlib import ExitStack, contextmanager from hashlib import sha256 +from pathlib import Path from typing import Final +from urllib.parse import urlsplit, urlunsplit import httpx +import psycopg import pytest from hypothesis import strategies as st from hypothesis.stateful import RuleBasedStateMachine, rule, run_state_machine_as_test -from integration._support.client import Gateway, eventually +from integration._support.client import Gateway, eventually, string_value from integration._support.database import read_rows +from integration._support.database_relay import database_relay from integration._support.generation import LIFECYCLE_SETTINGS, bounded_http_requests +from integration._support.process import owned_proxy from integration._support.wire import Reply, Request, wire_server +from psycopg import sql @pytest.mark.covers("quota_management.response_cache.generated_sequences_preserve_content_and_accounting") @@ -214,6 +222,99 @@ def test_key_budget_at_boundary_blocks_provider_then_explicit_reset_restores(gat assert upstream.get("/__observations").json()["requests"] == [] +RESET_SWEEP_QUERY: Final = b'"LiteLLM_VerificationToken"."budget_reset_at" < $' + + +@contextmanager +def scratch_database() -> Generator[str]: + name: Final = f"integration_{uuid.uuid4().hex}" + with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as admin: + admin.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(name))) + try: + yield urlunsplit(urlsplit(os.environ["DATABASE_URL"])._replace(path=f"/{name}")) + finally: + admin.execute(sql.SQL("DROP DATABASE {} WITH (FORCE)").format(sql.Identifier(name))) + + +@pytest.mark.covers("quota_management.budget.key.scheduled_reset_survives_transient_db_outage") +@pytest.mark.timeout(300) +def test_scheduled_budget_reset_reconnects_after_db_transport_failure_and_unblocks_key( + gateway: Gateway, tmp_path: Path +) -> None: + with ( + scratch_database() as scratch_url, + database_relay(scratch_url, RESET_SWEEP_QUERY) as (relay, relayed_url), + httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream, + owned_proxy( + gateway, + tmp_path, + { + "DATABASE_URL": relayed_url, + "PROXY_BUDGET_RESCHEDULER_MIN_TIME": "30", + "PROXY_BUDGET_RESCHEDULER_MAX_TIME": "30", + "PRISMA_HEALTH_WATCHDOG_ENABLED": "false", + }, + ) as candidate, + ): + model: Final = f"integration-{uuid.uuid4().hex}" + candidate.post( + "/model/new", + { + "model_name": model, + "litellm_params": { + "model": "openai/gpt-4o-mini", + "api_key": "integration-provider-key", + "api_base": f"{gateway.upstream_url}/v1", + "input_cost_per_token": 0.001, + "output_cost_per_token": 0.002, + }, + "model_info": {}, + }, + ) + key: Final = string_value( + candidate.post("/key/generate", {"models": [model], "max_budget": 0.06, "budget_duration": "5s"})["key"] + ) + digest: Final = sha256(key.encode()).hexdigest() + row_query: Final = ( + 'SELECT spend, budget_reset_at::text AS budget_reset_at FROM "LiteLLM_VerificationToken" WHERE token=%s' + ) + assert candidate.chat(model, key=key, text=f"spend it {uuid.uuid4().hex}")["usage"]["total_tokens"] == 40 + exhausted: Final = eventually( + lambda: read_rows(row_query, (digest,), database_url=scratch_url), + lambda rows: len(rows) == 1 and float(rows[0]["spend"]) >= 0.06, + seconds=70, + ) + assert float(exhausted[0]["spend"]) == pytest.approx(0.06) + upstream.get("/__observations").raise_for_status() + denied: Final = candidate.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": f"over budget {uuid.uuid4().hex}"}]}, + key=key, + ) + assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text + assert upstream.get("/__observations").json()["requests"] == [] + relay.arm() + assert relay.tripped.wait(90), "Scheduled reset sweep never reached the database" + eventually(lambda: relay.refused, lambda count: count >= 1, seconds=30) + reset: Final = eventually( + lambda: read_rows(row_query, (digest,), database_url=scratch_url), + lambda rows: len(rows) == 1 and float(rows[0]["spend"]) == 0, + seconds=80, + return_last_on_timeout=True, + ) + assert len(reset) == 1 and reset[0]["spend"] == 0.0, (exhausted, reset) + assert str(reset[0]["budget_reset_at"]) > str(exhausted[0]["budget_reset_at"]), (exhausted, reset) + prompt: Final = f"after reset {uuid.uuid4().hex}" + recovered: Final = candidate.request( + "POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": prompt}]}, key=key + ) + assert recovered.status_code == 200, recovered.text + assert recovered.json()["usage"]["total_tokens"] == 40, recovered.text + reached: Final = upstream.get("/__observations").json()["requests"] + assert len(reached) == 1 and reached[0]["body"]["messages"] == [{"role": "user", "content": prompt}], reached + + @pytest.mark.covers("quota_management.budget.key.count_tokens_reserves_nothing_so_completion_within_budget_succeeds") def test_repeated_count_tokens_on_budgeted_key_does_not_reserve_budget_or_block_later_completion( gateway: Gateway, @@ -333,6 +434,7 @@ def test_different_system_messages_do_not_share_a_cached_response(gateway: Gatew ): model: Final = scenario.model() prompt: Final = uuid.uuid4().hex + def completion_id(system: str, expected_calls: int) -> str: upstream.get("/__observations").raise_for_status() response: Final = gateway.request( diff --git a/tests/integration/spend/test_end_user_spend_without_proxy_user.py b/tests/integration/spend/test_end_user_spend_without_proxy_user.py new file mode 100644 index 00000000000..15d77f16db4 --- /dev/null +++ b/tests/integration/spend/test_end_user_spend_without_proxy_user.py @@ -0,0 +1,29 @@ +import uuid +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows + + +@pytest.mark.covers("spend.end_user.charged_when_key_has_no_user_id_and_auth_cache_is_redis") +def test_end_user_spend_lands_for_key_without_user_id_when_auth_cache_is_redis(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002) + key: Final = scenario.key(models=[model]) + end_user: Final = f"integration-end-user-{uuid.uuid4().hex}" + scenario.cleanups.callback(gateway.request, "POST", "/customer/delete", {"user_ids": [end_user]}) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": f"end user spend {end_user}"}], "user": end_user}, + key=key, + ) + assert response.status_code == 200, response.text + assert response.json()["usage"]["total_tokens"] == 40, response.text + charged: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_EndUserTable" WHERE user_id=%s', (end_user,)), + lambda values: len(values) == 1 and float(values[0]["spend"]) >= 0.06, + seconds=70, + ) + assert float(charged[0]["spend"]) == pytest.approx(20 * 0.001 + 20 * 0.002) diff --git a/tests/integration/spend/test_legacy_spend_logs_row_cap.py b/tests/integration/spend/test_legacy_spend_logs_row_cap.py new file mode 100644 index 00000000000..3a7ac8a83bd --- /dev/null +++ b/tests/integration/spend/test_legacy_spend_logs_row_cap.py @@ -0,0 +1,46 @@ +import os +import uuid +from datetime import datetime, timedelta, timezone +from typing import Final + +import psycopg +import pytest +from integration._support.client import Gateway +from integration._support.database import read_rows + +LEGACY_SPEND_LOGS_ROW_CAP: Final = 10000 + + +def _seed_spend_rows(user_id: str, count: int) -> tuple[str, ...]: + started: Final = datetime.now(timezone.utc) - timedelta(days=1) + with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection: + connection.execute( + 'INSERT INTO "LiteLLM_SpendLogs" ' + '(request_id, call_type, "startTime", "endTime", "user", status) ' + "SELECT %s || '-' || n, 'acompletion', %s + n * interval '1 second', %s + n * interval '1 second', %s, " + "'success' FROM generate_series(1, %s) AS n", + (user_id, started, started, user_id, count), + ) + return tuple(f"{user_id}-{n}" for n in range(1, count + 1)) + + +def _delete_spend_rows(user_id: str) -> None: + with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection: + connection.execute('DELETE FROM "LiteLLM_SpendLogs" WHERE "user" = %s', (user_id,)) + + +@pytest.mark.covers("spend.legacy_spend_logs.row_count_is_capped_at_the_most_recent_rows_and_flagged_truncated") +def test_legacy_spend_logs_returns_only_the_cap_of_most_recent_rows_and_flags_truncation(gateway: Gateway) -> None: + user_id: Final = f"integration-cap-{uuid.uuid4().hex}" + with gateway.scenario() as scenario: + scenario.cleanups.callback(_delete_spend_rows, user_id) + seeded: Final = _seed_spend_rows(user_id, LEGACY_SPEND_LOGS_ROW_CAP + 1) + assert read_rows('SELECT count(*)::text AS total FROM "LiteLLM_SpendLogs" WHERE "user" = %s', (user_id,)) == [ + {"total": str(LEGACY_SPEND_LOGS_ROW_CAP + 1)} + ] + response: Final = gateway.request("GET", "/spend/logs", params={"user_id": user_id}) + assert response.status_code == 200, response.text + returned: Final = tuple(row["request_id"] for row in response.json()) + assert len(returned) == LEGACY_SPEND_LOGS_ROW_CAP, f"{len(returned)} rows: {response.text[:300]}" + assert returned == tuple(reversed(seeded[1:])), response.text[:300] + assert response.headers.get("x-litellm-spend-logs-truncated") == "true", dict(response.headers) diff --git a/tests/integration/spend/test_messages_stream_usage_cost.py b/tests/integration/spend/test_messages_stream_usage_cost.py new file mode 100644 index 00000000000..692dd01e0b0 --- /dev/null +++ b/tests/integration/spend/test_messages_stream_usage_cost.py @@ -0,0 +1,176 @@ +import base64 +import json +import uuid +from pathlib import Path +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.process import owned_proxy +from integration._support.upstream import _aws_event_frame +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue, TypeAdapter + +BEDROCK_MODEL: Final = "anthropic.claude-haiku-4-5-20251001-v1:0" +INPUT_TOKENS: Final = 30 +CACHE_READ_TOKENS: Final = 900 +CACHE_CREATION_TOKENS: Final = 400 +OUTPUT_TOKENS: Final = 57 +INPUT_RATE: Final = 0.001 +OUTPUT_RATE: Final = 0.002 +CACHE_READ_RATE: Final = 0.0001 +CACHE_CREATION_RATE: Final = 0.00125 +STREAMED_USAGE: Final = TypeAdapter(dict[str, float]) +EXPECTED_SPEND: Final = ( + INPUT_TOKENS * INPUT_RATE + + CACHE_READ_TOKENS * CACHE_READ_RATE + + CACHE_CREATION_TOKENS * CACHE_CREATION_RATE + + OUTPUT_TOKENS * OUTPUT_RATE +) + + +def _invoke_chunk(payload: dict[str, JsonValue]) -> bytes: + encoded: Final = base64.b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode() + return _aws_event_frame("chunk", {"bytes": encoded}, "", "") + + +def _stream(message_id: str) -> bytes: + return ( + _invoke_chunk( + { + "type": "message_start", + "message": { + "id": message_id, + "type": "message", + "role": "assistant", + "model": BEDROCK_MODEL, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": { + "input_tokens": INPUT_TOKENS, + "cache_read_input_tokens": CACHE_READ_TOKENS, + "cache_creation_input_tokens": CACHE_CREATION_TOKENS, + "output_tokens": 0, + }, + }, + } + ) + + _invoke_chunk({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}) + + _invoke_chunk( + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "cached answer"}} + ) + + _invoke_chunk({"type": "content_block_stop", "index": 0}) + + _invoke_chunk( + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn"}, + "usage": { + "input_tokens": INPUT_TOKENS, + "cache_read_input_tokens": CACHE_READ_TOKENS, + "cache_creation_input_tokens": CACHE_CREATION_TOKENS, + "output_tokens": OUTPUT_TOKENS, + }, + } + ) + + _invoke_chunk({"type": "message_stop"}) + ) + + +def _proxy_config(directory: Path, model: str, upstream_url: str) -> Path: + config: Final = directory / "streamed_usage_cost_config.yaml" + config.write_text( + json.dumps( + { + "model_list": [ + { + "model_name": model, + "litellm_params": { + "model": f"bedrock/invoke/{BEDROCK_MODEL}", + "api_base": upstream_url, + "aws_access_key_id": "AKIASCRIPTEDPROVIDER", + "aws_secret_access_key": "scripted-secret", + "aws_region_name": "us-east-1", + "input_cost_per_token": INPUT_RATE, + "output_cost_per_token": OUTPUT_RATE, + "cache_read_input_token_cost": CACHE_READ_RATE, + "cache_creation_input_token_cost": CACHE_CREATION_RATE, + }, + } + ], + "general_settings": { + "master_key": "os.environ/LITELLM_MASTER_KEY", + "database_url": "os.environ/DATABASE_URL", + "store_model_in_db": True, + "disable_spend_logs": False, + "proxy_batch_write_at": 1, + "proxy_batch_polling_interval": 1, + }, + "litellm_settings": {"include_cost_in_streaming_usage": True}, + "router_settings": {"disable_cooldowns": True}, + } + ) + ) + return config + + +def _data_events(body: str) -> tuple[dict[str, JsonValue], ...]: + return tuple(json.loads(line.removeprefix("data:")) for line in body.splitlines() if line.startswith("data:")) + + +@pytest.mark.covers("spend.anthropic_messages_stream.streamed_usage_cost_equals_recorded_spend") +@pytest.mark.timeout(180) +def test_bedrock_messages_stream_usage_cost_matches_recorded_spend_with_custom_cache_rates( + gateway: Gateway, tmp_path: Path +) -> None: + message_id: Final = f"msg_{uuid.uuid4().hex}" + model: Final = f"{BEDROCK_MODEL}-{uuid.uuid4().hex}" + + def respond(request: Request) -> Reply: + assert request.target == f"/model/{BEDROCK_MODEL}/invoke-with-response-stream", request.target + assert json.loads(request.body)["messages"] == [{"role": "user", "content": "cached cost control"}], ( + request.body + ) + return Reply(content_type="application/vnd.amazon.eventstream", chunks=(_stream(message_id),)) + + with wire_server(respond) as wire: + config: Final = _proxy_config(tmp_path, model, wire.url) + with owned_proxy(gateway, tmp_path, {}, config=config) as candidate: + response: Final = candidate.request( + "POST", + "/v1/messages", + { + "model": model, + "messages": [{"role": "user", "content": "cached cost control"}], + "max_tokens": OUTPUT_TOKENS, + "stream": True, + }, + ) + assert response.status_code == 200, response.text + message_delta: Final = next( + event for event in _data_events(response.text) if event["type"] == "message_delta" + ) + streamed_usage: Final = STREAMED_USAGE.validate_python(message_delta["usage"]) + rows: Final = eventually( + lambda: read_rows( + 'SELECT status, prompt_tokens, completion_tokens, spend FROM "LiteLLM_SpendLogs" ' + "WHERE request_id=%s", + (message_id,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["status"] == "success", rows + assert rows[0]["prompt_tokens"] == INPUT_TOKENS + CACHE_READ_TOKENS + CACHE_CREATION_TOKENS, rows + assert rows[0]["completion_tokens"] == OUTPUT_TOKENS, rows + recorded_spend: Final = float(str(rows[0]["spend"])) + assert recorded_spend == pytest.approx(EXPECTED_SPEND), rows + assert streamed_usage == { + "input_tokens": INPUT_TOKENS, + "cache_read_input_tokens": CACHE_READ_TOKENS, + "cache_creation_input_tokens": CACHE_CREATION_TOKENS, + "output_tokens": OUTPUT_TOKENS, + "cost": pytest.approx(recorded_spend), + }, (streamed_usage, rows, response.text) + assert len(wire.drain()) == 1 diff --git a/tests/integration/spend/test_reset_budget_leader_election.py b/tests/integration/spend/test_reset_budget_leader_election.py new file mode 100644 index 00000000000..f9c165a0be0 --- /dev/null +++ b/tests/integration/spend/test_reset_budget_leader_election.py @@ -0,0 +1,75 @@ +import json +import os +from pathlib import Path +from types import MappingProxyType +from typing import Final + +import psycopg +import pytest +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.process import owned_proxy +from redis import Redis + +RESET_LEASE_KEY: Final = "cronjob_lock:reset_budget_job" +PEER_POD_LEASE: Final = json.dumps("integration-peer-pod-holding-the-reset-lease") +FAST_RESET_TICK: Final = MappingProxyType( + {"PROXY_BUDGET_RESCHEDULER_MIN_TIME": "2", "PROXY_BUDGET_RESCHEDULER_MAX_TIME": "2"} +) + + +def _team_spend(team: str) -> float: + rows: Final = read_rows('SELECT spend FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,)) + assert len(rows) == 1, rows + spend: Final = rows[0]["spend"] + assert isinstance(spend, (int, float)), rows + return float(spend) + + +def _make_team_budget_due(team: str, spend: float) -> None: + with psycopg.connect(os.environ["DATABASE_URL"]) as connection: + connection.execute( + "UPDATE \"LiteLLM_TeamTable\" SET spend = %s, budget_reset_at = now() - interval '1 day' " + "WHERE team_id = %s", + (spend, team), + ) + + +@pytest.mark.covers("spend.budget_reset.one_pod_sweeps_per_tick") +def test_reset_sweep_skips_ticks_while_another_pod_holds_the_lease_and_resumes_after_release( + gateway: Gateway, tmp_path: Path +) -> None: + with ( + gateway.scenario() as scenario, + Redis(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"])) as cache, + ): + model: Final = scenario.model(input_cost_per_token=0.0, output_cost_per_token=0.0) + team: Final = scenario.team(max_budget=1.0, budget_duration="30d") + key: Final = scenario.key(team_id=team) + _make_team_budget_due(team, spend=0.5) + assert _team_spend(team) == 0.5 + eventually( + lambda: cache.set(RESET_LEASE_KEY, PEER_POD_LEASE, ex=120, nx=True), + lambda claimed: claimed is True, + seconds=60, + ) + try: + with owned_proxy(gateway, tmp_path, FAST_RESET_TICK) as replica: + response: Final = replica.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": "classification request"}]}, + key=key, + ) + assert response.status_code == 200, response.text + held: Final = eventually( + lambda: _team_spend(team), lambda spend: spend != 0.5, seconds=8, return_last_on_timeout=True + ) + assert held == 0.5, f"team {team} was swept while another pod held the reset lease: spend={held}" + assert cache.get(RESET_LEASE_KEY) == PEER_POD_LEASE.encode() + cache.delete(RESET_LEASE_KEY) + swept: Final = eventually(lambda: _team_spend(team), lambda spend: spend == 0.0, seconds=15) + assert swept == 0.0 + eventually(lambda: cache.get(RESET_LEASE_KEY), lambda value: value is None, seconds=15) + finally: + cache.delete(RESET_LEASE_KEY) diff --git a/tests/integration/spend/test_responses_cache_write_itemization.py b/tests/integration/spend/test_responses_cache_write_itemization.py new file mode 100644 index 00000000000..19d73b0aacf --- /dev/null +++ b/tests/integration/spend/test_responses_cache_write_itemization.py @@ -0,0 +1,98 @@ +import json +import uuid +from typing import Final + +import pytest + +from tests.integration._support.client import Gateway, eventually, object_value, string_value +from tests.integration._support.database import read_rows +from tests.integration._support.upstream import delete_scenario, register_scenario +from tests.integration.cost_calculation.cost_tracking_case import JsonResponse + +INPUT_RATE: Final = 0.001 +OUTPUT_RATE: Final = 0.002 +CACHE_CREATION_RATE: Final = 0.004 +CACHE_READ_RATE: Final = 0.0001 +UNCACHED_INPUT_TOKENS: Final = 1000 +CACHE_WRITE_TOKENS: Final = 2000 +CACHED_TOKENS: Final = 8000 +INPUT_TOKENS: Final = UNCACHED_INPUT_TOKENS + CACHE_WRITE_TOKENS + CACHED_TOKENS +OUTPUT_TOKENS: Final = 500 + + +def responses_cache_write_response() -> JsonResponse: + return JsonResponse( + content_type="application/json", + body={ + "id": "resp_$REQUEST_ID", + "object": "response", + "created_at": 1700000000, + "status": "completed", + "model": "gpt-5.6", + "output": [ + { + "type": "message", + "id": "msg_$REQUEST_ID", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "scripted response", "annotations": []}], + } + ], + "usage": { + "input_tokens": INPUT_TOKENS, + "output_tokens": OUTPUT_TOKENS, + "total_tokens": INPUT_TOKENS + OUTPUT_TOKENS, + "input_tokens_details": { + "cached_tokens": CACHED_TOKENS, + "cache_write_tokens": CACHE_WRITE_TOKENS, + }, + "output_tokens_details": {"reasoning_tokens": 0}, + }, + }, + ) + + +@pytest.mark.covers("spend.responses_api.cache_write_tokens_itemized_as_cache_creation_cost") +def test_responses_cache_write_tokens_are_itemized_as_cache_creation_cost(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + scenario_id: Final = f"responses-cache-write-{uuid.uuid4().hex[:12]}" + handle: Final = register_scenario(scenario_id, responses_cache_write_response()) + scenario.cleanups.callback(delete_scenario, handle) + model: Final = scenario.model( + model="openai/gpt-5.6", + api_base=handle.api_base(), + input_cost_per_token=INPUT_RATE, + output_cost_per_token=OUTPUT_RATE, + cache_creation_input_token_cost=CACHE_CREATION_RATE, + cache_read_input_token_cost=CACHE_READ_RATE, + ) + response: Final = gateway.request("POST", "/v1/responses", {"model": model, "input": "cache write control"}) + assert response.status_code == 200, response.text + expected_cache_creation_cost: Final = CACHE_WRITE_TOKENS * CACHE_CREATION_RATE + expected_cache_read_cost: Final = CACHED_TOKENS * CACHE_READ_RATE + expected_input_cost: Final = ( + UNCACHED_INPUT_TOKENS * INPUT_RATE + expected_cache_creation_cost + expected_cache_read_cost + ) + expected_output_cost: Final = OUTPUT_TOKENS * OUTPUT_RATE + request_id: Final = string_value(object_value(response.json())["id"]) + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, metadata, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" ' + "WHERE request_id = %s", + (request_id,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["prompt_tokens"] == INPUT_TOKENS, response.text + assert rows[0]["completion_tokens"] == OUTPUT_TOKENS, response.text + assert float(rows[0]["spend"]) == pytest.approx(expected_input_cost + expected_output_cost, rel=1e-6), ( + response.text + ) + metadata: Final = rows[0]["metadata"] + parsed: Final = json.loads(metadata) if isinstance(metadata, str) else object_value(metadata) + breakdown: Final = object_value(parsed["cost_breakdown"]) + assert breakdown.get("cache_creation_cost") == pytest.approx(expected_cache_creation_cost, rel=1e-6), breakdown + assert breakdown.get("cache_read_cost") == pytest.approx(expected_cache_read_cost, rel=1e-6), breakdown + assert float(breakdown["input_cost"]) == pytest.approx(expected_input_cost, rel=1e-6), breakdown + assert float(breakdown["output_cost"]) == pytest.approx(expected_output_cost, rel=1e-6), breakdown diff --git a/tests/integration/spend/test_session_total_spend.py b/tests/integration/spend/test_session_total_spend.py new file mode 100644 index 00000000000..54df0c03d81 --- /dev/null +++ b/tests/integration/spend/test_session_total_spend.py @@ -0,0 +1,98 @@ +import json +import uuid +from datetime import datetime, timedelta, timezone +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + +INPUT_COST_PER_TOKEN: Final = 0.001 +OUTPUT_COST_PER_TOKEN: Final = 0.002 +ROUND_USAGE: Final = ((10, 5), (20, 10), (30, 15)) +ROUND_SPEND: Final = tuple( + prompt * INPUT_COST_PER_TOKEN + completion * OUTPUT_COST_PER_TOKEN for prompt, completion in ROUND_USAGE +) +SESSION_SPEND: Final = sum(ROUND_SPEND) + + +def _round_reply(request: Request, prompt: str, usage: tuple[int, int]) -> Reply: + assert request.method == "POST" and request.target == "/chat/completions", request.target + assert json.loads(request.body) == { + "model": "gpt-4o-mini", + "messages": [{"role": "user", "content": prompt}], + }, request.body + return Reply( + body=json.dumps( + { + "id": "chatcmpl-" + uuid.uuid4().hex, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": "round answer"}, "finish_reason": "stop"} + ], + "usage": {"prompt_tokens": usage[0], "completion_tokens": usage[1], "total_tokens": sum(usage)}, + } + ).encode() + ) + + +@pytest.mark.covers("spend.logs_ui.multi_round_session_total_spend_sums_every_round") +def test_logs_ui_session_total_spend_sums_every_round_of_a_multi_round_session(gateway: Gateway) -> None: + session_id: Final = f"session-{uuid.uuid4().hex}" + prompts: Final = tuple(f"round-{index}-{uuid.uuid4().hex}" for index in range(len(ROUND_USAGE))) + usage_by_prompt: Final = dict(zip(prompts, ROUND_USAGE, strict=True)) + + def provider(request: Request) -> Reply: + prompt: Final = str(json.loads(request.body)["messages"][0]["content"]) + return _round_reply(request, prompt, usage_by_prompt[prompt]) + + with wire_server(provider) as wire, gateway.scenario() as scenario: + alias: Final = scenario.model( + api_base=wire.url, + input_cost_per_token=INPUT_COST_PER_TOKEN, + output_cost_per_token=OUTPUT_COST_PER_TOKEN, + num_retries=0, + ) + request_ids: Final = tuple(_completed_round(gateway, alias, session_id, prompt) for prompt in prompts) + assert len(wire.drain()) == len(ROUND_USAGE) + rows: Final = eventually( + lambda: read_rows( + 'SELECT request_id, spend FROM "LiteLLM_SpendLogs" WHERE session_id=%s ORDER BY "startTime"', + (session_id,), + ), + lambda values: len(values) == len(ROUND_USAGE), + seconds=70, + ) + assert [row["request_id"] for row in rows] == list(request_ids), rows + assert [float(row["spend"]) for row in rows] == pytest.approx(list(ROUND_SPEND)), rows + now: Final = datetime.now(timezone.utc) + logs: Final = gateway.request( + "GET", + "/spend/logs/ui", + params={ + "session_id": session_id, + "start_date": (now - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"), + "end_date": (now + timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"), + }, + ) + assert logs.status_code == 200, logs.text + page: Final = logs.json() + assert page["total"] == len(ROUND_USAGE), logs.text + assert sorted(row["request_id"] for row in page["data"]) == sorted(request_ids), logs.text + assert [row["session_total_count"] for row in page["data"]] == [len(ROUND_USAGE)] * len(ROUND_USAGE), logs.text + assert [row["session_total_spend"] for row in page["data"]] == pytest.approx( + [SESSION_SPEND] * len(ROUND_USAGE) + ), logs.text + + +def _completed_round(gateway: Gateway, alias: str, session_id: str, prompt: str) -> str: + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": alias, "messages": [{"role": "user", "content": prompt}], "litellm_trace_id": session_id}, + ) + assert response.status_code == 200, response.text + return str(response.json()["id"]) diff --git a/tests/integration/spend/test_spend_log_write_batching.py b/tests/integration/spend/test_spend_log_write_batching.py new file mode 100644 index 00000000000..386f99b1d07 --- /dev/null +++ b/tests/integration/spend/test_spend_log_write_batching.py @@ -0,0 +1,75 @@ +import uuid +from pathlib import Path +from typing import Final + +import pytest +import yaml +from integration._support.client import Gateway, eventually, string_value +from integration._support.database import read_rows +from integration._support.process import owned_proxy +from pydantic import JsonValue + +WRITE_STATEMENT_MAX_BYTES: Final = 200_000 +MESSAGES_PER_REQUEST: Final = 60 +MESSAGE_CHARACTERS: Final = 2_000 +REQUESTS: Final = 4 + + +def _config_storing_prompts(tmp_path: Path) -> Path: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["general_settings"]["store_prompts_in_spend_logs"] = True + path: Final = tmp_path / "store-prompts.yaml" + path.write_text(yaml.safe_dump(config)) + return path + + +def _prompt_messages(marker: str) -> list[JsonValue]: + return [ + {"role": "user", "content": f"{marker}-{index}-".ljust(MESSAGE_CHARACTERS, "x")} + for index in range(MESSAGES_PER_REQUEST) + ] + + +def _persisted(request_ids: tuple[str, ...]) -> list[dict[str, JsonValue]]: + placeholders: Final = ", ".join("%s" for _ in request_ids) + return read_rows( + "SELECT request_id, xmin::text AS statement, octet_length(proxy_server_request::text) AS stored_bytes " + f'FROM "LiteLLM_SpendLogs" WHERE request_id IN ({placeholders}) ORDER BY request_id', + request_ids, + ) + + +@pytest.mark.covers("quota_management.spend_tracking.prompt_rows_are_written_in_byte_bounded_statements") +def test_prompt_carrying_spend_rows_flushed_together_are_written_in_byte_bounded_statements( + gateway: Gateway, tmp_path: Path +) -> None: + with ( + gateway.scenario() as scenario, + owned_proxy( + gateway, + tmp_path, + { + "SPEND_LOG_WRITE_BATCH_MAX_BYTES": str(WRITE_STATEMENT_MAX_BYTES), + "SPEND_LOG_QUEUE_POLL_INTERVAL": "15", + }, + config=_config_storing_prompts(tmp_path), + ) as owned, + ): + model: Final = scenario.model() + request_ids: Final = tuple( + string_value( + owned.post( + "/v1/chat/completions", + {"model": model, "messages": _prompt_messages(f"integration-prompt-{uuid.uuid4().hex}")}, + )["id"] + ) + for _ in range(REQUESTS) + ) + assert len(set(request_ids)) == REQUESTS, request_ids + rows: Final = eventually(lambda: _persisted(request_ids), lambda values: len(values) == REQUESTS, seconds=70) + stored_bytes: Final = tuple(row["stored_bytes"] for row in rows) + assert all( + isinstance(size, int) and WRITE_STATEMENT_MAX_BYTES // 2 < size < WRITE_STATEMENT_MAX_BYTES + for size in stored_bytes + ), rows + assert len({row["statement"] for row in rows}) == REQUESTS, rows diff --git a/tests/integration/spend/test_team_member_spend.py b/tests/integration/spend/test_team_member_spend.py index 89eb2a24bcf..cf1dac25793 100644 --- a/tests/integration/spend/test_team_member_spend.py +++ b/tests/integration/spend/test_team_member_spend.py @@ -1,9 +1,12 @@ +import os import uuid from typing import Final +import httpx import pytest from integration._support.client import Gateway, eventually, object_value from integration._support.database import read_rows +from redis import Redis @pytest.mark.covers("spend.team_member.member_without_budget_gets_membership_row_and_spend") @@ -52,3 +55,45 @@ def test_member_added_without_any_budget_is_charged_on_its_membership_row(gatewa if object_value(row)["user_id"] == user ] assert len(exposed) == 1 and exposed[0][1] == pytest.approx(0.06), info + + +@pytest.mark.covers("spend.team_member.stale_low_redis_counter_still_blocks_member_over_budget") +def test_member_over_budget_is_blocked_when_redis_counter_reads_stale_low(gateway: Gateway) -> None: + with ( + gateway.scenario() as scenario, + httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream, + Redis(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"])) as cache, + ): + model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002) + team: Final = scenario.team(models=[model]) + user: Final = scenario.user() + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"user_id": user, "role": "user"}, "max_budget_in_team": 0.05}, + ) + assert added.status_code == 200, added.text + key: Final = scenario.key(team_id=team, user_id=user, models=[model]) + assert gateway.chat(model, key=key, text=f"member budget {uuid.uuid4().hex}")["usage"]["total_tokens"] == 40 + charged: Final = eventually( + lambda: read_rows( + 'SELECT spend FROM "LiteLLM_TeamMembership" WHERE team_id=%s AND user_id=%s', (team, user) + ), + lambda values: len(values) == 1 and float(values[0]["spend"]) >= 0.06, + seconds=70, + ) + assert float(charged[0]["spend"]) == pytest.approx(0.06) + counter_key: Final = f"spend:team_member:{user}:{team}" + counted: Final = eventually(lambda: cache.get(counter_key), lambda value: value is not None, seconds=10) + assert float(counted) == pytest.approx(0.06), counted + cache.set(counter_key, "0.01") + upstream.get("/__observations").raise_for_status() + denied: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": f"stale counter {uuid.uuid4().hex}"}]}, + key=key, + ) + assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text + assert upstream.get("/__observations").json()["requests"] == [] + assert float(cache.get(counter_key)) == pytest.approx(0.06), denied.text diff --git a/tests/integration/spend/test_user_budget_on_team_keys.py b/tests/integration/spend/test_user_budget_on_team_keys.py new file mode 100644 index 00000000000..f8e0eea3454 --- /dev/null +++ b/tests/integration/spend/test_user_budget_on_team_keys.py @@ -0,0 +1,54 @@ +import uuid +from hashlib import sha256 +from pathlib import Path +from typing import Final + +import httpx +import pytest +import yaml +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.process import owned_proxy + + +@pytest.mark.covers("spend.user_budget.opted_in_team_key_is_denied_once_owner_budget_is_exhausted") +def test_team_key_is_denied_before_provider_once_owner_personal_budget_is_exhausted_when_opted_in( + gateway: Gateway, tmp_path: Path +) -> None: + configuration: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + configuration["general_settings"]["apply_user_budget_to_team_keys"] = True + path: Final = tmp_path / "apply-user-budget-to-team-keys.yaml" + path.write_text(yaml.safe_dump(configuration)) + with ( + owned_proxy(gateway, tmp_path, {}, config=path) as candidate, + candidate.scenario() as scenario, + httpx.Client(base_url=candidate.upstream_url, timeout=5, trust_env=False) as upstream, + ): + model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002) + user: Final = scenario.user(max_budget=0.06) + team: Final = scenario.team(models=[model]) + added: Final = candidate.request( + "POST", "/team/member_add", {"team_id": team, "member": {"user_id": user, "role": "user"}} + ) + assert added.status_code == 200, added.text + key: Final = scenario.key(team_id=team, user_id=user, models=[model]) + first: Final = candidate.chat(model, key=key, text=f"owner budget {uuid.uuid4().hex}") + assert first["usage"]["total_tokens"] == 40 + spent: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_UserTable" WHERE user_id=%s', (user,)), + lambda values: len(values) == 1 and float(values[0]["spend"]) >= 0.06, + seconds=70, + ) + assert float(spent[0]["spend"]) == pytest.approx(0.06) + assert read_rows( + 'SELECT max_budget FROM "LiteLLM_VerificationToken" WHERE token=%s', (sha256(key.encode()).hexdigest(),) + ) == [{"max_budget": None}] + upstream.get("/__observations").raise_for_status() + denied: Final = candidate.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": f"over owner budget {uuid.uuid4().hex}"}]}, + key=key, + ) + assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text + assert upstream.get("/__observations").json()["requests"] == []