From 65468219dba1095d477655fbcbe3cd257be194d8 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 04:13:03 +0000 Subject: [PATCH] test(integration): regression tests for August cost tracking and budgeting bugs (#42622) * test(integration): saving echoed model_info never persists cost map pricing as a deployment override (Pylon #6870) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): budget_duration change on /budget/update recomputes budget_reset_at (Pylon #6913) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): count_tokens on a budgeted key reserves no budget and a later completion still succeeds (Pylon #6966) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): /cost/estimate reports configured prices for a deployment absent from the cost map (Pylon #7014) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): cache the team member default budget in Redis as JSON (Pylon #7180) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): aggregated team daily activity reports whole-range team spend in one page (Pylon #7224) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): failed daily user rollup commits are retried so spend report and daily activity agree (Pylon #7268) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): CLI session token without org_id is charged to and capped by the team organization budget (Pylon #7291) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): gemini passthrough success releases its budget reservation from the spend counter (Pylon #7295) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): batch retrieval spend row sums reasoning tokens and counts output and error file failures (Pylon #7341) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): uncostable batches retire from the cost poll page so newer batches are costed (Pylon #7342) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): charge a team member added without any budget on its membership row (Pylon #7363) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): failed dispatched requests keep estimated input tokens in spend logs (Pylon #7519) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): bedrock passthrough converse guardrail ignores tool definitions (Pylon #7524) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): explicit null budget_duration on /team/new is not replaced by default_team_params (Pylon #7536) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): PATCH /organization/update with a null limit clears it (Pylon #7577) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): ultrafast service_tier bills ultrafast rates without leaking pricing fields upstream (Pylon #7587) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): keep the selected model in the response and spend log for an Azure Model Router alias (Pylon #7636) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): disconnected Bedrock /v1/messages stream still bills terminal usage (Pylon #7685) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): databricks cached prompt tokens bill at cache rates (Pylon #7738) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): completed batch spend row records reasoning tokens and error file failures (Pylon #7928) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): bill OCR annotation pages at annotation_cost_per_page (Pylon #7958) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): in-flight count tokens request reserves no key budget so a completion still reaches the provider (Pylon #7307) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): fail-closed key rejects known estimate over remaining budget before provider (Pylon #7691) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): streamed /v1/responses success callbacks keep provider response headers (Pylon #7775) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * Revert "test(integration): fail-closed key rejects known estimate over remaining budget before provider (Pylon #7691)" This reverts commit 910348be7a4e81640d33b1c8667963778670d78c. * test(integration): reconcile contracts manifest for bundled regression tests Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): isolate proxy config writes in bundled regression tests Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): address review feedback on budget reset bounds and callback batch accumulation Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): avoid rebinding the cache identity accumulator Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): assert forwarded messages per cache identity call Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): make budget reset and team default tests deterministic 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/wire.py | 8 + tests/integration/contracts.json | 72 ++++++ .../management/test_budget_updates.py | 29 +++ .../test_organization_budget_clear.py | 59 +++++ .../test_team_budget_duration_defaults.py | 61 +++++ .../test_team_member_budget_cache.py | 40 ++++ .../observability/test_callback_delivery.py | 116 +++++++++- .../observability/test_guardrail_effects.py | 102 ++++++++- .../pricing/test_configured_prices.py | 72 +++++- .../pricing/test_databricks_cache_pricing.py | 89 ++++++++ .../pricing/test_ocr_page_pricing.py | 59 +++++ .../pricing/test_service_tier_pricing.py | 71 ++++++ tests/integration/proxy_config.yaml | 1 + .../spend/test_batch_completion_accounting.py | 202 +++++++++++++++++ .../spend/test_batch_observability.py | 201 +++++++++++++++++ .../spend/test_batch_poll_starvation.py | 212 ++++++++++++++++++ .../integration/spend/test_cache_and_quota.py | 139 ++++++++++-- .../spend/test_daily_rollup_retry.py | 182 +++++++++++++++ ...nnected_bedrock_messages_stream_billing.py | 123 ++++++++++ .../spend/test_failed_dispatch_tokens.py | 50 +++++ .../spend/test_model_router_selected_model.py | 74 ++++++ .../test_org_budget_cli_session_token.py | 80 +++++++ .../test_passthrough_budget_reservation.py | 110 +++++++++ .../test_team_daily_activity_aggregated.py | 81 +++++++ .../spend/test_team_member_spend.py | 54 +++++ 25 files changed, 2265 insertions(+), 22 deletions(-) create mode 100644 tests/integration/management/test_budget_updates.py create mode 100644 tests/integration/management/test_organization_budget_clear.py create mode 100644 tests/integration/management/test_team_budget_duration_defaults.py create mode 100644 tests/integration/management/test_team_member_budget_cache.py create mode 100644 tests/integration/pricing/test_databricks_cache_pricing.py create mode 100644 tests/integration/pricing/test_ocr_page_pricing.py create mode 100644 tests/integration/pricing/test_service_tier_pricing.py create mode 100644 tests/integration/spend/test_batch_completion_accounting.py create mode 100644 tests/integration/spend/test_batch_observability.py create mode 100644 tests/integration/spend/test_batch_poll_starvation.py create mode 100644 tests/integration/spend/test_daily_rollup_retry.py create mode 100644 tests/integration/spend/test_disconnected_bedrock_messages_stream_billing.py create mode 100644 tests/integration/spend/test_failed_dispatch_tokens.py create mode 100644 tests/integration/spend/test_model_router_selected_model.py create mode 100644 tests/integration/spend/test_org_budget_cli_session_token.py create mode 100644 tests/integration/spend/test_passthrough_budget_reservation.py create mode 100644 tests/integration/spend/test_team_daily_activity_aggregated.py create mode 100644 tests/integration/spend/test_team_member_spend.py diff --git a/tests/integration/_support/wire.py b/tests/integration/_support/wire.py index 0c6acfde96c..5052c34e021 100644 --- a/tests/integration/_support/wire.py +++ b/tests/integration/_support/wire.py @@ -2,11 +2,13 @@ from __future__ import annotations import ssl import threading +import time from collections.abc import Callable, Generator, Mapping from contextlib import contextmanager from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from queue import SimpleQueue +from types import MappingProxyType from typing import Final @@ -26,6 +28,8 @@ class Reply: chunks: tuple[bytes, ...] | None = None abort_after: int | None = None gate_after_first: threading.Event | None = None + pause_between_chunks: float = 0 + headers: Mapping[str, str] = MappingProxyType({}) @dataclass(frozen=True, slots=True) @@ -64,6 +68,8 @@ def wire_server(respond: Callable[[Request], Reply], tls: ssl.SSLContext | None reply = Reply(status=500) self.send_response(reply.status) self.send_header("content-type", reply.content_type) + for name, value in reply.headers.items(): + self.send_header(name, value) if reply.chunks is None: self.send_header("content-length", str(len(reply.body))) else: @@ -81,6 +87,8 @@ def wire_server(respond: Callable[[Request], Reply], tls: ssl.SSLContext | None self.wfile.flush() if index == 0 and reply.gate_after_first is not None: assert reply.gate_after_first.wait(timeout=5), "Stream barrier was never released" + if reply.pause_between_chunks and index + 1 < len(reply.chunks): + time.sleep(reply.pause_between_chunks) else: self.wfile.write(b"0\r\n\r\n") self.wfile.flush() diff --git a/tests/integration/contracts.json b/tests/integration/contracts.json index a4d9fedf9b2..c0e128edd5a 100644 --- a/tests/integration/contracts.json +++ b/tests/integration/contracts.json @@ -1789,6 +1789,78 @@ "tests/integration/mcp/test_mcp_lifecycle.py::test_same_url_server_grants_scope_discovery_and_direct_or_virtual_execution[bearer]": [ "other.mcp.permissions.same_url_servers_enforce_discovery_and_execution" ], + "tests/integration/management/test_budget_updates.py::test_shortening_budget_duration_moves_reset_at_onto_the_new_schedule": [ + "mgmt.budget.update.duration_change_recomputes_reset_at" + ], + "tests/integration/management/test_organization_budget_clear.py::test_patch_organization_update_with_null_tpm_limit_clears_it_and_keeps_sibling_limits": [ + "mgmt.organization.update.null_clears_budget_limit" + ], + "tests/integration/management/test_team_budget_duration_defaults.py::test_team_new_explicit_null_budget_duration_is_not_replaced_by_default": [ + "mgmt.team.new.explicit_null_budget_duration_overrides_default" + ], + "tests/integration/management/test_team_member_budget_cache.py::test_team_member_default_budget_lands_in_redis_after_first_member_call": [ + "mgmt.team_member_budget.default_budget_is_cached_in_redis_as_json" + ], + "tests/integration/observability/test_callback_delivery.py::test_streamed_responses_success_callback_carries_provider_apim_request_id": [ + "other.observability.callbacks.streamed_responses_events_carry_provider_response_headers" + ], + "tests/integration/observability/test_guardrail_effects.py::test_bedrock_passthrough_converse_guardrail_ignores_denied_term_in_tool_definition": [ + "other.observability.guardrails.bedrock_passthrough_converse_scans_only_caller_content" + ], + "tests/integration/pricing/test_configured_prices.py::test_cost_estimate_reports_configured_prices_for_model_absent_from_cost_map": [ + "quota_management.cost_estimate.configured_price.reported_for_model_absent_from_cost_map" + ], + "tests/integration/pricing/test_configured_prices.py::test_saving_echoed_model_info_does_not_freeze_cost_map_price_into_deployment": [ + "pricing.model_update.echoed_cost_map_price_is_not_persisted_as_override" + ], + "tests/integration/pricing/test_databricks_cache_pricing.py::test_databricks_cached_prompt_tokens_bill_at_cache_rates_not_input_rate": [ + "pricing.databricks.cached_prompt_tokens_bill_at_cache_rates" + ], + "tests/integration/pricing/test_ocr_page_pricing.py::test_ocr_annotation_pages_are_billed_at_annotation_cost_per_page": [ + "pricing.ocr.annotation_pages_billed_at_annotation_rate" + ], + "tests/integration/pricing/test_service_tier_pricing.py::test_ultrafast_service_tier_bills_ultrafast_rates_and_keeps_pricing_off_the_wire": [ + "quota_management.spend_tracking.service_tier_pricing.ultrafast_bills_ultrafast_rates" + ], + "tests/integration/spend/test_batch_completion_accounting.py::test_completed_batch_spend_row_records_reasoning_tokens_and_error_file_failures": [ + "quota_management.spend_tracking.batch_costs.reasoning_tokens_and_error_file_failures_recorded" + ], + "tests/integration/spend/test_batch_observability.py::test_batch_retrieval_row_sums_reasoning_tokens_and_counts_output_and_error_file_failures": [ + "spend.batches.retrieval_row_aggregates_reasoning_tokens_and_per_request_counts" + ], + "tests/integration/spend/test_batch_poll_starvation.py::test_batches_gone_at_provider_do_not_starve_a_newer_batch_out_of_cost_polling": [ + "quota_management.spend_tracking.batch_costs.uncostable_rows_retire_so_newer_batches_are_costed" + ], + "tests/integration/spend/test_cache_and_quota.py::test_in_flight_count_tokens_does_not_reserve_key_budget_away_from_a_completion": [ + "quota_management.budget.key.in_flight_count_tokens_reserves_nothing_so_completion_reaches_provider" + ], + "tests/integration/spend/test_cache_and_quota.py::test_repeated_count_tokens_on_budgeted_key_does_not_reserve_budget_or_block_later_completion": [ + "quota_management.budget.key.count_tokens_reserves_nothing_so_completion_within_budget_succeeds" + ], + "tests/integration/spend/test_daily_rollup_retry.py::test_failed_daily_user_rollup_commit_is_retried_so_spend_report_and_daily_activity_agree": [ + "spend.daily_rollup.failed_user_commit_is_retried_until_report_and_daily_activity_agree" + ], + "tests/integration/spend/test_disconnected_bedrock_messages_stream_billing.py::test_client_disconnect_mid_bedrock_messages_stream_still_bills_terminal_usage": [ + "spend.anthropic_messages_stream.client_disconnect_bills_terminal_bedrock_usage" + ], + "tests/integration/spend/test_failed_dispatch_tokens.py::test_provider_500_after_dispatch_records_estimated_prompt_tokens_on_failure_row": [ + "spend.failed_dispatch.failure_row_records_estimated_input_tokens" + ], + "tests/integration/spend/test_model_router_selected_model.py::test_model_router_alias_without_router_in_name_keeps_selected_model_in_response_and_spend_log": [ + "spend.model_router.selected_model_is_returned_and_persisted_for_plain_alias" + ], + "tests/integration/spend/test_org_budget_cli_session_token.py::test_cli_session_token_without_org_id_charges_and_caps_the_team_organization": [ + "quota_management.organization_budget.cli_session_token_without_org_id_charges_team_organization" + ], + "tests/integration/spend/test_passthrough_budget_reservation.py::test_repeated_gemini_passthrough_calls_stay_served_while_key_spend_is_below_max_budget": [ + "spend.budget_reservation.gemini_passthrough_success_releases_reservation_from_spend_counter" + ], + "tests/integration/spend/test_team_daily_activity_aggregated.py::test_aggregated_team_activity_reports_the_whole_range_team_spend_in_one_page": [ + "quota_management.spend_tracking.team_daily_activity_aggregated_reports_whole_range_team_spend" + ], + "tests/integration/spend/test_team_member_spend.py::test_member_added_without_any_budget_is_charged_on_its_membership_row": [ + "spend.team_member.member_without_budget_gets_membership_row_and_spend" + ], "tests/integration/management/test_vector_store_config_ownership.py::test_config_store_is_listed_beside_db_store_and_survives_listing": [ "mgmt.vector_store.list.keeps_config_store_beside_db_stores" ], diff --git a/tests/integration/management/test_budget_updates.py b/tests/integration/management/test_budget_updates.py new file mode 100644 index 00000000000..25d525bcaaa --- /dev/null +++ b/tests/integration/management/test_budget_updates.py @@ -0,0 +1,29 @@ +from datetime import datetime, timedelta, timezone +from typing import Final + +import pytest + +from tests.integration._support.client import Gateway, string_value +from tests.integration._support.database import read_rows + + +def _persisted_reset_at(budget_id: str) -> datetime: + rows: Final = read_rows( + 'SELECT budget_reset_at::text AS reset_at FROM "LiteLLM_BudgetTable" WHERE budget_id = %s', (budget_id,) + ) + assert len(rows) == 1, rows + reset_at: Final = datetime.fromisoformat(string_value(rows[0]["reset_at"])) + return reset_at if reset_at.tzinfo is not None else reset_at.replace(tzinfo=timezone.utc) + + +@pytest.mark.covers("mgmt.budget.update.duration_change_recomputes_reset_at") +def test_shortening_budget_duration_moves_reset_at_onto_the_new_schedule(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + budget_id: Final = scenario.budget(max_budget=10.0, budget_duration="10d") + ten_day_reset_at: Final = _persisted_reset_at(budget_id) + before: Final = datetime.now(timezone.utc) + response: Final = gateway.request("POST", "/budget/update", {"budget_id": budget_id, "budget_duration": "1d"}) + assert response.status_code == 200, response.text + updated: Final = _persisted_reset_at(budget_id) + assert updated < ten_day_reset_at, f"{updated} not before {ten_day_reset_at}" + assert before < updated <= before + timedelta(days=1, minutes=5), f"{updated} not within 1d of {before}" diff --git a/tests/integration/management/test_organization_budget_clear.py b/tests/integration/management/test_organization_budget_clear.py new file mode 100644 index 00000000000..8109dfd5a2d --- /dev/null +++ b/tests/integration/management/test_organization_budget_clear.py @@ -0,0 +1,59 @@ +import uuid +from typing import Final + +import pytest + +from tests.integration._support.client import Gateway, object_value, string_value +from tests.integration._support.database import read_rows + + +def _budget_rows(budget_id: str) -> list[dict[str, object]]: + return read_rows( + 'SELECT tpm_limit, rpm_limit, max_budget FROM "LiteLLM_BudgetTable" WHERE budget_id = %s', (budget_id,) + ) + + +@pytest.mark.covers("mgmt.organization.update.null_clears_budget_limit") +def test_patch_organization_update_with_null_tpm_limit_clears_it_and_keeps_sibling_limits(gateway: Gateway) -> None: + created: Final = gateway.post( + "/organization/new", + { + "organization_alias": f"integration-{uuid.uuid4().hex}", + "tpm_limit": 4000, + "rpm_limit": 40, + "max_budget": 12.5, + }, + ) + organization_id: Final = string_value(created["organization_id"]) + budget_id: Final = string_value(created["budget_id"]) + try: + assert _budget_rows(budget_id) == [{"tpm_limit": 4000, "rpm_limit": 40, "max_budget": 12.5}] + updated: Final = gateway.request( + "PATCH", "/organization/update", {"organization_id": organization_id, "tpm_limit": None} + ) + assert updated.status_code == 200, updated.text + updated_budget: Final = object_value(object_value(updated.json())["litellm_budget_table"]) + assert (updated_budget["tpm_limit"], updated_budget["rpm_limit"], updated_budget["max_budget"]) == ( + None, + 40, + 12.5, + ), updated.text + assert _budget_rows(budget_id) == [{"tpm_limit": None, "rpm_limit": 40, "max_budget": 12.5}] + info: Final = gateway.request("GET", "/organization/info", params={"organization_id": organization_id}) + assert info.status_code == 200, info.text + info_budget: Final = object_value(object_value(info.json())["litellm_budget_table"]) + assert (info_budget["tpm_limit"], info_budget["rpm_limit"], info_budget["max_budget"]) == ( + None, + 40, + 12.5, + ), info.text + finally: + deleted: Final = gateway.request("DELETE", "/organization/delete", {"organization_ids": [organization_id]}) + assert deleted.status_code == 200, deleted.text + gateway.post("/budget/delete", {"id": budget_id}) + assert ( + read_rows( + 'SELECT organization_id FROM "LiteLLM_OrganizationTable" WHERE organization_id = %s', (organization_id,) + ) + == [] + ) diff --git a/tests/integration/management/test_team_budget_duration_defaults.py b/tests/integration/management/test_team_budget_duration_defaults.py new file mode 100644 index 00000000000..fd459f6a7ef --- /dev/null +++ b/tests/integration/management/test_team_budget_duration_defaults.py @@ -0,0 +1,61 @@ +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, string_value +from tests.integration._support.database import read_rows +from tests.integration._support.process import owned_proxy + + +def _budget_row(team_id: str) -> dict[str, JsonValue]: + rows: Final = read_rows( + 'SELECT max_budget, budget_duration, budget_reset_at::text FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team_id,), + ) + assert len(rows) == 1, rows + return rows[0] + + +@pytest.mark.covers("mgmt.team.new.explicit_null_budget_duration_overrides_default") +def test_team_new_explicit_null_budget_duration_is_not_replaced_by_default(gateway: Gateway, tmp_path: Path) -> None: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["litellm_settings"]["default_team_params"] = {"budget_duration": "30d"} + path: Final = tmp_path / "team-defaults.yaml" + path.write_text(yaml.safe_dump(config)) + with ( + owned_proxy(gateway, tmp_path, {"STORE_MODEL_IN_DB": "False"}, config=path) as candidate, + candidate.scenario() as scenario, + ): + never_resetting: Final = candidate.request( + "POST", + "/team/new", + {"team_alias": f"integration-{uuid.uuid4().hex}", "max_budget": 500, "budget_duration": None}, + ) + assert never_resetting.status_code == 200, never_resetting.text + never_resetting_id: Final = string_value(never_resetting.json()["team_id"]) + scenario.cleanups.callback(scenario.delete_team, never_resetting_id) + assert never_resetting.json()["max_budget"] == 500.0, never_resetting.text + assert never_resetting.json()["budget_duration"] is None, never_resetting.text + assert never_resetting.json()["budget_reset_at"] is None, never_resetting.text + assert _budget_row(never_resetting_id) == { + "max_budget": 500.0, + "budget_duration": None, + "budget_reset_at": None, + } + + inheriting: Final = candidate.request( + "POST", "/team/new", {"team_alias": f"integration-{uuid.uuid4().hex}", "max_budget": 500} + ) + assert inheriting.status_code == 200, inheriting.text + inheriting_id: Final = string_value(inheriting.json()["team_id"]) + scenario.cleanups.callback(scenario.delete_team, inheriting_id) + assert inheriting.json()["budget_duration"] == "30d", inheriting.text + assert inheriting.json()["budget_reset_at"] is not None, inheriting.text + inheriting_row: Final = _budget_row(inheriting_id) + assert inheriting_row["max_budget"] == 500.0, inheriting_row + assert inheriting_row["budget_duration"] == "30d", inheriting_row + assert inheriting_row["budget_reset_at"] is not None, inheriting_row diff --git a/tests/integration/management/test_team_member_budget_cache.py b/tests/integration/management/test_team_member_budget_cache.py new file mode 100644 index 00000000000..9c4181915c0 --- /dev/null +++ b/tests/integration/management/test_team_member_budget_cache.py @@ -0,0 +1,40 @@ +import os +from typing import Final + +import pytest +from pydantic import JsonValue, TypeAdapter +from redis import Redis + +from tests.integration._support.client import Gateway, eventually, object_value, string_value +from tests.integration._support.database import read_rows + +_CACHED_BUDGET: Final = TypeAdapter(dict[str, JsonValue]) + + +@pytest.mark.covers("mgmt.team_member_budget.default_budget_is_cached_in_redis_as_json") +def test_team_member_default_budget_lands_in_redis_after_first_member_call(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + model: Final = scenario.model() + user: Final = scenario.user() + team: Final = scenario.team(team_member_budget=25) + key: Final = scenario.key(team_id=team, user_id=user, models=[model]) + teams: Final = read_rows('SELECT metadata FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,)) + assert len(teams) == 1, teams + budget_id: Final = string_value(object_value(teams[0]["metadata"])["team_member_budget_id"]) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": "member budget cache"}]}, + key=key, + ) + assert response.status_code == 200, response.text + with Redis(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"])) as cache: + cached: Final = eventually( + lambda: cache.get(f"team_member_default_budget:{budget_id}"), + lambda value: value is not None, + seconds=10, + ) + assert isinstance(cached, bytes), cached + budget: Final = _CACHED_BUDGET.validate_json(cached) + assert budget["budget_id"] == budget_id, cached + assert budget["max_budget"] == 25, cached diff --git a/tests/integration/observability/test_callback_delivery.py b/tests/integration/observability/test_callback_delivery.py index 7a5f420d2cc..543e60da4a6 100644 --- a/tests/integration/observability/test_callback_delivery.py +++ b/tests/integration/observability/test_callback_delivery.py @@ -102,7 +102,7 @@ def test_concurrent_success_and_failure_join_callbacks_and_rows_without_credenti responses: Final = tuple(pool.map(request, tags)) assert tuple(response.status_code for response in responses) == (200, 400, 200, 400) assert len(provider.drain()) == 4 - batches = [] + batches: Final[list[Request]] = [] # mutable-ok: drain() consumes the queue, later polls must keep earlier batches def delivered() -> tuple[dict, ...]: batches.extend(endpoint.drain()) @@ -135,7 +135,7 @@ def test_concurrent_success_and_failure_join_callbacks_and_rows_without_credenti assert "synthetic callback failure" in json.dumps(event["error_information"]) rows: Final = eventually( lambda identity=event["id"]: read_rows( - 'SELECT request_id, spend, prompt_tokens, completion_tokens, request_tags ' + "SELECT request_id, spend, prompt_tokens, completion_tokens, request_tags " 'FROM "LiteLLM_SpendLogs" WHERE request_id=%s', (identity,), ), @@ -156,6 +156,118 @@ def test_concurrent_success_and_failure_join_callbacks_and_rows_without_credenti assert event["prompt_tokens"] == event["completion_tokens"] == rows[0]["completion_tokens"] == 0 +def _responses_frames(identity: str, text: str) -> tuple[bytes, ...]: + output: Final = [ + { + "type": "message", + "id": f"msg_{identity}", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": text, "annotations": []}], + } + ] + completed: Final = { + "id": identity, + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-4o-mini", + "output": output, + "usage": { + "input_tokens": 11, + "output_tokens": 4, + "total_tokens": 15, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}, + }, + } + events: Final = ( + {"type": "response.created", "response": {**completed, "status": "in_progress", "output": [], "usage": None}}, + { + "type": "response.output_text.delta", + "item_id": f"msg_{identity}", + "output_index": 0, + "content_index": 0, + "delta": text, + }, + {"type": "response.completed", "response": completed}, + ) + return tuple(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode() for event in events) + + +@pytest.mark.covers("other.observability.callbacks.streamed_responses_events_carry_provider_response_headers") +def test_streamed_responses_success_callback_carries_provider_apim_request_id(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "resp_" + uuid.uuid4().hex + correlation: Final = "azure-correlation-" + marker + region: Final = "East US 2" + secret: Final = "synthetic-provider-secret-" + marker + sink_secret: Final = "synthetic-sink-secret-" + marker + + def upstream(request: Request) -> Reply: + assert request.target.endswith("/responses"), request.target + assert request.headers["authorization"] == f"Bearer {secret}" + assert json.loads(request.body) == { + "model": "gpt-4o-mini", + "input": "header control " + marker, + "stream": True, + }, request.body + return Reply( + content_type="text/event-stream", + chunks=_responses_frames(marker, "streamed control"), + headers={"apim-request-id": correlation, "x-ms-region": region}, + ) + + def sink(request: Request) -> Reply: + assert request.headers["authorization"] == f"Bearer {sink_secret}" + return Reply() + + with wire_server(upstream) as provider, wire_server(sink) as endpoint: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["litellm_settings"].update({"callbacks": ["generic_api"], "DEFAULT_FLUSH_INTERVAL_SECONDS": 1}) + path: Final = tmp_path / "callbacks.yaml" + path.write_text(yaml.safe_dump(config)) + with ( + owned_proxy( + gateway, + tmp_path, + { + "GENERIC_LOGGER_ENDPOINT": endpoint.url, + "GENERIC_LOGGER_HEADERS": f"Authorization=Bearer {sink_secret}", + }, + config=path, + ) as candidate, + candidate.scenario() as scenario, + ): + model: Final = scenario.model(api_base=provider.url + "/v1", api_key=secret) + response: Final = candidate.request( + "POST", "/v1/responses", {"model": model, "input": "header control " + marker, "stream": True} + ) + assert response.status_code == 200, response.text + assert f'"item_id":"msg_{marker}"' in response.text, response.text + assert '"type":"response.completed"' in response.text, response.text + assert len(provider.drain()) == 1 + batches: Final[list[Request]] = [] # mutable-ok: drain() consumes the queue, later polls must keep earlier batches + + def delivered() -> tuple[dict, ...]: + batches.extend(endpoint.drain()) + return tuple( + event for batch in batches for event in json.loads(batch.body) if event.get("model_group") == model + ) + + events: Final = eventually(delivered, lambda values: len(values) == 1, seconds=10) + assert (events[0]["status"], events[0]["stream"], events[0]["call_type"]) == ("success", True, "aresponses") + additional_headers: Final = events[0]["hidden_params"]["additional_headers"] or {} + provider_headers: Final = { + name: value + for name, value in additional_headers.items() + if name in ("llm_provider-apim-request-id", "llm_provider-x-ms-region") + } + assert provider_headers == { + "llm_provider-apim-request-id": correlation, + "llm_provider-x-ms-region": region, + }, json.dumps(events[0]["hidden_params"]) + + _RAISING_HOOK: Final = """ from litellm.integrations.custom_logger import CustomLogger diff --git a/tests/integration/observability/test_guardrail_effects.py b/tests/integration/observability/test_guardrail_effects.py index 5a79b619906..f96cae593dc 100644 --- a/tests/integration/observability/test_guardrail_effects.py +++ b/tests/integration/observability/test_guardrail_effects.py @@ -5,9 +5,7 @@ from typing import Final import pytest import yaml - -from integration._support.client import Gateway, eventually -from integration._support.database import read_rows +from integration._support.client import Gateway 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 @@ -146,6 +144,104 @@ def test_guardrail_denial_prevents_provider_and_preserves_allowed_control(gatewa assert len(policy.drain()) == 2 +@pytest.mark.covers("other.observability.guardrails.bedrock_passthrough_converse_scans_only_caller_content") +def test_bedrock_passthrough_converse_guardrail_ignores_denied_term_in_tool_definition( + gateway: Gateway, tmp_path: Path +) -> None: + identity: Final = "guardrail" + uuid.uuid4().hex + denied: Final = "synthetic denied marker" + allowed: Final = "synthetic allowed weather question" + access_key: Final = "AKIASYNTHETICPASSTHROUGH" + tool_config: Final = { + "tools": [ + { + "toolSpec": { + "name": "lookup_weather", + "description": f"Look up the forecast, never answer a {denied}", + "inputSchema": { + "json": { + "type": "object", + "properties": {"city": {"type": "string", "enum": [denied]}}, + "required": ["city"], + } + }, + } + } + ] + } + + def guardrail(request: Request) -> Reply: + assert request.target == "/beta/litellm_basic_guardrail_api" + texts: Final = json.loads(request.body)["texts"] + result: Final = ( + {"action": "BLOCKED", "blocked_reason": "synthetic policy denial"} + if any(denied in text for text in texts) + else {"action": "NONE"} + ) + return Reply(body=json.dumps(result).encode()) + + def runtime(request: Request) -> Reply: + assert request.target == "/model/anthropic.claude-3-haiku-20240307-v1:0/converse" + assert request.headers["authorization"].startswith(f"AWS4-HMAC-SHA256 Credential={access_key}/"), ( + request.headers + ) + return Reply( + body=json.dumps( + { + "output": {"message": {"role": "assistant", "content": [{"text": "sunny passthrough control"}]}}, + "stopReason": "end_turn", + "usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15}, + "metrics": {"latencyMs": 1}, + } + ).encode() + ) + + with wire_server(guardrail) as policy, wire_server(runtime) as bedrock, gateway.scenario() as scenario: + model: Final = scenario.model( + model="bedrock/anthropic.claude-3-haiku-20240307-v1:0", + api_key=None, + api_base=bedrock.url, + aws_access_key_id=access_key, + aws_secret_access_key="synthetic-secret", + aws_region_name="us-east-1", + ) + 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 / "bedrock-passthrough.yaml" + path.write_text(yaml.safe_dump(config)) + with owned_proxy(gateway, tmp_path, {}, config=path) as candidate: + route: Final = f"/bedrock/model/{model}/converse" + passed: Final = candidate.request( + "POST", + route, + {"messages": [{"role": "user", "content": [{"text": allowed}]}], "toolConfig": tool_config}, + ) + assert passed.status_code == 200, passed.text + assert passed.json()["output"]["message"]["content"] == [{"text": "sunny passthrough control"}] + forwarded: Final = bedrock.drain() + assert len(forwarded) == 1, "the runtime peer must see exactly the allowed request" + assert json.loads(forwarded[0].body)["toolConfig"] == tool_config + blocked: Final = candidate.request( + "POST", + route, + {"messages": [{"role": "user", "content": [{"text": denied}]}], "toolConfig": tool_config}, + ) + assert blocked.status_code == 400 and "synthetic policy denial" in blocked.text, blocked.text + assert bedrock.drain() == () + assert [json.loads(request.body)["texts"] for request in policy.drain()] == [[allowed], [denied]] + + @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 655d74c1402..e39833516c2 100644 --- a/tests/integration/pricing/test_configured_prices.py +++ b/tests/integration/pricing/test_configured_prices.py @@ -1,10 +1,12 @@ -from collections.abc import Iterator, Mapping -from typing import Final -from pathlib import Path +import json import uuid +from collections.abc import Iterator, Mapping +from pathlib import Path +from typing import Final import pytest import yaml +from pydantic import JsonValue from tests.integration._support.client import Gateway, eventually, object_value, string_value from tests.integration._support.database import read_rows @@ -28,6 +30,31 @@ def test_custom_price_is_reported_and_charged(gateway: Gateway) -> None: assert params["output_cost_per_token"] == 0.002 +@pytest.mark.covers("quota_management.cost_estimate.configured_price.reported_for_model_absent_from_cost_map") +def test_cost_estimate_reports_configured_prices_for_model_absent_from_cost_map(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"openai/integration-on-prem-{uuid.uuid4().hex}", + input_cost_per_token=0.003, + output_cost_per_token=0.007, + ) + response: Final = gateway.request( + "POST", + "/cost/estimate", + {"model": model, "input_tokens": 1000, "output_tokens": 500, "num_requests_per_day": 10}, + ) + assert response.status_code == 200, response.text + body: Final = object_value(response.json()) + assert body["input_cost_per_token"] == pytest.approx(0.003), response.text + assert body["output_cost_per_token"] == pytest.approx(0.007), response.text + assert body["input_cost_per_request"] == pytest.approx(1000 * 0.003), response.text + assert body["output_cost_per_request"] == pytest.approx(500 * 0.007), response.text + margin: Final = body["margin_cost_per_request"] + assert isinstance(margin, float), response.text + assert body["cost_per_request"] == pytest.approx(1000 * 0.003 + 500 * 0.007 + margin), response.text + assert body["daily_cost"] == pytest.approx(10 * (1000 * 0.003 + 500 * 0.007 + margin)), response.text + + @pytest.mark.covers("quota_management.spend_tracking.default_prices.survive_nullable_sibling_reload") def test_default_prices_survive_nullable_sibling_and_reload(gateway: Gateway) -> None: for registration_order in (("custom", "omitted", "nullable"), ("nullable", "omitted", "custom")): @@ -103,6 +130,45 @@ def test_default_prices_survive_nullable_sibling_and_reload(gateway: Gateway) -> assert float(rows[0]["spend"]) == pytest.approx(expected, rel=1e-6) +COST_MAP_DISPLAY_PRICING_KEYS: Final = frozenset( + { + "input_cost_per_token", + "output_cost_per_token", + "cache_read_input_token_cost", + "cache_creation_input_token_cost", + } +) + + +def persisted_model_info(identity: str) -> dict[str, JsonValue]: + rows: Final = read_rows('SELECT model_info FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (identity,)) + assert len(rows) == 1, f"Deployment {identity} has {len(rows)} rows" + stored: Final = rows[0]["model_info"] + return object_value(json.loads(stored) if isinstance(stored, str) else stored) + + +@pytest.mark.covers("pricing.model_update.echoed_cost_map_price_is_not_persisted_as_override") +def test_saving_echoed_model_info_does_not_freeze_cost_map_price_into_deployment(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + model: Final = scenario.model() + 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) + displayed: Final = object_value(target["model_info"]) + identity: Final = string_value(displayed["id"]) + assert isinstance(displayed["input_cost_per_token"], float), displayed + assert isinstance(displayed["output_cost_per_token"], float), displayed + fresh: Final = persisted_model_info(identity) + assert {key: value for key, value in fresh.items() if key in COST_MAP_DISPLAY_PRICING_KEYS} == {}, fresh + 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 COST_MAP_DISPLAY_PRICING_KEYS} == {}, 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_databricks_cache_pricing.py b/tests/integration/pricing/test_databricks_cache_pricing.py new file mode 100644 index 00000000000..b32076d1bf9 --- /dev/null +++ b/tests/integration/pricing/test_databricks_cache_pricing.py @@ -0,0 +1,89 @@ +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_PROMPT_TOKENS: Final = 1000 +CACHE_CREATION_TOKENS: Final = 2000 +CACHE_READ_TOKENS: Final = 8000 +PROMPT_TOKENS: Final = UNCACHED_PROMPT_TOKENS + CACHE_CREATION_TOKENS + CACHE_READ_TOKENS +COMPLETION_TOKENS: Final = 500 + + +def databricks_cached_response() -> JsonResponse: + return JsonResponse( + content_type="application/json", + body={ + "id": "chatcmpl-$REQUEST_ID", + "object": "chat.completion", + "created": 1700000000, + "model": "databricks-claude-integration", + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": "cached reply"}, "finish_reason": "stop"} + ], + "usage": { + "prompt_tokens": PROMPT_TOKENS, + "completion_tokens": COMPLETION_TOKENS, + "total_tokens": PROMPT_TOKENS + COMPLETION_TOKENS, + "cache_creation_input_tokens": CACHE_CREATION_TOKENS, + "cache_read_input_tokens": CACHE_READ_TOKENS, + }, + }, + ) + + +@pytest.mark.covers("pricing.databricks.cached_prompt_tokens_bill_at_cache_rates") +def test_databricks_cached_prompt_tokens_bill_at_cache_rates_not_input_rate(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + scenario_id: Final = f"databricks-cache-{uuid.uuid4().hex[:12]}" + handle: Final = register_scenario(scenario_id, databricks_cached_response()) + scenario.cleanups.callback(delete_scenario, handle) + model: Final = scenario.model( + model="databricks/databricks-claude-integration", + 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/chat/completions", {"model": model, "messages": [{"role": "user", "content": "cache control"}]} + ) + assert response.status_code == 200, response.text + expected_prompt_cost: Final = ( + UNCACHED_PROMPT_TOKENS * INPUT_RATE + + CACHE_CREATION_TOKENS * CACHE_CREATION_RATE + + CACHE_READ_TOKENS * CACHE_READ_RATE + ) + expected_completion_cost: Final = COMPLETION_TOKENS * OUTPUT_RATE + assert float(response.headers["x-litellm-response-cost"]) == pytest.approx( + expected_prompt_cost + expected_completion_cost, rel=1e-6 + ), response.text + 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"] == PROMPT_TOKENS + assert rows[0]["completion_tokens"] == COMPLETION_TOKENS + assert float(rows[0]["spend"]) == pytest.approx(expected_prompt_cost + expected_completion_cost, rel=1e-6) + 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 float(breakdown["input_cost"]) == pytest.approx(expected_prompt_cost, rel=1e-6) + assert float(breakdown["output_cost"]) == pytest.approx(expected_completion_cost, rel=1e-6) diff --git a/tests/integration/pricing/test_ocr_page_pricing.py b/tests/integration/pricing/test_ocr_page_pricing.py new file mode 100644 index 00000000000..65f94ea673e --- /dev/null +++ b/tests/integration/pricing/test_ocr_page_pricing.py @@ -0,0 +1,59 @@ +import uuid +from typing import Final + +import pytest + +from tests.integration._support.client import Gateway, eventually, 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 + + +@pytest.mark.covers("pricing.ocr.annotation_pages_billed_at_annotation_rate") +def test_ocr_annotation_pages_are_billed_at_annotation_cost_per_page(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + scenario_id: Final = f"ocr-annotation-{uuid.uuid4().hex[:12]}" + handle: Final = register_scenario( + scenario_id, + JsonResponse( + content_type="application/json", + body={ + "pages": [{"index": index, "markdown": f"page {index}"} for index in range(3)], + "model": "integration-ocr", + "document_annotation": '{"title": "annotated"}', + "usage_info": {"pages_processed": 3, "pages_processed_annotation": 2, "doc_size_bytes": 4096}, + }, + ), + ) + scenario.cleanups.callback(delete_scenario, handle) + model: Final = scenario.model( + model=f"mistral/integration-ocr-{scenario_id}", + api_base=f"{handle.api_base()}/v1", + ocr_cost_per_page=0.002, + annotation_cost_per_page=0.01, + ) + response: Final = gateway.request( + "POST", + "/v1/ocr", + { + "model": model, + "document": {"type": "document_url", "document_url": "https://example.com/annotated.pdf"}, + "document_annotation_format": {"type": "json_schema", "json_schema": {"name": "title"}}, + }, + ) + assert response.status_code == 200, response.text + assert response.json()["usage_info"] == { + "pages_processed": 3, + "pages_processed_annotation": 2, + "credits": None, + "doc_size_bytes": 4096, + }, response.text + expected: Final = 3 * 0.002 + 2 * 0.01 + assert float(response.headers["x-litellm-response-cost"]) == pytest.approx(expected), response.text + request_id: Final = string_value(response.headers["x-litellm-call-id"]) + rows: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_SpendLogs" WHERE request_id=%s', (request_id,)), + lambda values: len(values) == 1, + seconds=70, + ) + assert float(rows[0]["spend"]) == pytest.approx(expected) diff --git a/tests/integration/pricing/test_service_tier_pricing.py b/tests/integration/pricing/test_service_tier_pricing.py new file mode 100644 index 00000000000..e0d26392f7f --- /dev/null +++ b/tests/integration/pricing/test_service_tier_pricing.py @@ -0,0 +1,71 @@ +import json +from typing import Final + +import httpx +import pytest + +from tests.integration._support.client import JSON_OBJECT, Gateway, eventually, object_value, string_value +from tests.integration._support.database import read_rows + +STANDARD_INPUT_RATE: Final = 0.001 +STANDARD_OUTPUT_RATE: Final = 0.002 +ULTRAFAST_INPUT_RATE: Final = 0.01 +ULTRAFAST_OUTPUT_RATE: Final = 0.02 + + +def assert_chat_bills_rates( + gateway: Gateway, model: str, service_tier: str | None, input_rate: float, output_rate: float +) -> None: + with httpx.Client(base_url=gateway.upstream_url, trust_env=False) as upstream: + upstream.get("/__observations").raise_for_status() + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": f"service tier {service_tier} control"}], + **({} if service_tier is None else {"service_tier": service_tier}), + }, + ) + assert response.status_code == 200, response.text + expected: Final = 20 * input_rate + 20 * output_rate + assert float(response.headers["x-litellm-response-cost"]) == pytest.approx(expected, rel=1e-6), response.text + observations: Final = JSON_OBJECT.validate_json(upstream.get("/__observations").content)["requests"] + assert isinstance(observations, list) + assert len(observations) == 1 + body: Final = object_value(object_value(observations[0])["body"]) + assert body == { + "model": "gpt-4o-mini", + "messages": [{"role": "user", "content": f"service tier {service_tier} control"}], + **({} if service_tier is None else {"service_tier": service_tier}), + }, response.text + 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"] == 20 + assert rows[0]["completion_tokens"] == 20 + assert float(rows[0]["spend"]) == pytest.approx(expected, rel=1e-6) + 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 float(breakdown["input_cost"]) == pytest.approx(20 * input_rate, rel=1e-6) + assert float(breakdown["output_cost"]) == pytest.approx(20 * output_rate, rel=1e-6) + + +@pytest.mark.covers("quota_management.spend_tracking.service_tier_pricing.ultrafast_bills_ultrafast_rates") +def test_ultrafast_service_tier_bills_ultrafast_rates_and_keeps_pricing_off_the_wire(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + model: Final = scenario.model( + input_cost_per_token=STANDARD_INPUT_RATE, + output_cost_per_token=STANDARD_OUTPUT_RATE, + input_cost_per_token_ultrafast=ULTRAFAST_INPUT_RATE, + output_cost_per_token_ultrafast=ULTRAFAST_OUTPUT_RATE, + ) + assert_chat_bills_rates(gateway, model, "ultrafast", ULTRAFAST_INPUT_RATE, ULTRAFAST_OUTPUT_RATE) + assert_chat_bills_rates(gateway, model, None, STANDARD_INPUT_RATE, STANDARD_OUTPUT_RATE) diff --git a/tests/integration/proxy_config.yaml b/tests/integration/proxy_config.yaml index d05a0aed325..a3d29e42120 100644 --- a/tests/integration/proxy_config.yaml +++ b/tests/integration/proxy_config.yaml @@ -5,6 +5,7 @@ general_settings: store_model_in_db: true disable_spend_logs: false proxy_batch_write_at: 1 + proxy_batch_polling_interval: 1 litellm_settings: enable_redis_auth_cache: true cache: true diff --git a/tests/integration/spend/test_batch_completion_accounting.py b/tests/integration/spend/test_batch_completion_accounting.py new file mode 100644 index 00000000000..0cbeda934f6 --- /dev/null +++ b/tests/integration/spend/test_batch_completion_accounting.py @@ -0,0 +1,202 @@ +from __future__ import annotations + +import json +import uuid +from hashlib import sha256 +from typing import Final + +import pytest +from integration._support.client import JSON_OBJECT, Gateway, eventually, string_value +from integration._support.database import read_rows +from integration._support.upstream import delete_scenario, register_scenario +from integration.cost_calculation.cost_tracking_case import JsonResponse, RoutedResponse, TextResponse +from pydantic import JsonValue + +FIRST_LINE: Final = {"prompt_tokens": 10, "completion_tokens": 7, "reasoning_tokens": 4} +SECOND_LINE: Final = {"prompt_tokens": 5, "completion_tokens": 3, "reasoning_tokens": 2} +ERROR_FILE_LINES: Final = 2 + + +def _succeeded_line(index: int, model: str, prompt_tokens: int, completion_tokens: int, reasoning_tokens: int) -> str: + return json.dumps( + { + "id": f"batch_req_{index}", + "custom_id": f"r{index}", + "response": { + "status_code": 200, + "request_id": f"$REQUEST_ID-{index}", + "body": { + "id": f"chatcmpl-$REQUEST_ID-{index}", + "object": "chat.completion", + "model": model, + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"} + ], + "usage": { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + "completion_tokens_details": {"reasoning_tokens": reasoning_tokens}, + }, + }, + }, + "error": None, + }, + separators=(",", ":"), + ) + + +def _failed_line(index: int) -> str: + return json.dumps( + { + "id": f"batch_req_{index}", + "custom_id": f"r{index}", + "response": { + "status_code": 400, + "request_id": f"$REQUEST_ID-{index}", + "body": {"error": {"message": "rejected line", "type": "invalid_request_error", "code": "400"}}, + }, + "error": {"code": "bad_request", "message": "rejected line"}, + }, + separators=(",", ":"), + ) + + +def _batch_routes(model: str) -> RoutedResponse: + output_lines: Final = ( + _succeeded_line(1, model, **FIRST_LINE), + _succeeded_line(2, model, **SECOND_LINE), + _failed_line(3), + ) + error_lines: Final = tuple(_failed_line(index) for index in range(4, 4 + ERROR_FILE_LINES)) + completed: Final = { + "id": "batch-$REQUEST_ID", + "object": "batch", + "endpoint": "/v1/chat/completions", + "errors": None, + "input_file_id": "file-in-$REQUEST_ID", + "completion_window": "24h", + "status": "completed", + "output_file_id": "file-out-$REQUEST_ID", + "error_file_id": "file-err-$REQUEST_ID", + "created_at": 1, + "in_progress_at": 1, + "completed_at": 1, + "expires_at": 1, + "request_counts": {"total": 5, "completed": 2, "failed": 3}, + "metadata": None, + } + return RoutedResponse( + content_type="application/x-routed", + routes={ + "POST /files": JsonResponse( + content_type="application/json", + body={ + "id": "file-in-$REQUEST_ID", + "object": "file", + "purpose": "batch", + "bytes": 100, + "created_at": 1, + "filename": "in.jsonl", + "status": "processed", + }, + ), + "POST /batches": JsonResponse( + content_type="application/json", + body={**completed, "status": "validating", "output_file_id": None, "error_file_id": None}, + ), + "GET /batches/batch-$REQUEST_ID": JsonResponse(content_type="application/json", body=completed), + "GET /files/file-out-$REQUEST_ID/content": TextResponse( + content_type="application/jsonl", body="\n".join(output_lines) + "\n" + ), + "GET /files/file-err-$REQUEST_ID/content": TextResponse( + content_type="application/jsonl", body="\n".join(error_lines) + "\n" + ), + }, + ) + + +def _input_file(model: str) -> bytes: + return ( + "\n".join( + json.dumps( + { + "custom_id": f"r{index}", + "method": "POST", + "url": "/v1/chat/completions", + "body": {"model": model, "messages": [{"role": "user", "content": "batch accounting"}]}, + }, + separators=(",", ":"), + ) + for index in range(1, 6) + ) + + "\n" + ).encode() + + +def _metadata(value: object) -> dict[str, JsonValue]: + return JSON_OBJECT.validate_json(value) if isinstance(value, str) else JSON_OBJECT.validate_python(value) + + +@pytest.mark.covers("quota_management.spend_tracking.batch_costs.reasoning_tokens_and_error_file_failures_recorded") +def test_completed_batch_spend_row_records_reasoning_tokens_and_error_file_failures(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + key: Final = scenario.key() + scenario_id: Final = f"batch-accounting-{uuid.uuid4().hex[:12]}" + handle: Final = register_scenario(scenario_id, _batch_routes("gpt-4o-mini")) + scenario.cleanups.callback(delete_scenario, handle) + model: Final = scenario.model(api_base=handle.api_base()) + file_response: Final = gateway.request_multipart( + "/v1/files", + {"purpose": "batch", "model": model}, + {"file": ("in.jsonl", _input_file(model), "application/jsonl")}, + key=key, + ) + assert file_response.status_code == 200, file_response.text + batch_response: Final = gateway.request( + "POST", + "/v1/batches", + { + "input_file_id": string_value(JSON_OBJECT.validate_json(file_response.content)["id"]), + "endpoint": "/v1/chat/completions", + "completion_window": "24h", + "model": model, + }, + key=key, + ) + assert batch_response.status_code == 200, batch_response.text + batch_id: Final = string_value(JSON_OBJECT.validate_json(batch_response.content)["id"]) + retrieval: Final = gateway.request("GET", f"/v1/batches/{batch_id}", key=key) + assert retrieval.status_code == 200, retrieval.text + assert retrieval.json()["status"] == "completed", retrieval.text + rows: Final = eventually( + lambda: read_rows( + 'SELECT status, prompt_tokens, completion_tokens, metadata FROM "LiteLLM_SpendLogs" ' + "WHERE api_key=%s AND call_type='aretrieve_batch'", + (sha256(key.encode()).hexdigest(),), + ), + lambda values: len(values) == 1, + seconds=70, + ) + row: Final = rows[0] + metadata: Final = _metadata(row["metadata"]) + prompt_tokens: Final = FIRST_LINE["prompt_tokens"] + SECOND_LINE["prompt_tokens"] + completion_tokens: Final = FIRST_LINE["completion_tokens"] + SECOND_LINE["completion_tokens"] + reasoning_tokens: Final = FIRST_LINE["reasoning_tokens"] + SECOND_LINE["reasoning_tokens"] + assert row["status"] == "success", retrieval.text + assert (row["prompt_tokens"], row["completion_tokens"]) == (prompt_tokens, completion_tokens), retrieval.text + assert (metadata["batch_successful_requests"], metadata["batch_failed_requests"]) == ( + 2, + 1 + ERROR_FILE_LINES, + ), json.dumps(metadata) + usage: Final = JSON_OBJECT.validate_python(metadata["usage_object"]) + details: Final = JSON_OBJECT.validate_python(usage["completion_tokens_details"]) + assert (usage["prompt_tokens"], usage["completion_tokens"], usage["total_tokens"]) == ( + prompt_tokens, + completion_tokens, + prompt_tokens + completion_tokens, + ), json.dumps(metadata) + assert {name: value for name, value in details.items() if value is not None} == { + "reasoning_tokens": reasoning_tokens, + "text_tokens": completion_tokens - reasoning_tokens, + }, json.dumps(metadata) diff --git a/tests/integration/spend/test_batch_observability.py b/tests/integration/spend/test_batch_observability.py new file mode 100644 index 00000000000..ca9c26ff513 --- /dev/null +++ b/tests/integration/spend/test_batch_observability.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +import json +import uuid +from hashlib import sha256 +from typing import Final + +import pytest +from integration._support.client import JSON_OBJECT, Gateway, eventually, object_value, string_value +from integration._support.database import read_rows +from integration._support.upstream import delete_scenario, register_scenario +from integration.cost_calculation.cost_tracking_case import JsonResponse, RoutedResponse, TextResponse +from pydantic import JsonValue + +REASONING_TOKENS: Final = (30, 50) +PROMPT_TOKENS: Final = 10 +COMPLETION_TOKENS: Final = 100 +ERROR_FILE_FAILURES: Final = 2 + + +def _successful_line(index: int, reasoning_tokens: int) -> str: + return json.dumps( + { + "id": f"batch_req_{index}", + "custom_id": f"r{index}", + "response": { + "status_code": 200, + "request_id": f"$REQUEST_ID-{index}", + "body": { + "id": f"chatcmpl-$REQUEST_ID-{index}", + "object": "chat.completion", + "model": "gpt-4o-mini", + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"} + ], + "usage": { + "prompt_tokens": PROMPT_TOKENS, + "completion_tokens": COMPLETION_TOKENS, + "total_tokens": PROMPT_TOKENS + COMPLETION_TOKENS, + "completion_tokens_details": {"reasoning_tokens": reasoning_tokens}, + }, + }, + }, + "error": None, + }, + separators=(",", ":"), + ) + + +def _failed_line(index: int) -> str: + return json.dumps( + { + "id": f"batch_req_{index}", + "custom_id": f"r{index}", + "response": {"status_code": 400, "request_id": f"$REQUEST_ID-{index}", "body": {"error": "bad"}}, + "error": {"code": "bad_request", "message": "failed"}, + }, + separators=(",", ":"), + ) + + +def _batch(status: str, *, files_ready: bool) -> dict[str, JsonValue]: + return { + "id": "batch-$REQUEST_ID", + "object": "batch", + "endpoint": "/v1/chat/completions", + "errors": None, + "input_file_id": "file-in-$REQUEST_ID", + "completion_window": "24h", + "status": status, + "output_file_id": "file-out-$REQUEST_ID" if files_ready else None, + "error_file_id": "file-err-$REQUEST_ID" if files_ready else None, + "created_at": 1, + "in_progress_at": 1, + "completed_at": 1 if files_ready else None, + "expires_at": 1, + "request_counts": {"total": 5, "completed": 2, "failed": 3}, + "metadata": None, + } + + +def _provider_routes() -> RoutedResponse: + output_lines: Final = ( + _successful_line(1, REASONING_TOKENS[0]), + _failed_line(2), + _successful_line(3, REASONING_TOKENS[1]), + ) + error_lines: Final = tuple(_failed_line(index) for index in range(4, 4 + ERROR_FILE_FAILURES)) + return RoutedResponse( + content_type="application/x-routed", + routes={ + "POST /files": JsonResponse( + content_type="application/json", + body={ + "id": "file-in-$REQUEST_ID", + "object": "file", + "purpose": "batch", + "bytes": 100, + "created_at": 1, + "filename": "in.jsonl", + "status": "processed", + }, + ), + "POST /batches": JsonResponse( + content_type="application/json", body=_batch("validating", files_ready=False) + ), + "GET /batches/batch-$REQUEST_ID": JsonResponse( + content_type="application/json", body=_batch("completed", files_ready=True) + ), + "GET /files/file-out-$REQUEST_ID/content": TextResponse( + content_type="application/jsonl", body="\n".join(output_lines) + "\n" + ), + "GET /files/file-err-$REQUEST_ID/content": TextResponse( + content_type="application/jsonl", body="\n".join(error_lines) + "\n" + ), + }, + ) + + +def _input_file(model_name: str) -> bytes: + return ( + "\n".join( + json.dumps( + { + "custom_id": f"r{index}", + "method": "POST", + "url": "/v1/chat/completions", + "body": {"model": model_name, "messages": [{"role": "user", "content": "batch observability"}]}, + }, + separators=(",", ":"), + ) + for index in range(1, 6) + ) + + "\n" + ).encode() + + +def _retrieval_rows(key: str) -> tuple[dict[str, JsonValue], ...]: + return tuple( + read_rows( + 'SELECT prompt_tokens, completion_tokens, metadata FROM "LiteLLM_SpendLogs" ' + "WHERE api_key=%s AND call_type='aretrieve_batch'", + (sha256(key.encode()).hexdigest(),), + ) + ) + + +def _metadata(row: dict[str, JsonValue]) -> dict[str, JsonValue]: + value: Final = row["metadata"] + return object_value(JSON_OBJECT.validate_json(value) if isinstance(value, str) else value) + + +@pytest.mark.covers("spend.batches.retrieval_row_aggregates_reasoning_tokens_and_per_request_counts") +def test_batch_retrieval_row_sums_reasoning_tokens_and_counts_output_and_error_file_failures( + gateway: Gateway, +) -> None: + with gateway.scenario() as scenario: + scenario_id: Final = f"batch-observability-{uuid.uuid4().hex[:12]}" + handle: Final = register_scenario(scenario_id, _provider_routes()) + scenario.cleanups.callback(delete_scenario, handle) + model_name: Final = scenario.model(api_base=handle.api_base()) + key: Final = scenario.key(models=[model_name]) + file_response: Final = gateway.request_multipart( + "/v1/files", + {"purpose": "batch", "model": model_name}, + {"file": ("in.jsonl", _input_file(model_name), "application/jsonl")}, + key=key, + ) + assert file_response.status_code == 200, file_response.text + input_file_id: Final = string_value(JSON_OBJECT.validate_json(file_response.content)["id"]) + batch_response: Final = gateway.request( + "POST", + "/v1/batches", + { + "input_file_id": input_file_id, + "endpoint": "/v1/chat/completions", + "completion_window": "24h", + "model": model_name, + }, + key=key, + ) + assert batch_response.status_code == 200, batch_response.text + batch_id: Final = string_value(JSON_OBJECT.validate_json(batch_response.content)["id"]) + retrieval: Final = eventually( + lambda: gateway.request("GET", f"/v1/batches/{batch_id}", key=key), + lambda response: response.status_code == 200 and response.json()["status"] == "completed", + seconds=30, + ) + assert retrieval.status_code == 200, retrieval.text + rows: Final = eventually(lambda: _retrieval_rows(key), lambda values: len(values) == 1, seconds=70) + row: Final = rows[0] + metadata: Final = _metadata(row) + usage: Final = object_value(metadata["usage_object"]) + assert row["prompt_tokens"] == 2 * PROMPT_TOKENS, retrieval.text + assert row["completion_tokens"] == 2 * COMPLETION_TOKENS, retrieval.text + assert object_value(usage["completion_tokens_details"])["reasoning_tokens"] == sum(REASONING_TOKENS), ( + retrieval.text, + usage, + ) + assert metadata["batch_successful_requests"] == 2, (retrieval.text, metadata) + assert metadata["batch_failed_requests"] == 1 + ERROR_FILE_FAILURES, (retrieval.text, metadata) diff --git a/tests/integration/spend/test_batch_poll_starvation.py b/tests/integration/spend/test_batch_poll_starvation.py new file mode 100644 index 00000000000..f68f10e3d8a --- /dev/null +++ b/tests/integration/spend/test_batch_poll_starvation.py @@ -0,0 +1,212 @@ +import json +import os +from hashlib import sha256 +from typing import Final + +import pytest +from integration._support.client import JSON_OBJECT, Gateway, Scenario, eventually, object_value, string_value +from integration._support.database import read_rows +from integration._support.upstream import delete_scenario, register_scenario +from integration.cost_calculation.cost_tracking_case import JsonResponse, RoutedResponse, TextResponse +from pydantic import JsonValue + +from litellm.constants import MAX_OBJECTS_PER_POLL_CYCLE + +INPUT_COST_PER_TOKEN: Final = 0.001 +OUTPUT_COST_PER_TOKEN: Final = 0.002 +PROMPT_TOKENS: Final = 100 +COMPLETION_TOKENS: Final = 50 +BATCH_COST_SHARE: Final = 0.5 + +_INPUT_FILE: Final = JsonResponse( + content_type="application/json", + body={ + "id": "file-in-$REQUEST_ID", + "object": "file", + "purpose": "batch", + "bytes": 100, + "created_at": 1, + "filename": "in.jsonl", + "status": "processed", + }, +) + + +def _batch(status: str, output_file_id: str | None) -> dict[str, JsonValue]: + return { + "id": "batch-$REQUEST_ID", + "object": "batch", + "endpoint": "/v1/chat/completions", + "errors": None, + "input_file_id": "file-in-$REQUEST_ID", + "completion_window": "24h", + "status": status, + "output_file_id": output_file_id, + "error_file_id": None, + "created_at": 1, + "in_progress_at": 1, + "completed_at": 1 if status == "completed" else None, + "expires_at": 1, + "request_counts": {"total": 1, "completed": 1 if status == "completed" else 0, "failed": 0}, + "metadata": None, + } + + +def _accepting_routes() -> dict[str, JsonResponse | TextResponse]: + return { + "POST /files": _INPUT_FILE, + "POST /batches": JsonResponse(content_type="application/json", body=_batch("validating", None)), + } + + +def _gone_at_provider_routes() -> RoutedResponse: + return RoutedResponse( + content_type="application/x-routed", + routes={ + **_accepting_routes(), + "GET /batches/batch-$REQUEST_ID": JsonResponse( + content_type="application/json", + status=404, + body={ + "error": { + "message": "No batch found with id 'batch-$REQUEST_ID'.", + "type": "invalid_request_error", + "param": "id", + "code": "batch_not_found", + } + }, + ), + }, + ) + + +def _completed_routes() -> RoutedResponse: + output_line: Final = { + "id": "batch_req_1", + "custom_id": "r1", + "response": { + "status_code": 200, + "request_id": "$REQUEST_ID-1", + "body": { + "id": "chatcmpl-$REQUEST_ID-1", + "object": "chat.completion", + "model": "gpt-4o-mini", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], + "usage": { + "prompt_tokens": PROMPT_TOKENS, + "completion_tokens": COMPLETION_TOKENS, + "total_tokens": PROMPT_TOKENS + COMPLETION_TOKENS, + }, + }, + }, + "error": None, + } + return RoutedResponse( + content_type="application/x-routed", + routes={ + **_accepting_routes(), + "GET /batches/batch-$REQUEST_ID": JsonResponse( + content_type="application/json", body=_batch("completed", "file-out-$REQUEST_ID") + ), + "GET /files/file-out-$REQUEST_ID/content": TextResponse( + content_type="application/jsonl", body=json.dumps(output_line, separators=(",", ":")) + "\n" + ), + }, + ) + + +def _scripted_deployment(scenario: Scenario, marker: str, routes: RoutedResponse) -> str: + scenario_id: Final = f"poll-{marker}-{sha256(os.urandom(16)).hexdigest()[:12]}" + handle: Final = register_scenario(scenario_id, routes) + scenario.cleanups.callback(delete_scenario, handle) + created: Final = scenario.gateway.post( + "/model/new", + { + "model_name": f"poll-{marker}-{sha256(scenario_id.encode()).hexdigest()[:12]}", + "litellm_params": { + "model": "openai/gpt-4o-mini", + "api_key": "sk-scripted-provider", + "api_base": handle.api_base(), + "input_cost_per_token": INPUT_COST_PER_TOKEN, + "output_cost_per_token": OUTPUT_COST_PER_TOKEN, + }, + }, + ) + scenario.cleanups.callback(scenario.delete_model, string_value(object_value(created["model_info"])["id"])) + return string_value(created["model_name"]) + + +def _submitted_batch_id(gateway: Gateway, key: str, model_name: str) -> str: + request_line: Final = { + "custom_id": "r1", + "method": "POST", + "url": "/v1/chat/completions", + "body": {"model": model_name, "messages": [{"role": "user", "content": "poll starvation"}]}, + } + file_response: Final = gateway.request_multipart( + "/v1/files", + {"purpose": "batch", "target_model_names": model_name}, + {"file": ("in.jsonl", (json.dumps(request_line) + "\n").encode(), "application/jsonl")}, + key=key, + ) + assert file_response.is_success, file_response.text + batch_response: Final = gateway.request( + "POST", + "/v1/batches", + { + "input_file_id": string_value(JSON_OBJECT.validate_json(file_response.content)["id"]), + "endpoint": "/v1/chat/completions", + "completion_window": "24h", + "model": model_name, + }, + key=key, + ) + assert batch_response.is_success, batch_response.text + return string_value(JSON_OBJECT.validate_json(batch_response.content)["id"]) + + +def _managed_rows(batch_ids: tuple[str, ...]) -> list[dict[str, JsonValue]]: + placeholders: Final = ", ".join("%s" for _ in batch_ids) + return read_rows( + f'SELECT batch_processed FROM "LiteLLM_ManagedObjectTable" WHERE unified_object_id IN ({placeholders})', + batch_ids, + ) + + +@pytest.mark.timeout(180) +@pytest.mark.covers("quota_management.spend_tracking.batch_costs.uncostable_rows_retire_so_newer_batches_are_costed") +def test_batches_gone_at_provider_do_not_starve_a_newer_batch_out_of_cost_polling(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + key: Final = scenario.key() + gone_batch_ids: Final = tuple( + _submitted_batch_id( + gateway, key, _scripted_deployment(scenario, f"gone{index}", _gone_at_provider_routes()) + ) + for index in range(MAX_OBJECTS_PER_POLL_CYCLE) + ) + costable_batch_id: Final = _submitted_batch_id( + gateway, key, _scripted_deployment(scenario, "costable", _completed_routes()) + ) + spend_rows: Final = eventually( + lambda: read_rows( + 'SELECT call_type, status, prompt_tokens, completion_tokens, spend FROM "LiteLLM_SpendLogs" ' + "WHERE api_key = %s AND call_type = %s", + (sha256(key.encode()).hexdigest(), "aretrieve_batch"), + ), + lambda rows: len(rows) == 1, + seconds=120, + ) + assert spend_rows == [ + { + "call_type": "aretrieve_batch", + "status": "success", + "prompt_tokens": PROMPT_TOKENS, + "completion_tokens": COMPLETION_TOKENS, + "spend": pytest.approx( + BATCH_COST_SHARE + * (PROMPT_TOKENS * INPUT_COST_PER_TOKEN + COMPLETION_TOKENS * OUTPUT_COST_PER_TOKEN) + ), + } + ] + assert _managed_rows((costable_batch_id,)) == [{"batch_processed": True}] + assert _managed_rows(gone_batch_ids) == [{"batch_processed": True}] * MAX_OBJECTS_PER_POLL_CYCLE diff --git a/tests/integration/spend/test_cache_and_quota.py b/tests/integration/spend/test_cache_and_quota.py index d32297765f6..97ef785aa15 100644 --- a/tests/integration/spend/test_cache_and_quota.py +++ b/tests/integration/spend/test_cache_and_quota.py @@ -1,4 +1,7 @@ +import json +import threading import uuid +from concurrent.futures import ThreadPoolExecutor from contextlib import ExitStack from hashlib import sha256 from typing import Final @@ -7,10 +10,10 @@ import httpx 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.database import read_rows from integration._support.generation import LIFECYCLE_SETTINGS, bounded_http_requests +from integration._support.wire import Reply, Request, wire_server @pytest.mark.covers("quota_management.response_cache.generated_sequences_preserve_content_and_accounting") @@ -211,6 +214,117 @@ def test_key_budget_at_boundary_blocks_provider_then_explicit_reset_restores(gat assert upstream.get("/__observations").json()["requests"] == [] +@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, +) -> None: + with ( + gateway.scenario() as scenario, + httpx.Client(base_url=gateway.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) + key: Final = scenario.key(models=[model], max_budget=0.1) + digest: Final = sha256(key.encode()).hexdigest() + upstream.get("/__observations").raise_for_status() + counts: Final = tuple( + gateway.request( + "POST", + "/v1/messages/count_tokens", + {"model": model, "messages": [{"role": "user", "content": "hello!!!"}]}, + key=key, + headers={"anthropic-version": "2023-06-01"}, + ) + for _ in range(3) + ) + for count in counts: + assert count.status_code == 200, count.text + assert count.json() == counts[0].json(), count.text + input_tokens: Final = counts[0].json()["input_tokens"] + assert isinstance(input_tokens, int) and input_tokens > 0, counts[0].text + assert upstream.get("/__observations").json()["requests"] == [] + completion: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": f"after counting {uuid.uuid4().hex}"}]}, + key=key, + ) + assert completion.status_code == 200, completion.text + assert completion.json()["usage"]["total_tokens"] == 40, completion.text + assert [request["path"] for request in upstream.get("/__observations").json()["requests"]] == [ + "/v1/chat/completions" + ] + spent: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_VerificationToken" WHERE token=%s', (digest,)), + lambda values: len(values) == 1 and float(values[0]["spend"]) > 0, + seconds=70, + ) + assert float(spent[0]["spend"]) == pytest.approx(20 * 0.001 + 20 * 0.002) + rows: Final = eventually( + lambda: read_rows('SELECT call_type, spend FROM "LiteLLM_SpendLogs" WHERE api_key=%s', (digest,)), + lambda values: len(values) >= 1, + seconds=70, + ) + assert [(row["call_type"], float(row["spend"])) for row in rows] == [("acompletion", pytest.approx(0.06))] + + +@pytest.mark.covers( + "quota_management.budget.key.in_flight_count_tokens_reserves_nothing_so_completion_reaches_provider" +) +def test_in_flight_count_tokens_does_not_reserve_key_budget_away_from_a_completion(gateway: Gateway) -> None: + counting_reached_provider: Final = threading.Event() + completion_answered: Final = threading.Event() + + def respond(request: Request) -> Reply: + counting_reached_provider.set() + assert completion_answered.wait(timeout=30), "completion never ran while count tokens was in flight" + return Reply(body=b'{"totalTokens": 12, "promptTokensDetails": [{"modality": "TEXT", "tokenCount": 12}]}') + + with ( + wire_server(respond) as wire, + gateway.scenario() as scenario, + httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream, + ThreadPoolExecutor(max_workers=1) as background, + ): + counted: Final = scenario.model( + model="gemini/gemini-3.8-flash", + api_base=wire.url, + api_key="synthetic-gemini-key", + input_cost_per_token=0.001, + output_cost_per_token=0.002, + ) + completed: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002) + key: Final = scenario.key(models=[counted, completed], max_budget=0.06) + contents: Final = [{"role": "user", "parts": [{"text": "hello"}]}] + counting: Final = background.submit( + gateway.request, "POST", f"/v1beta/models/{counted}:countTokens", {"contents": contents}, key=key + ) + assert counting_reached_provider.wait(timeout=30), "count tokens request never reached the provider" + upstream.get("/__observations").raise_for_status() + prompt: Final = f"after count tokens {uuid.uuid4().hex}" + completion: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": completed, "messages": [{"role": "user", "content": prompt}]}, + key=key, + ) + completion_answered.set() + count: Final = counting.result(timeout=30) + assert completion.status_code == 200 and completion.json()["usage"]["total_tokens"] == 40, completion.text + assert [call["body"]["messages"] for call in upstream.get("/__observations").json()["requests"]] == [ + [{"role": "user", "content": prompt}] + ] + assert count.status_code == 200, count.text + assert count.json() == {"totalTokens": 12, "promptTokensDetails": [{"modality": "TEXT", "tokenCount": 12}]}, ( + count.text + ) + provider_calls: Final = wire.drain() + assert [(call.method, call.target) for call in provider_calls] == [ + ("POST", "/v1beta/models/gemini-3.8-flash:countTokens") + ] + assert provider_calls[0].headers["x-goog-api-key"] == "synthetic-gemini-key" + assert json.loads(provider_calls[0].body) == {"contents": contents} + + @pytest.mark.covers("quota_management.response_cache.system_messages_partition_cache_identity") def test_different_system_messages_do_not_share_a_cached_response(gateway: Gateway) -> None: with ( @@ -219,8 +333,7 @@ def test_different_system_messages_do_not_share_a_cached_response(gateway: Gatew ): model: Final = scenario.model() prompt: Final = uuid.uuid4().hex - identities: dict[str, str] = {} - for system, expected_calls in (("first policy", 1), ("second policy", 1), ("first policy", 0)): + def completion_id(system: str, expected_calls: int) -> str: upstream.get("/__observations").raise_for_status() response: Final = gateway.request( "POST", @@ -232,14 +345,12 @@ def test_different_system_messages_do_not_share_a_cached_response(gateway: Gatew ) assert response.status_code == 200 and response.json()["usage"]["total_tokens"] == 40, response.text calls: Final = upstream.get("/__observations").json()["requests"] - assert len(calls) == expected_calls - if system in identities: - assert response.json()["id"] == identities[system] - else: - assert response.json()["id"] not in identities.values() - identities = {**identities, system: response.json()["id"]} - if calls: - assert calls[0]["body"]["messages"] == [ - {"role": "system", "content": system}, - {"role": "user", "content": prompt}, - ] + assert [call["body"]["messages"] for call in calls] == [ + [{"role": "system", "content": system}, {"role": "user", "content": prompt}] + ] * expected_calls, calls + return response.json()["id"] + + first_policy_id: Final = completion_id("first policy", 1) + second_policy_id: Final = completion_id("second policy", 1) + assert first_policy_id != second_policy_id + assert completion_id("first policy", 0) == first_policy_id diff --git a/tests/integration/spend/test_daily_rollup_retry.py b/tests/integration/spend/test_daily_rollup_retry.py new file mode 100644 index 00000000000..cf1b989639a --- /dev/null +++ b/tests/integration/spend/test_daily_rollup_retry.py @@ -0,0 +1,182 @@ +import json +import os +import uuid +from collections.abc import Iterable +from hashlib import sha256 +from typing import Final + +import psycopg +import pytest +from integration._support.client import Gateway, delete_key_if_present, eventually, string_value +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server +from psycopg import sql + + +def _execute(statements: Iterable[sql.Composable]) -> None: + with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection: + for statement in statements: + connection.execute(statement) + + +def _install_daily_user_rollup_fault(user_id: str) -> str: + suffix: Final = f"fault-{uuid.uuid4().hex}" + sequence: Final = sql.Identifier(f"{suffix}_attempts") + function: Final = sql.Identifier(suffix) + _execute( + ( + sql.SQL("CREATE SEQUENCE {}").format(sequence), + sql.SQL( + "CREATE FUNCTION {}() RETURNS trigger LANGUAGE plpgsql AS $fault$ " + "BEGIN PERFORM nextval({}); " + "RAISE EXCEPTION 'synthetic daily rollup outage' USING ERRCODE = '55P03'; " + "END $fault$" + ).format(function, sql.Literal(f"{suffix}_attempts")), + sql.SQL( + 'CREATE TRIGGER {} BEFORE INSERT ON "LiteLLM_DailyUserSpend" ' + "FOR EACH ROW WHEN (NEW.user_id = {}) EXECUTE FUNCTION {}()" + ).format(sql.Identifier(suffix), sql.Literal(user_id), function), + ) + ) + return suffix + + +def _lift_daily_user_rollup_fault(suffix: str) -> None: + _execute( + ( + sql.SQL('DROP TRIGGER IF EXISTS {} ON "LiteLLM_DailyUserSpend"').format(sql.Identifier(suffix)), + sql.SQL("DROP FUNCTION IF EXISTS {}()").format(sql.Identifier(suffix)), + sql.SQL("DROP SEQUENCE IF EXISTS {}").format(sql.Identifier(f"{suffix}_attempts")), + ) + ) + + +def _rollup_attempts(suffix: str) -> int: + with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection: + row: Final = connection.execute( + sql.SQL("SELECT CASE WHEN is_called THEN last_value ELSE 0 END FROM {}").format( + sql.Identifier(f"{suffix}_attempts") + ) + ).fetchone() + assert row is not None + return int(row[0]) + + +@pytest.mark.covers("spend.daily_rollup.failed_user_commit_is_retried_until_report_and_daily_activity_agree") +def test_failed_daily_user_rollup_commit_is_retried_so_spend_report_and_daily_activity_agree( + gateway: Gateway, +) -> None: + def provider(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/v1/chat/completions" + 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": "synthetic rollup answer"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 20, "completion_tokens": 20, "total_tokens": 40}, + } + ).encode() + ) + + with wire_server(provider) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + api_base=wire.url + "/v1", input_cost_per_token=0.001, output_cost_per_token=0.002, num_retries=0 + ) + user: Final = scenario.user() + key: Final = string_value(gateway.post("/key/generate", {"user_id": user, "models": [model]})["key"]) + scenario.cleanups.callback(delete_key_if_present, gateway, key) + digest: Final = sha256(key.encode()).hexdigest() + suffix: Final = _install_daily_user_rollup_fault(user) + scenario.cleanups.callback(_lift_daily_user_rollup_fault, suffix) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": "rollup retry control"}]}, + key=key, + ) + assert response.status_code == 200, response.text + assert float(response.headers["x-litellm-response-cost"]) == pytest.approx(20 * 0.001 + 20 * 0.002) + body: Final = response.json() + spend_rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, DATE("startTime")::text AS day, model FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (body["id"],), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert float(spend_rows[0]["spend"]) == pytest.approx(0.06) + day: Final = string_value(spend_rows[0]["day"]) + stored_model: Final = string_value(spend_rows[0]["model"]) + eventually(lambda: _rollup_attempts(suffix), lambda attempts: attempts >= 1, seconds=70) + _lift_daily_user_rollup_fault(suffix) + activity: Final = eventually( + lambda: gateway.request( + "GET", + "/user/daily/activity/aggregated", + params={"start_date": day, "end_date": day, "api_key": digest}, + ), + lambda polled: ( + polled.status_code == 200 + and len(polled.json().get("results", ())) > 0 + and polled.json()["results"][0]["breakdown"]["api_keys"] + .get(digest, {}) + .get("metrics", {}) + .get("spend", 0) + == pytest.approx(0.06) + ), + seconds=90, + ) + assert activity.status_code == 200, activity.text + metrics: Final = activity.json()["results"][0]["breakdown"]["api_keys"][digest]["metrics"] + assert metrics == { + "spend": pytest.approx(0.06), + "flat_cost": pytest.approx(0.0), + "prompt_tokens": 20, + "completion_tokens": 20, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "compression_saved_tokens": 0, + "compression_savings_spend": pytest.approx(0.0), + "prompt_caching_savings_spend": pytest.approx(0.0), + "gateway_injected_caching_savings_spend": pytest.approx(0.0), + "autorouter_savings_spend": pytest.approx(0.0), + "total_tokens": 40, + "successful_requests": 1, + "failed_requests": 0, + "api_requests": 1, + "total_response_time_ms": metrics["total_response_time_ms"], + "timed_requests": metrics["timed_requests"], + } + report: Final = gateway.request( + "GET", + "/global/spend/report", + params={"start_date": day, "end_date": day, "api_key": digest}, + ) + assert report.status_code == 200, report.text + assert report.json() == [ + { + "api_key": digest, + "total_cost": pytest.approx(0.06), + "total_input_tokens": 20, + "total_output_tokens": 20, + "model_details": [ + { + "model": stored_model, + "total_cost": pytest.approx(0.06), + "total_input_tokens": 20, + "total_output_tokens": 20, + } + ], + } + ] + assert metrics["spend"] == pytest.approx(report.json()[0]["total_cost"]) diff --git a/tests/integration/spend/test_disconnected_bedrock_messages_stream_billing.py b/tests/integration/spend/test_disconnected_bedrock_messages_stream_billing.py new file mode 100644 index 00000000000..91f2c1c0760 --- /dev/null +++ b/tests/integration/spend/test_disconnected_bedrock_messages_stream_billing.py @@ -0,0 +1,123 @@ +import base64 +import json +import uuid +from hashlib import sha256 +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.upstream import _aws_event_frame +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + +BEDROCK_MODEL: Final = "anthropic.claude-haiku-4-5-20251001-v1:0" +INPUT_TOKENS: Final = 30 +FULL_OUTPUT_TOKENS: Final = 412 +INPUT_RATE: Final = 0.001 +OUTPUT_RATE: Final = 0.002 + + +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 _message_start(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, "output_tokens": 0}, + }, + } + ) + _invoke_chunk({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}) + + +def _text_delta(text: str) -> bytes: + return _invoke_chunk({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}}) + + +def _terminal_usage() -> bytes: + return ( + _invoke_chunk({"type": "content_block_stop", "index": 0}) + + _invoke_chunk( + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn"}, + "usage": {"output_tokens": FULL_OUTPUT_TOKENS}, + } + ) + + _invoke_chunk({"type": "message_stop"}) + ) + + +@pytest.mark.covers("spend.anthropic_messages_stream.client_disconnect_bills_terminal_bedrock_usage") +@pytest.mark.timeout(120) +def test_client_disconnect_mid_bedrock_messages_stream_still_bills_terminal_usage(gateway: Gateway) -> None: + message_id: Final = f"msg_{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": "disconnect control"}], request.body + return Reply( + content_type="application/vnd.amazon.eventstream", + chunks=( + _message_start(message_id) + _text_delta("first"), + _text_delta("second"), + _text_delta("third"), + _terminal_usage(), + ), + pause_between_chunks=0.5, + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"bedrock/invoke/{BEDROCK_MODEL}", + api_base=wire.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, + ) + key: Final = scenario.key(models=[model]) + with gateway.client.stream( + "POST", + "/v1/messages", + json={ + "model": model, + "messages": [{"role": "user", "content": "disconnect control"}], + "max_tokens": FULL_OUTPUT_TOKENS, + "stream": True, + }, + headers={"Authorization": f"Bearer {key}"}, + ) as response: + assert response.status_code == 200, response.read().decode() + first_event: Final = next(line for line in response.iter_lines() if line.startswith("data:")) + assert json.loads(first_event.removeprefix("data:"))["type"] == "message_start", first_event + + rows: Final = eventually( + lambda: read_rows( + 'SELECT request_id, status, prompt_tokens, completion_tokens, spend FROM "LiteLLM_SpendLogs" ' + "WHERE api_key=%s", + (sha256(key.encode()).hexdigest(),), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["request_id"] == message_id, rows + assert rows[0]["status"] == "success", rows + assert rows[0]["prompt_tokens"] == INPUT_TOKENS, rows + assert rows[0]["completion_tokens"] == FULL_OUTPUT_TOKENS, rows + assert float(str(rows[0]["spend"])) == pytest.approx( + INPUT_TOKENS * INPUT_RATE + FULL_OUTPUT_TOKENS * OUTPUT_RATE + ), rows + assert len(wire.drain()) == 1 diff --git a/tests/integration/spend/test_failed_dispatch_tokens.py b/tests/integration/spend/test_failed_dispatch_tokens.py new file mode 100644 index 00000000000..5778ff9654d --- /dev/null +++ b/tests/integration/spend/test_failed_dispatch_tokens.py @@ -0,0 +1,50 @@ +import json +import uuid +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 + + +@pytest.mark.covers("spend.failed_dispatch.failure_row_records_estimated_input_tokens") +def test_provider_500_after_dispatch_records_estimated_prompt_tokens_on_failure_row(gateway: Gateway) -> None: + prompt: Final = "failed dispatch accounting " + uuid.uuid4().hex + system: Final = "You are a terse accounting assistant" + + def provider(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/v1/chat/completions" + body: Final = json.loads(request.body) + assert body["model"] == "gpt-4o-mini" + assert body["messages"] == [{"role": "system", "content": system}, {"role": "user", "content": prompt}] + return Reply( + status=500, + body=b'{"error":{"message":"synthetic provider outage","type":"server_error","code":"500"}}', + ) + + with wire_server(provider) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(api_base=wire.url + "/v1", num_retries=0) + key: Final = scenario.key(models=[model]) + failed: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "system", "content": system}, {"role": "user", "content": prompt}]}, + key=key, + ) + assert failed.status_code == 500 and "synthetic provider outage" in failed.text, failed.text + call_id: Final = failed.headers["x-litellm-call-id"] + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT status, spend, prompt_tokens, completion_tokens, total_tokens FROM "LiteLLM_SpendLogs" ' + "WHERE request_id=%s", + (call_id,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + row: Final = rows[0] + assert row["status"] == "failure" and float(row["spend"]) == 0 and row["completion_tokens"] == 0, row + assert row["prompt_tokens"] > 0, f"failure row lost the dispatched input tokens: {row}" + assert row["total_tokens"] == row["prompt_tokens"], row diff --git a/tests/integration/spend/test_model_router_selected_model.py b/tests/integration/spend/test_model_router_selected_model.py new file mode 100644 index 00000000000..a95ee4f9feb --- /dev/null +++ b/tests/integration/spend/test_model_router_selected_model.py @@ -0,0 +1,74 @@ +import json +import uuid +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 + +ROUTER_DEPLOYMENT: Final = "router-deploy" +SELECTED_MODEL: Final = "grok-4-1-fast-reasoning" +SELECTED_MODEL_WITH_PROVIDER: Final = f"azure_ai/{SELECTED_MODEL}" + + +@pytest.mark.covers("spend.model_router.selected_model_is_returned_and_persisted_for_plain_alias") +def test_model_router_alias_without_router_in_name_keeps_selected_model_in_response_and_spend_log( + gateway: Gateway, +) -> None: + prompt: Final = uuid.uuid4().hex + + def provider(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/chat/completions", request.target + assert json.loads(request.body) == { + "model": ROUTER_DEPLOYMENT, + "messages": [{"role": "user", "content": prompt}], + "stream": False, + }, request.body + return Reply( + body=json.dumps( + { + "id": "chatcmpl-" + uuid.uuid4().hex, + "object": "chat.completion", + "created": 1, + "model": SELECTED_MODEL, + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "routed answer"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 20, "completion_tokens": 20, "total_tokens": 40}, + } + ).encode() + ) + + with wire_server(provider) as wire, gateway.scenario() as scenario: + alias: Final = scenario.model( + model=f"azure_ai/model_router/{ROUTER_DEPLOYMENT}", api_base=wire.url, num_retries=0 + ) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": alias, "messages": [{"role": "user", "content": prompt}]}, + ) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["model"] == SELECTED_MODEL_WITH_PROVIDER, response.text + assert body["choices"][0]["message"]["content"] == "routed answer", response.text + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT model, model_group, status FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (body["id"],), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows == [{"model": SELECTED_MODEL_WITH_PROVIDER, "model_group": alias, "status": "success"}] + logs: Final = gateway.request("GET", "/spend/logs", params={"request_id": body["id"]}) + assert logs.status_code == 200, logs.text + assert [(row["model"], row["model_group"]) for row in logs.json()] == [(SELECTED_MODEL_WITH_PROVIDER, alias)], ( + logs.text + ) diff --git a/tests/integration/spend/test_org_budget_cli_session_token.py b/tests/integration/spend/test_org_budget_cli_session_token.py new file mode 100644 index 00000000000..f821e337374 --- /dev/null +++ b/tests/integration/spend/test_org_budget_cli_session_token.py @@ -0,0 +1,80 @@ +import os +import uuid +from typing import Final + +import httpx +import pytest +from integration._support.client import Gateway, eventually, string_value +from integration._support.database import read_rows + +from litellm.proxy._types import LiteLLM_UserTable +from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken + + +def _cli_session_token(user_id: str, team_id: str) -> str: + cli_user: Final = LiteLLM_UserTable(user_id=user_id, user_role="internal_user", teams=[team_id], models=[]) + return ExperimentalUIJWTToken.get_cli_jwt_auth_token(user_info=cli_user, team_id=team_id, team_alias="cli-team") + + +@pytest.mark.covers("quota_management.organization_budget.cli_session_token_without_org_id_charges_team_organization") +def test_cli_session_token_without_org_id_charges_and_caps_the_team_organization( + gateway: Gateway, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("LITELLM_SALT_KEY", os.environ.get("LITELLM_SALT_KEY", "sk-integration-salt")) + with ( + gateway.scenario() as scenario, + httpx.Client(base_url=gateway.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) + organization: Final = gateway.post( + "/organization/new", {"organization_alias": f"integration-{uuid.uuid4().hex}", "max_budget": 0.06} + ) + org_id: Final = string_value(organization["organization_id"]) + scenario.cleanups.callback( + lambda: gateway.request("DELETE", "/organization/delete", {"organization_ids": [org_id]}) + ) + user_id: Final = scenario.user() + team_id: Final = scenario.team( + organization_id=org_id, models=[model], members_with_roles=[{"role": "user", "user_id": user_id}] + ) + token: Final = _cli_session_token(user_id, team_id) + prompt: Final = f"org budget {uuid.uuid4().hex}" + upstream.get("/__observations").raise_for_status() + first: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": prompt}]}, + key=token, + ) + assert first.status_code == 200 and first.json()["usage"]["total_tokens"] == 40, first.text + reached_upstream: Final = upstream.get("/__observations").json()["requests"] + assert len(reached_upstream) == 1, reached_upstream + assert reached_upstream[0]["body"]["model"] == "gpt-4o-mini", reached_upstream + assert reached_upstream[0]["body"]["messages"] == [{"role": "user", "content": prompt}], reached_upstream + logged: Final = eventually( + lambda: read_rows( + 'SELECT organization_id, team_id, spend FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (first.json()["id"],), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert [(row["organization_id"], row["team_id"], float(row["spend"])) for row in logged] == [ + (org_id, team_id, pytest.approx(0.06)) + ] + charged: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_OrganizationTable" WHERE organization_id=%s', (org_id,)), + lambda values: len(values) == 1 and float(values[0]["spend"]) >= 0.06, + seconds=70, + ) + assert float(charged[0]["spend"]) == pytest.approx(0.06) + assert float(gateway.get("/organization/info", {"organization_id": org_id})["spend"]) == pytest.approx(0.06) + denied: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": f"over org budget {uuid.uuid4().hex}"}]}, + key=token, + ) + assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text + assert f"Organization={org_id}" in denied.json()["error"]["message"], denied.text + assert upstream.get("/__observations").json()["requests"] == [] diff --git a/tests/integration/spend/test_passthrough_budget_reservation.py b/tests/integration/spend/test_passthrough_budget_reservation.py new file mode 100644 index 00000000000..f072333c616 --- /dev/null +++ b/tests/integration/spend/test_passthrough_budget_reservation.py @@ -0,0 +1,110 @@ +import uuid +from hashlib import sha256 +from pathlib import Path +from typing import Final + +import pytest +import yaml +from integration._support.client import Gateway, eventually, object_value, string_value +from integration._support.database import read_rows +from integration._support.process import owned_proxy +from integration._support.upstream import delete_scenario, register_scenario +from integration.cost_calculation.cost_tracking_case import JsonResponse +from pydantic import JsonValue + +INPUT_COST_PER_TOKEN: Final = 0.000001 +OUTPUT_COST_PER_TOKEN: Final = 0.001 +PROMPT_TOKENS: Final = 10 +CANDIDATE_TOKENS: Final = 5 +COST_PER_CALL: Final = PROMPT_TOKENS * INPUT_COST_PER_TOKEN + CANDIDATE_TOKENS * OUTPUT_COST_PER_TOKEN +MAX_BUDGET: Final = 0.02 +CALLS_WITHIN_BUDGET: Final = 4 + + +def _key_spend(digest: str) -> float: + rows: Final = read_rows('SELECT spend FROM "LiteLLM_VerificationToken" WHERE token=%s', (digest,)) + assert len(rows) == 1, rows + return float(rows[0]["spend"]) + + +def _generate_content_request(model: str) -> dict[str, JsonValue]: + return {"contents": [{"role": "user", "parts": [{"text": f"budget {model}"}]}]} + + +def _generate_content_response(model: str) -> JsonResponse: + return JsonResponse( + content_type="application/json", + body={ + "candidates": [ + { + "content": {"parts": [{"text": f"scripted answer {model}"}], "role": "model"}, + "finishReason": "STOP", + "index": 0, + } + ], + "usageMetadata": { + "promptTokenCount": PROMPT_TOKENS, + "candidatesTokenCount": CANDIDATE_TOKENS, + "totalTokenCount": PROMPT_TOKENS + CANDIDATE_TOKENS, + }, + "modelVersion": model, + }, + ) + + +def _served_call(gateway: Gateway, model: str, key: str, scenario_id: str, call: int) -> None: + digest: Final = sha256(key.encode()).hexdigest() + spend_before: Final = _key_spend(digest) + assert spend_before == pytest.approx((call - 1) * COST_PER_CALL) and spend_before < MAX_BUDGET + response: Final = gateway.request( + "POST", + f"/gemini/v1beta/models/{model}:generateContent", + _generate_content_request(model), + headers={"x-goog-api-key": key, "x-pass-x-scripted-scenario": scenario_id}, + ) + assert response.status_code == 200, f"call {call} with key spend {spend_before}: {response.text}" + assert response.json() == _generate_content_response(model).body, response.text + eventually(lambda: _key_spend(digest), lambda spend: spend >= call * COST_PER_CALL - 1e-9, seconds=70) + + +@pytest.mark.covers("spend.budget_reservation.gemini_passthrough_success_releases_reservation_from_spend_counter") +def test_repeated_gemini_passthrough_calls_stay_served_while_key_spend_is_below_max_budget( + gateway: Gateway, tmp_path: Path +) -> None: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["environment_variables"] = { + "GEMINI_API_BASE": gateway.upstream_url, + "GEMINI_API_KEY": "scripted", + } + path: Final = tmp_path / "gemini-passthrough.yaml" + path.write_text(yaml.safe_dump(config)) + with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario: + model: Final = f"gemini-passthrough-{uuid.uuid4().hex}" + created: Final = candidate.post( + "/model/new", + { + "model_name": model, + "litellm_params": { + "model": "gemini/gemini-2.5-flash", + "api_key": "scripted", + "api_base": gateway.upstream_url, + "input_cost_per_token": INPUT_COST_PER_TOKEN, + "output_cost_per_token": OUTPUT_COST_PER_TOKEN, + }, + "model_info": {"id": model, "max_output_tokens": 10}, + }, + ) + scenario.cleanups.callback(scenario.delete_model, string_value(object_value(created["model_info"])["id"])) + handle: Final = register_scenario(f"sc-{model}", _generate_content_response(model)) + scenario.cleanups.callback(delete_scenario, handle) + key: Final = scenario.key(models=[model], max_budget=MAX_BUDGET) + for call in range(1, CALLS_WITHIN_BUDGET + 1): + _served_call(candidate, model, key, handle.scenario_id, call) + assert _key_spend(sha256(key.encode()).hexdigest()) == pytest.approx(CALLS_WITHIN_BUDGET * COST_PER_CALL) + denied: Final = candidate.request( + "POST", + f"/gemini/v1beta/models/{model}:generateContent", + _generate_content_request(model), + headers={"x-goog-api-key": key, "x-pass-x-scripted-scenario": handle.scenario_id}, + ) + assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text diff --git a/tests/integration/spend/test_team_daily_activity_aggregated.py b/tests/integration/spend/test_team_daily_activity_aggregated.py new file mode 100644 index 00000000000..53c2f2efb1d --- /dev/null +++ b/tests/integration/spend/test_team_daily_activity_aggregated.py @@ -0,0 +1,81 @@ +import uuid +from datetime import datetime, timedelta, timezone +from hashlib import sha256 +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually, object_value +from integration._support.database import read_rows + + +@pytest.mark.covers("quota_management.spend_tracking.team_daily_activity_aggregated_reports_whole_range_team_spend") +def test_aggregated_team_activity_reports_the_whole_range_team_spend_in_one_page(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002) + team: Final = scenario.team(models=[model]) + keys: Final = tuple(scenario.key(team_id=team, models=[model]) for _ in range(2)) + digests: Final = tuple(sha256(key.encode()).hexdigest() for key in keys) + for key in keys: + for _ in range(2): + reply: Final = gateway.chat(model, key=key, text=f"team activity {uuid.uuid4().hex}") + assert reply["usage"]["total_tokens"] == 40, reply + logged: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_SpendLogs" WHERE team_id=%s', (team,)), + lambda values: len(values) == 4, + seconds=70, + ) + assert sum(float(row["spend"]) for row in logged) == pytest.approx(0.24) + daily: Final = eventually( + lambda: read_rows( + 'SELECT api_key, spend, successful_requests FROM "LiteLLM_DailyTeamSpend" WHERE team_id=%s', (team,) + ), + lambda values: sum(float(row["spend"]) for row in values) >= 0.24 - 1e-9, + seconds=70, + ) + assert sorted(row["api_key"] for row in daily) == sorted(digests), daily + assert all(float(row["spend"]) == pytest.approx(0.12) and row["successful_requests"] == 2 for row in daily) + today: Final = datetime.now(timezone.utc) + response: Final = gateway.request( + "GET", + "/team/daily/activity/aggregated", + params={ + "team_ids": team, + "start_date": (today - timedelta(days=1)).strftime("%Y-%m-%d"), + "end_date": (today + timedelta(days=1)).strftime("%Y-%m-%d"), + "timezone": "0", + }, + ) + assert response.status_code == 200, response.text + body: Final = object_value(response.json()) + metadata: Final = object_value(body["metadata"]) + assert ( + metadata["total_spend"], + metadata["total_prompt_tokens"], + metadata["total_completion_tokens"], + metadata["total_tokens"], + metadata["total_api_requests"], + metadata["total_successful_requests"], + metadata["total_failed_requests"], + metadata["page"], + metadata["total_pages"], + metadata["has_more"], + ) == (pytest.approx(0.24), 80, 80, 160, 4, 4, 0, 1, 1, False), response.text + results: Final = body["results"] + assert isinstance(results, list) and len(results) == 1, response.text + day: Final = object_value(results[0]) + assert object_value(day["metrics"])["spend"] == pytest.approx(0.24), response.text + entities: Final = object_value(object_value(day["breakdown"])["entities"]) + assert set(entities) == {team}, response.text + team_bucket: Final = object_value(entities[team]) + team_metrics: Final = object_value(team_bucket["metrics"]) + assert (team_metrics["spend"], team_metrics["api_requests"], team_metrics["successful_requests"]) == ( + pytest.approx(0.24), + 4, + 4, + ), response.text + per_key: Final = object_value(team_bucket["api_key_breakdown"]) + assert set(per_key) == set(digests), response.text + assert tuple(object_value(object_value(per_key[digest])["metrics"])["spend"] for digest in digests) == ( + pytest.approx(0.12), + pytest.approx(0.12), + ), response.text diff --git a/tests/integration/spend/test_team_member_spend.py b/tests/integration/spend/test_team_member_spend.py new file mode 100644 index 00000000000..89eb2a24bcf --- /dev/null +++ b/tests/integration/spend/test_team_member_spend.py @@ -0,0 +1,54 @@ +import uuid +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually, object_value +from integration._support.database import read_rows + + +@pytest.mark.covers("spend.team_member.member_without_budget_gets_membership_row_and_spend") +def test_member_added_without_any_budget_is_charged_on_its_membership_row(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + 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"}} + ) + assert added.status_code == 200, added.text + memberships: Final = added.json()["updated_team_memberships"] + assert [ + {"user_id": row["user_id"], "team_id": row["team_id"], "budget_id": row["budget_id"], "spend": row["spend"]} + for row in memberships + ] == [{"user_id": user, "team_id": team, "budget_id": None, "spend": 0}], added.text + assert read_rows( + 'SELECT budget_id, spend, total_spend FROM "LiteLLM_TeamMembership" WHERE team_id=%s AND user_id=%s', + (team, user), + ) == [{"budget_id": None, "spend": 0.0, "total_spend": 0.0}] + key: Final = scenario.key(team_id=team, user_id=user, models=[model]) + assert gateway.chat(model, key=key, text=f"member spend {uuid.uuid4().hex}")["usage"]["total_tokens"] == 40 + charged: Final = eventually( + lambda: read_rows( + 'SELECT spend, total_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) + assert float(charged[0]["total_spend"]) == pytest.approx(0.06) + team_rows: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_TeamTable" WHERE team_id=%s', (team,)), + lambda values: len(values) == 1 and float(values[0]["spend"]) >= 0.06, + seconds=70, + ) + assert float(team_rows[0]["spend"]) == pytest.approx(0.06) + info: Final = gateway.get("/team/info", {"team_id": team}) + listed: Final = info["team_memberships"] + assert isinstance(listed, list) + exposed: Final = [ + (object_value(row)["user_id"], object_value(row)["spend"]) + for row in listed + if object_value(row)["user_id"] == user + ] + assert len(exposed) == 1 and exposed[0][1] == pytest.approx(0.06), info