scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-21 00:21:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 74

fix(proxy): enforce tag budgets for tags a pre-call hook adds

Browse source 
Tag max budgets were only checked in auth against the tags in the request
body. A custom guardrail that sets metadata.tags in its pre-call hook runs
after auth, so the tag it added was charged in spend logs but never blocked.
Snapshot the request tags before pre_call_hook (once, so fallback retries
keep the original baseline), then budget-check the tags the hook added
after the proxy_server_request snapshot is refreshed, reading both
litellm_metadata and metadata since spend attribution reads both. The check
runs only on the routes auth budget-checks and skips zero-cost models like
auth does. The raised BudgetExceededError carries the resolved llm_provider
so the failure record matches the auth-path one.
This commit is contained in:
Yucheng He 2026-09-15 01:35:19 -07:00
parent 3ed6c19b8d
commit 0333ca1f34
4 changed files with 456 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

61
litellm/proxy/auth/auth_checks.py
View file

@ -76,6 +76,7 @@ from litellm.proxy.common_utils.cache_pydantic_utils import CacheCodec
from litellm.proxy.common_utils.http_parsing_utils import (
_safe_get_request_headers,
_safe_get_request_query_params,
get_tags_from_request_body,
)
from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
from litellm.proxy.common_utils.user_api_key_cache import (
@ -102,6 +103,7 @@ from litellm.proxy.guardrails.tool_name_extraction import (
TOOL_CAPABLE_CALL_TYPES,
extract_request_tool_names,
)
from litellm.proxy.hooks.rate_limiter_utils import resolve_llm_provider_for_rate_limit
from litellm.proxy.route_llm_request import route_request
from litellm.proxy.spend_tracking.budget_reservation import get_budget_window_start
from litellm.proxy.spend_tracking.carried_budget_state import carry_organization_budget_state
@ -851,6 +853,13 @@ BUDGET_ENFORCED_SIDE_EFFECT_ROUTES: Final = frozenset(
)
def route_skips_budget_checks(route: str) -> bool:
"""Budget checks only guard LLM API routes and the health routes that spend money."""
return route not in BUDGET_ENFORCED_SIDE_EFFECT_ROUTES and (
route in MODEL_DISCOVERY_ROUTES or not RouteChecks.is_llm_api_route(route=route)
)
async def common_checks(
request_body: dict,
team_object: LiteLLM_TeamTable | None,
@ -898,10 +907,7 @@ async def common_checks(
team_id=valid_token.team_id if valid_token is not None else None,
)
skip_all_budget_checks: Final = skip_budget_checks or (
route not in BUDGET_ENFORCED_SIDE_EFFECT_ROUTES
and (route in MODEL_DISCOVERY_ROUTES or not RouteChecks.is_llm_api_route(route=route))
)
skip_all_budget_checks: Final = skip_budget_checks or route_skips_budget_checks(route=route)
membership_user_id: Final = (
valid_token.user_id if valid_token is not None and (bool(_model) or not skip_all_budget_checks) else None
@ -2099,7 +2105,7 @@ async def _fetch_uncached_tags(
@log_db_metrics
async def get_tag_objects_batch(
tag_names: list[str],
tag_names: Sequence[str],
prisma_client: PrismaClient | None,
user_api_key_cache: UserApiKeyCache,
parent_otel_span: Span | None = None,
@ -5821,16 +5827,46 @@ async def _tag_max_budget_check(
Raises:
BudgetExceededError if any tag is over its max budget.
Triggers a budget alert if any tag is over its max budget.
"""
from litellm.proxy.common_utils.http_parsing_utils import get_tags_from_request_body
await tag_max_budget_check_for_tags(
tags=get_tags_from_request_body(request_body=request_body),
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
if prisma_client is None:
return
# Get tags from request metadata
tags: Final = get_tags_from_request_body(request_body=request_body)
if not tags:
def _llm_provider_for_budget_error(model: str | list[str] | None) -> str:
"""Provider for the budget error's failure record, resolved from the request model when one is given."""
model_name: Final = model if isinstance(model, str) else (model[0] if model else None)
if model_name is None:
return ""
_, llm_provider = resolve_llm_provider_for_rate_limit(model_name)
return llm_provider
async def tag_max_budget_check_for_tags(
tags: Sequence[str],
prisma_client: PrismaClient | None,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
model: str | list[str] | None = None,
llm_router: Router | None = None,
) -> None:
"""
Check if any of ``tags`` is over its max budget.
Auth calls this with the tags in the request body. The request pipeline calls
it again after ``pre_call_hook`` with the tags a hook added, passing ``model``
and ``llm_router`` so a zero-cost model skips the check the way auth does and
the raised error names the provider. This is a plain read with no reservation,
the same fallback ``disable_budget_reservation`` uses, so concurrent requests
can overshoot the ceiling slightly.
Raises:
BudgetExceededError if any tag is over its max budget.
"""
if prisma_client is None or not tags or _is_model_cost_zero(model=model, llm_router=llm_router):
return
# Batch fetch all tags in one go
@ -5863,6 +5899,7 @@ async def _tag_max_budget_check(
current_cost=tag_spend,
max_budget=tag_object.litellm_budget_table.max_budget,
message=f"Budget has been exceeded! Tag={tag_name} Current cost: {tag_spend}, Max budget: {tag_object.litellm_budget_table.max_budget}",
llm_provider=_llm_provider_for_budget_error(model=model),
entity_type=Litellm_EntityType.TAG.value,
entity_id=tag_name,
)

75
litellm/proxy/common_request_processing.py
View file

@ -14,7 +14,7 @@ import httpx
import orjson
from fastapi import HTTPException, Request, status
from fastapi.responses import JSONResponse, Response, StreamingResponse
from pydantic import ValidationError
from pydantic import TypeAdapter, ValidationError
from starlette.types import Receive, Scope, Send
import litellm
@ -50,12 +50,17 @@ from litellm.litellm_core_utils.streaming_handler import (
backfill_missing_cache_usage_fields,
)
from litellm.proxy._types import ProxyException, UserAPIKeyAuth
from litellm.proxy.auth.auth_checks import can_key_call_resolved_model
from litellm.proxy.auth.auth_utils import check_response_size_is_safe
from litellm.proxy.auth.auth_checks import (
can_key_call_resolved_model,
route_skips_budget_checks,
tag_max_budget_check_for_tags,
)
from litellm.proxy.auth.auth_utils import check_response_size_is_safe, get_request_route
from litellm.proxy.common_utils.callback_utils import (
get_logging_caching_headers,
get_remaining_tokens_and_requests_from_request_data,
)
from litellm.proxy.common_utils.http_parsing_utils import get_tags_from_request_body
from litellm.proxy.common_utils.openai_error_payload import (
attribute_of,
error_status_code,
@ -642,6 +647,60 @@ async def _resolve_per_request_model_group_alias(
return target
_REQUEST_MODEL: Final[TypeAdapter[str | list[str] | None]] = TypeAdapter(str | list[str] | None)
def _request_model(data: Mapping[str, object]) -> str | list[str] | None:
"""The request's model name or names, or None when the field is missing or malformed."""
try:
return _REQUEST_MODEL.validate_python(data.get("model"), strict=True)
except ValidationError:
return None
def _tags_on_request(data: Mapping[str, object]) -> tuple[str, ...]:
"""Every tag on the request under either metadata key, since spend attribution reads both."""
without_litellm_metadata: Final = MappingProxyType(
{key: value for key, value in data.items() if key != "litellm_metadata"}
)
return tuple(
dict.fromkeys(
(
*get_tags_from_request_body(request_body=data),
*get_tags_from_request_body(request_body=without_litellm_metadata),
)
)
)
async def _enforce_tag_budgets_for_added_tags(
data: Mapping[str, object],
tags_before_pre_call_hook: frozenset[str],
route: str,
llm_router: Router | None,
proxy_logging_obj: ProxyLogging,
) -> None:
"""Budget-check the tags that ``pre_call_hook`` added to the request.
Tag budgets are enforced in auth against the tags in the request body, and
guardrails run after auth, so a tag a guardrail sets is only checked here,
on the same routes auth checks.
"""
added_tags: Final = tuple(tag for tag in _tags_on_request(data) if tag not in tags_before_pre_call_hook)
if not added_tags or route_skips_budget_checks(route=route):
return
from litellm.proxy.proxy_server import prisma_client, user_api_key_cache
await tag_max_budget_check_for_tags(
tags=added_tags,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
model=_request_model(data),
llm_router=llm_router,
)
async def _parse_event_data_for_error(event_line: str | bytes) -> int | None:
"""Parses an event line and returns an error code if present, else None."""
event_line = event_line.decode("utf-8") if isinstance(event_line, bytes) else event_line
@ -1522,6 +1581,7 @@ def _timing_values(
class ProxyBaseLLMRequestProcessing:
def __init__(self, data: dict):
self.data = data
self._tags_before_pre_call_hook: frozenset[str] | None = None
@staticmethod
def _merge_passthrough_streaming_headers(
@ -2011,6 +2071,8 @@ class ProxyBaseLLMRequestProcessing:
# to run below.
await _arm_auto_router_compression(data=self.data, llm_router=llm_router)
if self._tags_before_pre_call_hook is None:
self._tags_before_pre_call_hook = frozenset(_tags_on_request(self.data))
self.data = await proxy_logging_obj.pre_call_hook(
user_api_key_dict=user_api_key_dict,
data=self.data,
@ -2032,6 +2094,13 @@ class ProxyBaseLLMRequestProcessing:
if "messages" in self.data and self.data["messages"]:
logging_obj.update_messages(self.data["messages"])
await _enforce_tag_budgets_for_added_tags(
data=self.data,
tags_before_pre_call_hook=self._tags_before_pre_call_hook,
route=get_request_route(request=request),
llm_router=llm_router,
proxy_logging_obj=proxy_logging_obj,
)
return self.data, logging_obj
async def _pre_call_with_fallbacks(

107
tests/test_litellm/proxy/auth/test_auth_checks.py
View file

@ -2094,14 +2094,14 @@ def _tag_registry_row(tag_name: str):
return SimpleNamespace(tag_name=tag_name)
def _tag_db_row(tag_name: str, max_budget=None):
def _tag_db_row(tag_name: str, max_budget=None, spend: float = 0.0):
row = MagicMock()
row.tag_name = tag_name
budget = None if max_budget is None else {"max_budget": max_budget}
row.dict = MagicMock(
return_value={
"tag_name": tag_name,
"spend": 0.0,
"spend": spend,
"models": [],
"litellm_budget_table": budget,
}
@ -2378,6 +2378,109 @@ async def test_tag_max_budget_check_still_enforces_registered_tag_over_budget():
assert [call.kwargs["where"]["tag_name"]["in"] for call in batch_calls] == [["paid-tag"]]
def _over_budget_tag_prisma(tag_name: str, max_budget: float, spend: float):
"""A tag row whose recorded spend (the counter's authoritative fallback) is already over budget."""
async def fake_find_many(**kwargs):
if "where" not in kwargs:
return [_tag_registry_row(tag_name)]
return [_tag_db_row(name, max_budget=max_budget, spend=spend) for name in kwargs["where"]["tag_name"]["in"]]
mock_prisma = MagicMock()
mock_prisma.db.litellm_tagtable.find_many = AsyncMock(side_effect=fake_find_many)
return mock_prisma
def _zero_cost_router(model_name: str) -> "Router":
from litellm.router import Router
return Router(
model_list=[
{
"model_name": model_name,
"litellm_params": {"model": "openai/gpt-4.1-mini", "api_key": "sk-test"},
"model_info": {"input_cost_per_token": 0, "output_cost_per_token": 0},
}
]
)
@pytest.mark.asyncio
async def test_tag_max_budget_check_for_tags_rejects_over_budget_hook_added_tag():
"""A tag a pre-call hook added after auth is enforced like a tag auth saw in the body, with the provider filled in."""
from litellm.proxy.auth.auth_checks import tag_max_budget_check_for_tags
from litellm.proxy.utils import ProxyLogging
with pytest.raises(litellm.BudgetExceededError) as exc_info:
await tag_max_budget_check_for_tags(
tags=("guardrail-added-over-budget-tag",),
prisma_client=_over_budget_tag_prisma("guardrail-added-over-budget-tag", max_budget=1.0, spend=2.5),
user_api_key_cache=UserApiKeyCache(),
proxy_logging_obj=ProxyLogging(user_api_key_cache=None),
model="gpt-4.1-mini",
llm_router=None,
)
assert exc_info.value.entity_id == "guardrail-added-over-budget-tag"
assert exc_info.value.current_cost == 2.5
assert exc_info.value.max_budget == 1.0
assert exc_info.value.llm_provider == "openai"
@pytest.mark.asyncio
async def test_tag_max_budget_check_for_tags_without_model_leaves_provider_for_auth_to_resolve():
"""Auth resolves the provider for its own budget errors later, so the auth-path call must not pre-fill one."""
from litellm.proxy.auth.auth_checks import tag_max_budget_check_for_tags
from litellm.proxy.utils import ProxyLogging
with pytest.raises(litellm.BudgetExceededError) as exc_info:
await tag_max_budget_check_for_tags(
tags=("guardrail-added-over-budget-tag",),
prisma_client=_over_budget_tag_prisma("guardrail-added-over-budget-tag", max_budget=1.0, spend=2.5),
user_api_key_cache=UserApiKeyCache(),
proxy_logging_obj=ProxyLogging(user_api_key_cache=None),
)
assert exc_info.value.llm_provider == ""
@pytest.mark.asyncio
@pytest.mark.parametrize("model", ["free-model", ["free-model", "free-model"]], ids=["str", "list"])
async def test_tag_max_budget_check_for_tags_skips_zero_cost_model_like_auth(model):
"""Auth skips every budget check for a zero-cost model; the post-hook tag check must agree."""
from litellm.proxy.auth.auth_checks import tag_max_budget_check_for_tags
from litellm.proxy.utils import ProxyLogging
mock_prisma = _over_budget_tag_prisma("guardrail-added-over-budget-tag", max_budget=1.0, spend=2.5)
await tag_max_budget_check_for_tags(
tags=("guardrail-added-over-budget-tag",),
prisma_client=mock_prisma,
user_api_key_cache=UserApiKeyCache(),
proxy_logging_obj=ProxyLogging(user_api_key_cache=None),
model=model,
llm_router=_zero_cost_router("free-model"),
)
mock_prisma.db.litellm_tagtable.find_many.assert_not_awaited()
@pytest.mark.parametrize(
"route, expected",
[
("/v1/chat/completions", False),
("/v1/messages", False),
("/v1/batches", False),
("/guardrails/apply_guardrail", True),
("/prompts/test", True),
("/health", False),
("/v1/models", True),
],
)
def test_route_skips_budget_checks_matches_auth_scope(route, expected):
from litellm.proxy.auth.auth_checks import route_skips_budget_checks
assert route_skips_budget_checks(route=route) is expected
@pytest.mark.asyncio
async def test_get_team_object_raises_404_when_not_found():
from unittest.mock import AsyncMock, MagicMock

230
tests/test_litellm/proxy/test_common_request_processing.py
View file

@ -377,6 +377,236 @@ class TestProxyBaseLLMRequestProcessing:
assert "litellm_logging_obj" not in persisted_body
json.dumps(persisted_body)
@staticmethod
def _tag_budget_rig(monkeypatch, request_data: dict, pre_call_hook, route: str = "/v1/chat/completions"):
"""Wire common_processing_pre_call_logic with a fake request body and pre-call hook, capturing the tag check."""
mock_request = MagicMock(spec=Request)
mock_request.headers = {}
mock_request.scope = {"path": route}
async def mock_add_litellm_data_to_request(*args, **kwargs):
return request_data
mock_proxy_logging_obj = MagicMock(spec=ProxyLogging)
mock_proxy_logging_obj.pre_call_hook = AsyncMock(side_effect=pre_call_hook)
monkeypatch.setattr(
litellm.proxy.common_request_processing,
"add_litellm_data_to_request",
mock_add_litellm_data_to_request,
)
tag_check = AsyncMock()
monkeypatch.setattr(litellm.proxy.common_request_processing, "tag_max_budget_check_for_tags", tag_check)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", MagicMock())
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
return mock_request, mock_proxy_logging_obj, tag_check
@pytest.mark.asyncio
async def test_common_processing_pre_call_logic_enforces_tag_budgets_for_tags_added_in_pre_call_hook(
self, monkeypatch
):
"""
Tag budgets are checked in auth against the tags in the request body. A custom
guardrail runs later, inside pre_call_hook, so a tag it adds must be budget-checked
after the hook or the request reaches the model with an over-budget tag. The check
runs after the post-guardrail body snapshot, so the failure record carries what the
guardrails left behind, not the raw prompt.
"""
from litellm.router import Router
processing_obj = ProxyBaseLLMRequestProcessing(data={})
llm_router = Router(
model_list=[
{"model_name": "live-mini", "litellm_params": {"model": "openai/gpt-4.1-mini", "api_key": "sk-test"}}
]
)
async def mock_pre_call_hook(user_api_key_dict, data, call_type):
data["messages"][0]["content"] = "<MASKED>"
data["metadata"]["tags"].extend(["guardrail-tag", "existing-tag", "guardrail-tag"])
return data
mock_request, mock_proxy_logging_obj, tag_check = self._tag_budget_rig(
monkeypatch,
request_data={
"model": "live-mini",
"messages": [{"role": "user", "content": "my ssn is 123"}],
"metadata": {"tags": ["existing-tag"]},
"proxy_server_request": {"body": {"messages": [{"role": "user", "content": "my ssn is 123"}]}},
},
pre_call_hook=mock_pre_call_hook,
)
tag_check.side_effect = litellm.BudgetExceededError(current_cost=2.0, max_budget=1.0)
with pytest.raises(litellm.BudgetExceededError):
await processing_obj.common_processing_pre_call_logic(
request=mock_request,
general_settings={},
user_api_key_dict=ProxyUserAPIKeyAuth(token="test-token"),
proxy_logging_obj=mock_proxy_logging_obj,
proxy_config=None,
route_type="acompletion",
llm_router=llm_router,
)
tag_check.assert_awaited_once()
call_kwargs = tag_check.call_args.kwargs
assert call_kwargs["tags"] == ("guardrail-tag",)
assert call_kwargs["model"] == "live-mini"
assert call_kwargs["llm_router"] is llm_router
assert call_kwargs["proxy_logging_obj"] is mock_proxy_logging_obj
assert processing_obj.data["proxy_server_request"]["body"]["messages"][0]["content"] == "<MASKED>"
@pytest.mark.asyncio
async def test_common_processing_pre_call_logic_sees_guardrail_tags_under_either_metadata_key(self, monkeypatch):
"""Spend attribution reads metadata.tags even on routes that carry litellm_metadata, so the check must too."""
processing_obj = ProxyBaseLLMRequestProcessing(data={})
async def mock_pre_call_hook(user_api_key_dict, data, call_type):
data.setdefault("metadata", {}).setdefault("tags", []).append("guardrail-tag")
return data
mock_request, mock_proxy_logging_obj, tag_check = self._tag_budget_rig(
monkeypatch,
request_data={"model": "live-mini", "litellm_metadata": {"tags": ["existing-tag"]}},
pre_call_hook=mock_pre_call_hook,
route="/v1/messages",
)
await processing_obj.common_processing_pre_call_logic(
request=mock_request,
general_settings={},
user_api_key_dict=ProxyUserAPIKeyAuth(token="test-token"),
proxy_logging_obj=mock_proxy_logging_obj,
proxy_config=MagicMock(spec=ProxyConfig),
route_type="anthropic_messages",
)
assert tag_check.call_args.kwargs["tags"] == ("guardrail-tag",)
@pytest.mark.asyncio
async def test_common_processing_pre_call_logic_keeps_the_first_tag_snapshot_across_fallback_retries(
self, monkeypatch
):
"""A fallback retry reuses the mutated request, so the tag a guardrail added on attempt one still counts as added."""
processing_obj = ProxyBaseLLMRequestProcessing(data={})
async def mock_pre_call_hook(user_api_key_dict, data, call_type):
if "guardrail-tag" not in data["metadata"]["tags"]:
data["metadata"]["tags"].append("guardrail-tag")
return data
mock_request, mock_proxy_logging_obj, tag_check = self._tag_budget_rig(
monkeypatch,
request_data={"model": "live-mini", "metadata": {"tags": []}},
pre_call_hook=mock_pre_call_hook,
)
for _ in range(2):
await processing_obj.common_processing_pre_call_logic(
request=mock_request,
general_settings={},
user_api_key_dict=ProxyUserAPIKeyAuth(token="test-token"),
proxy_logging_obj=mock_proxy_logging_obj,
proxy_config=MagicMock(spec=ProxyConfig),
route_type="acompletion",
)
assert [call.kwargs["tags"] for call in tag_check.await_args_list] == [("guardrail-tag",), ("guardrail-tag",)]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"tags_after_hook",
[["existing-tag"], ["existing-tag", "existing-tag"], []],
ids=["unchanged", "duplicated", "removed"],
)
async def test_common_processing_pre_call_logic_skips_tag_budget_check_when_pre_call_hook_adds_no_tags(
self, monkeypatch, tags_after_hook
):
"""Auth already checked the tags the body carried, so only added tags cost a second check."""
processing_obj = ProxyBaseLLMRequestProcessing(data={})
async def mock_pre_call_hook(user_api_key_dict, data, call_type):
data["metadata"]["tags"] = list(tags_after_hook)
return data
mock_request, mock_proxy_logging_obj, tag_check = self._tag_budget_rig(
monkeypatch,
request_data={"model": "live-mini", "metadata": {"tags": ["existing-tag"]}},
pre_call_hook=mock_pre_call_hook,
)
returned_data, _ = await processing_obj.common_processing_pre_call_logic(
request=mock_request,
general_settings={},
user_api_key_dict=ProxyUserAPIKeyAuth(token="test-token"),
proxy_logging_obj=mock_proxy_logging_obj,
proxy_config=MagicMock(spec=ProxyConfig),
route_type="acompletion",
)
assert returned_data["metadata"]["tags"] == tags_after_hook
tag_check.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"model, expected",
[("live-mini", "live-mini"), (["mini-a", "mini-b"], ["mini-a", "mini-b"]), (123, None), ([1], None)],
ids=["str", "list", "int", "list-of-int"],
)
async def test_common_processing_pre_call_logic_hands_the_tag_check_only_well_formed_models(
self, monkeypatch, model, expected
):
"""The zero-cost skip and provider lookup take a model name or list; anything else is treated as unknown."""
processing_obj = ProxyBaseLLMRequestProcessing(data={})
async def mock_pre_call_hook(user_api_key_dict, data, call_type):
data["metadata"]["tags"].append("guardrail-tag")
return data
mock_request, mock_proxy_logging_obj, tag_check = self._tag_budget_rig(
monkeypatch,
request_data={"model": model, "metadata": {"tags": []}},
pre_call_hook=mock_pre_call_hook,
)
await processing_obj.common_processing_pre_call_logic(
request=mock_request,
general_settings={},
user_api_key_dict=ProxyUserAPIKeyAuth(token="test-token"),
proxy_logging_obj=mock_proxy_logging_obj,
proxy_config=MagicMock(spec=ProxyConfig),
route_type="acompletion",
)
assert tag_check.call_args.kwargs["model"] == expected
@pytest.mark.asyncio
async def test_common_processing_pre_call_logic_skips_tag_budget_check_on_routes_auth_exempts(self, monkeypatch):
"""Auth skips budget checks on non-LLM routes, so a hook-added tag must not be enforced there either."""
processing_obj = ProxyBaseLLMRequestProcessing(data={})
async def mock_pre_call_hook(user_api_key_dict, data, call_type):
data["metadata"]["tags"].append("guardrail-tag")
return data
mock_request, mock_proxy_logging_obj, tag_check = self._tag_budget_rig(
monkeypatch,
request_data={"metadata": {"tags": []}},
pre_call_hook=mock_pre_call_hook,
route="/guardrails/apply_guardrail",
)
await processing_obj.common_processing_pre_call_logic(
request=mock_request,
general_settings={},
user_api_key_dict=ProxyUserAPIKeyAuth(token="test-token"),
proxy_logging_obj=mock_proxy_logging_obj,
proxy_config=MagicMock(spec=ProxyConfig),
route_type="apply_guardrail",
)
tag_check.assert_not_awaited()
@pytest.mark.asyncio
async def test_common_processing_pre_call_logic_arms_auto_router_compression_before_guardrails(
self, monkeypatch