scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

test(e2e): add management suite covering key/team/user/org lifecycle and route permissions (#32300)

Browse source 
* test(e2e): add management suite covering key/team/user/org lifecycle and route permissions

* test(e2e): decouple the enforcement-flip assertion from upstream health

Polling for a 200 on the newly-allowed model required it to be a routable,
healthy upstream, which is not the contract under test; poll until the
key_model_access_denied 403 lifts instead, excluding 401 so a revoked key
cannot read as success. Also document that the delete test's deferred teardown
firing on an already-deleted key is deliberate: cleanup must survive the test
failing before the in-body delete, and the repeat delete is a warn-free no-op
(the proxy answers 404 No keys found)

* test(e2e): inline the management suite's model and tpm literals

* test(e2e): drop the models_mgmt suite line from the folder list

* test(e2e): write the tpm limit as a plain integer literal
This commit is contained in:
mubashir1osmani 2026-07-06 19:11:27 -07:00 • committed by GitHub
parent 40048814ee
commit a1873d89cc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 653 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
tests/e2e/CLAUDE.md
View file

@ -13,7 +13,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
- `realtime/` - realtime websocket sessions, including the pipecat audio path
- `budgets/` - budget definition, enforcement, and reset windows (key, team, tag, soft, multi-window)
- `spend_tracking/` - spend logging and cost attribution on `/spend/*`
- `models_mgmt/` - model-management routes (add/update, tpm persistence)
- `management/` - key/team/user/organization management routes: create/update/delete persistence via the info routes, team membership, and llm-only-key route denials
- `logging/` - logging-integration delivery (datadog and friends)
- `security/` - secret handling and log-leak protection
- `router/` - routing and reliability behavior (rate limits, fallbacks, cooldowns)

17
tests/e2e/management/conftest.py Normal file
View file

@ -0,0 +1,17 @@
"""Management suite client fixture; lifecycle/skip/marker live in the parent conftest."""
import pytest
from management_client import ManagementClient, build_client
def pytest_configure(config: pytest.Config) -> None:
config.addinivalue_line(
"markers",
"covers: registry cell a test covers, e.g. mgmt.key.generate.persists",
)
@pytest.fixture(scope="session")
def client() -> ManagementClient:
return build_client()

222
tests/e2e/management/management_client.py Normal file
View file

@ -0,0 +1,222 @@
"""Client for the management-routes e2e suite: the shared Gateway plus the
key/team/user/organization writes, the info/list read-backs the tests assert,
and the raw-status calls judged by HTTP outcome (chat under a scoped key, an
llm-only key hitting a management route).
"""
from __future__ import annotations
from dataclasses import dataclass
from e2e_gateway import Gateway, build_gateway
from e2e_http import NoBody, ProbeResult, StreamingResponse, unwrap
from models import (
ChatBody,
ChatMessage,
KeyDeleteBody,
KeyGenerateBody,
KeyListParams,
KeyListResponse,
KeyUpdateBody,
OrgDeleteBody,
OrgInfoParams,
OrgInfoResponse,
OrgNewBody,
OrgNewResponse,
TeamData,
TeamDeleteBody,
TeamInfoParams,
TeamInfoResponse,
TeamMemberAddBody,
TeamMemberDeleteBody,
TeamMemberEntry,
TeamNewBody,
TeamNewResponse,
UserDeleteBody,
UserInfoParams,
UserInfoResponse,
UserListParams,
UserListResponse,
UserNewBody,
UserNewResponse,
)
MODEL_ACCESS_DENIED_MARKER = "key_model_access_denied"
ROUTE_NOT_ALLOWED_MARKER = "not allowed to call this route"
@dataclass(frozen=True, slots=True)
class ManagementClient:
gateway: Gateway
def llm_only_key(self) -> str:
return self.gateway.generate_key(KeyGenerateBody(models=[], allowed_routes=["llm_api_routes"]))
def update_key_models(self, key: str, models: list[str]) -> None:
_ = unwrap(
self.gateway.transport.post(
"/key/update",
headers=self.gateway.transport.master,
json=KeyUpdateBody(key=key, models=models),
response_type=NoBody,
)
)
def delete_key_strict(self, key: str) -> None:
"""Strict delete for the act phase of a test: a failed delete is a hard
failure, unlike the warn-only Gateway.delete_key used at teardown."""
_ = unwrap(
self.gateway.transport.post(
"/key/delete",
headers=self.gateway.transport.master,
json=KeyDeleteBody(keys=[key]),
response_type=NoBody,
)
)
def key_alias_count(self, key_alias: str) -> int:
return unwrap(
self.gateway.transport.get(
"/key/list",
headers=self.gateway.transport.master,
params=KeyListParams(key_alias=key_alias),
response_type=KeyListResponse,
)
).total_count
def create_team(self, body: TeamNewBody) -> str:
return unwrap(
self.gateway.transport.post(
"/team/new",
headers=self.gateway.transport.master,
json=body,
response_type=TeamNewResponse,
)
).team_id
def delete_team(self, team_id: str) -> None:
_ = self.gateway.transport.post(
"/team/delete",
headers=self.gateway.transport.master,
json=TeamDeleteBody(team_ids=[team_id]),
response_type=NoBody,
)
def team_info(self, team_id: str) -> TeamData:
return unwrap(
self.gateway.transport.get(
"/team/info",
headers=self.gateway.transport.master,
params=TeamInfoParams(team_id=team_id),
response_type=TeamInfoResponse,
)
).team_info
def team_info_status(self, team_id: str) -> ProbeResult:
return self.gateway.transport.probe("/team/info", params=TeamInfoParams(team_id=team_id))
def add_team_member(self, team_id: str, user_id: str) -> None:
_ = unwrap(
self.gateway.transport.post(
"/team/member_add",
headers=self.gateway.transport.master,
json=TeamMemberAddBody(team_id=team_id, member=TeamMemberEntry(role="user", user_id=user_id)),
response_type=NoBody,
)
)
def delete_team_member(self, team_id: str, user_id: str) -> None:
_ = unwrap(
self.gateway.transport.post(
"/team/member_delete",
headers=self.gateway.transport.master,
json=TeamMemberDeleteBody(team_id=team_id, user_id=user_id),
response_type=NoBody,
)
)
def create_user(self, body: UserNewBody) -> str:
return unwrap(
self.gateway.transport.post(
"/user/new",
headers=self.gateway.transport.master,
json=body,
response_type=UserNewResponse,
)
).user_id
def delete_user(self, user_id: str) -> None:
_ = self.gateway.transport.post(
"/user/delete",
headers=self.gateway.transport.master,
json=UserDeleteBody(user_ids=[user_id]),
response_type=NoBody,
)
def user_info(self, user_id: str) -> UserInfoResponse:
return unwrap(
self.gateway.transport.get(
"/user/info",
headers=self.gateway.transport.master,
params=UserInfoParams(user_id=user_id),
response_type=UserInfoResponse,
)
)
def user_count(self, user_id: str) -> int:
return unwrap(
self.gateway.transport.get(
"/user/list",
headers=self.gateway.transport.master,
params=UserListParams(user_ids=user_id),
response_type=UserListResponse,
)
).total
def create_org(self, body: OrgNewBody) -> str:
return unwrap(
self.gateway.transport.post(
"/organization/new",
headers=self.gateway.transport.master,
json=body,
response_type=OrgNewResponse,
)
).organization_id
def delete_org(self, organization_id: str) -> None:
_ = self.gateway.transport.delete(
"/organization/delete",
headers=self.gateway.transport.master,
json=OrgDeleteBody(organization_ids=[organization_id]),
response_type=NoBody,
)
def org_info(self, organization_id: str) -> OrgInfoResponse:
return unwrap(
self.gateway.transport.get(
"/organization/info",
headers=self.gateway.transport.master,
params=OrgInfoParams(organization_id=organization_id),
response_type=OrgInfoResponse,
)
)
def chat_status(self, key: str, model: str, content: str) -> StreamingResponse:
return self.gateway.transport.send(
"/chat/completions",
headers=self.gateway.transport.bearer(key),
json=ChatBody(model=model, messages=[ChatMessage(role="user", content=content)], max_tokens=16),
)
def key_generate_status(self, key: str, body: KeyGenerateBody) -> StreamingResponse:
return self.gateway.transport.send("/key/generate", headers=self.gateway.transport.bearer(key), json=body)
def team_new_status(self, key: str, body: TeamNewBody) -> StreamingResponse:
return self.gateway.transport.send("/team/new", headers=self.gateway.transport.bearer(key), json=body)
def user_new_status(self, key: str, body: UserNewBody) -> StreamingResponse:
return self.gateway.transport.send("/user/new", headers=self.gateway.transport.bearer(key), json=body)
def build_client() -> ManagementClient:
return ManagementClient(gateway=build_gateway())

286
tests/e2e/management/test_management_e2e.py Normal file
View file

@ -0,0 +1,286 @@
"""Live e2e: the key/team/user/organization management routes' lifecycle contract.
Each test creates its resources under unique names (deleted on teardown) and
asserts both halves of the contract: the recorded state (the info route reflects
the write) and the enforced behavior (the data plane serves or refuses traffic
accordingly). Key writes reach the data plane when its auth cache entry expires,
so the traffic-facing read-backs poll to a deadline instead of asserting once.
"""
from __future__ import annotations
import time
from collections.abc import Callable
import pytest
from e2e_config import unique_marker
from e2e_http import StreamingResponse
from lifecycle import ResourceManager
from management_client import (
MODEL_ACCESS_DENIED_MARKER,
ROUTE_NOT_ALLOWED_MARKER,
ManagementClient,
)
from models import KeyGenerateBody, OrgNewBody, TeamNewBody, UserNewBody
pytestmark = pytest.mark.e2e
def _poll[T](client: ManagementClient, attempt: Callable[[], T | None], failure: str) -> T:
deadline = time.monotonic() + client.gateway.poll_timeout
while time.monotonic() < deadline:
found = attempt()
if found is not None:
return found
time.sleep(client.gateway.poll_interval)
pytest.fail(failure)
def _generate_key(client: ManagementClient, resources: ResourceManager, body: KeyGenerateBody) -> str:
key = client.gateway.generate_key(body)
resources.defer(lambda: client.gateway.delete_key(key))
return key
def _create_team(client: ManagementClient, resources: ResourceManager, alias: str, models: list[str]) -> str:
team_id = client.create_team(TeamNewBody(team_alias=alias, models=models))
resources.defer(lambda: client.delete_team(team_id))
return team_id
def _create_user(client: ManagementClient, resources: ResourceManager, body: UserNewBody) -> str:
user_id = client.create_user(body)
resources.defer(lambda: client.delete_user(user_id))
return user_id
def _is_model_denial(outcome: StreamingResponse) -> bool:
return outcome.status_code == 403 and MODEL_ACCESS_DENIED_MARKER in outcome.body
def _assert_model_denied(outcome: StreamingResponse, model: str) -> None:
assert outcome.status_code == 403, (
f"chat on {model!r} outside the key's model list must be denied 403, got "
f"{outcome.status_code}: {outcome.body[:300]}"
)
assert MODEL_ACCESS_DENIED_MARKER in outcome.body, (
f"403 body must be a model-access denial, got: {outcome.body[:300]}"
)
def _poll_chat_ok(client: ManagementClient, key: str, model: str) -> None:
def attempt() -> bool | None:
outcome = client.chat_status(key, model, f"reply with one word {unique_marker()}")
return True if outcome.ok else None
_ = _poll(client, attempt, f"chat on {model} never succeeded for the key before the deadline")
def _poll_chat_denied(client: ManagementClient, key: str, model: str) -> None:
def attempt() -> bool | None:
return True if _is_model_denial(client.chat_status(key, model, f"say hi {unique_marker()}")) else None
_ = _poll(
client,
attempt,
f"chat on {model} was never denied with {MODEL_ACCESS_DENIED_MARKER} before the deadline",
)
def _poll_model_access_granted(client: ManagementClient, key: str, model: str) -> None:
"""The key's model-access check stopped denying `model`: any outcome other than
the key_model_access_denied 403 (a 200, or an upstream error) proves the flip.
Requiring a 200 would couple the assertion to `model` being a healthy routable
upstream, which is not the enforcement contract under test."""
def attempt() -> bool | None:
outcome = client.chat_status(key, model, f"say hi {unique_marker()}")
if _is_model_denial(outcome) or outcome.status_code == 401:
return None
return True
_ = _poll(client, attempt, f"model-access denial on {model} never lifted before the deadline")
class TestKeyRoutes:
@pytest.mark.covers("mgmt.key.generate.persists")
def test_generate_persists_to_key_info_and_scopes_chat(
self, client: ManagementClient, resources: ResourceManager
) -> None:
alias = f"e2e-mgmt-key-{unique_marker()}"
key = _generate_key(
client,
resources,
KeyGenerateBody(models=["gemini-2.5-flash"], key_alias=alias, tpm_limit=424242),
)
info = client.gateway.key_info(key)
assert info.key_alias == alias, f"/key/info reports key_alias {info.key_alias!r}, configured {alias!r}"
assert info.models == ["gemini-2.5-flash"], (
f"/key/info reports models {info.models}, configured ['gemini-2.5-flash']"
)
assert info.tpm_limit == 424242, (
f"/key/info reports tpm_limit {info.tpm_limit}, configured 424242"
)
_poll_chat_ok(client, key, "gemini-2.5-flash")
_assert_model_denied(
client.chat_status(key, "gpt-5.5", f"say hi {unique_marker()}"), "gpt-5.5"
)
@pytest.mark.covers("mgmt.key.update.persists")
def test_update_models_persists_and_flips_enforcement(
self, client: ManagementClient, resources: ResourceManager
) -> None:
key = _generate_key(client, resources, KeyGenerateBody(models=["gemini-2.5-flash"]))
_poll_chat_ok(client, key, "gemini-2.5-flash")
_assert_model_denied(
client.chat_status(key, "gpt-5.5", f"say hi {unique_marker()}"), "gpt-5.5"
)
client.update_key_models(key, ["gpt-5.5"])
info = client.gateway.key_info(key)
assert info.models == ["gpt-5.5"], (
f"/key/info reports models {info.models} after /key/update to ['gpt-5.5']"
)
_poll_model_access_granted(client, key, "gpt-5.5")
_poll_chat_denied(client, key, "gemini-2.5-flash")
@pytest.mark.covers("mgmt.key.delete.persists")
def test_delete_revokes_the_key_on_chat(self, client: ManagementClient, resources: ResourceManager) -> None:
"""The teardown's deferred delete fires again on the already-deleted key by
design: the deferred cleanup must survive this test failing before the
in-body delete, and a repeat /key/delete is a cheap no-op the warn-only
teardown absorbs."""
key = _generate_key(client, resources, KeyGenerateBody(models=["gemini-2.5-flash"]))
_poll_chat_ok(client, key, "gemini-2.5-flash")
client.delete_key_strict(key)
def rejected() -> bool | None:
outcome = client.chat_status(key, "gemini-2.5-flash", f"say hi {unique_marker()}")
return True if outcome.status_code == 401 else None
_ = _poll(client, rejected, "deleted key was still accepted on chat (never rejected 401) at the deadline")
class TestTeamRoutes:
@pytest.mark.covers("management.team.new.persists")
def test_new_persists_to_team_info_and_binds_keys(
self, client: ManagementClient, resources: ResourceManager
) -> None:
alias = f"e2e-mgmt-team-{unique_marker()}"
team_id = _create_team(client, resources, alias, ["gemini-2.5-flash"])
info = client.team_info(team_id)
assert info.team_alias == alias, f"/team/info reports team_alias {info.team_alias!r}, configured {alias!r}"
assert info.models == ["gemini-2.5-flash"], (
f"/team/info reports models {info.models}, configured ['gemini-2.5-flash']"
)
key = _generate_key(client, resources, KeyGenerateBody(team_id=team_id))
key_info = client.gateway.key_info(key)
assert key_info.team_id == team_id, (
f"key generated under team {team_id} carries team_id {key_info.team_id!r} in /key/info"
)
@pytest.mark.covers("mgmt.team.member_add.persists")
def test_member_add_and_delete_persist_to_team_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
user_id = _create_user(
client,
resources,
UserNewBody(user_email=f"e2e-mgmt-{unique_marker()}@example.com", user_role="internal_user"),
)
team_id = _create_team(client, resources, f"e2e-mgmt-team-{unique_marker()}", ["gemini-2.5-flash"])
client.add_team_member(team_id, user_id)
member = next(
(entry for entry in client.team_info(team_id).members_with_roles if entry.user_id == user_id), None
)
assert member is not None, f"/team/info does not list {user_id} after /team/member_add"
assert member.role == "user", f"member {user_id} added with role 'user' but /team/info reports {member.role!r}"
client.delete_team_member(team_id, user_id)
remaining = client.team_info(team_id).members_with_roles
assert all(entry.user_id != user_id for entry in remaining), (
f"/team/info still lists {user_id} after /team/member_delete"
)
class TestUserRoutes:
@pytest.mark.covers("mgmt.user.new.persists")
def test_new_persists_to_user_info(self, client: ManagementClient, resources: ResourceManager) -> None:
email = f"e2e-mgmt-{unique_marker()}@example.com"
user_id = _create_user(client, resources, UserNewBody(user_email=email, user_role="internal_user"))
info = client.user_info(user_id).user_info
assert info.user_email == email, f"/user/info reports user_email {info.user_email!r}, configured {email!r}"
assert info.user_role == "internal_user", (
f"/user/info reports user_role {info.user_role!r}, configured 'internal_user'"
)
class TestOrganizationRoutes:
@pytest.mark.covers("mgmt.organization.new.persists")
def test_new_persists_to_organization_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
alias = f"e2e-mgmt-org-{unique_marker()}"
org_id = client.create_org(OrgNewBody(organization_alias=alias, models=["gemini-2.5-flash"]))
resources.defer(lambda: client.delete_org(org_id))
info = client.org_info(org_id)
assert info.organization_alias == alias, (
f"/organization/info reports alias {info.organization_alias!r}, configured {alias!r}"
)
assert info.models == ["gemini-2.5-flash"], (
f"/organization/info reports models {info.models}, configured ['gemini-2.5-flash']"
)
def _assert_route_forbidden(route: str, outcome: StreamingResponse) -> None:
assert outcome.status_code == 403, (
f"llm-only key POSTing {route} must be denied exactly 403, got {outcome.status_code}: {outcome.body[:300]}"
)
assert ROUTE_NOT_ALLOWED_MARKER in outcome.body, (
f"{route} denial body must be a route-permission denial, got: {outcome.body[:300]}"
)
class TestManagementRoutePermissions:
@pytest.mark.covers("mgmt.key.generate.member_forbidden")
def test_llm_only_key_forbidden_from_management_writes(
self, client: ManagementClient, resources: ResourceManager
) -> None:
key = client.llm_only_key()
resources.defer(lambda: client.gateway.delete_key(key))
marker = unique_marker()
alias = f"e2e-mgmt-forbidden-key-{marker}"
team_id = f"e2e-mgmt-forbidden-team-{marker}"
user_id = f"e2e-mgmt-forbidden-user-{marker}"
_assert_route_forbidden(
"/key/generate", client.key_generate_status(key, KeyGenerateBody(models=[], key_alias=alias))
)
_assert_route_forbidden(
"/team/new", client.team_new_status(key, TeamNewBody(team_alias=team_id, team_id=team_id))
)
_assert_route_forbidden(
"/user/new",
client.user_new_status(
key,
UserNewBody(user_email=f"{user_id}@example.com", user_role="internal_user", user_id=user_id),
),
)
assert client.key_alias_count(alias) == 0, f"key {alias} was created despite the 403 route denial"
team_probe = client.team_info_status(team_id)
assert team_probe.status_code == 404, (
f"team {team_id} was created despite the 403 route denial: "
f"/team/info returned {team_probe.status_code}: {team_probe.body[:300]}"
)
assert client.user_count(user_id) == 0, f"user {user_id} was created despite the 403 route denial"

127
tests/e2e/models.py
View file

@ -61,6 +61,10 @@ class LiteLLMBudgetTable(BaseModel):
class KeyInfo(BaseModel):
key_alias: str | None = None
models: list[str] = []
tpm_limit: int | None = None
team_id: str | None = None
spend: float | None = None
max_budget: float | None = None
budget_reset_at: str | None = None
@ -408,3 +412,126 @@ class ModelNewResponse(BaseModel):
class ModelDeleteBody(BaseModel):
id: str
# ---------- key / team / user / organization management ----------
class KeyUpdateBody(BaseModel):
key: str
models: list[str]
class KeyListParams(BaseModel):
key_alias: str
class KeyListResponse(BaseModel):
total_count: int
class TeamMemberEntry(BaseModel):
role: Literal["admin", "user"]
user_id: str
class TeamNewBody(BaseModel):
team_alias: str
models: list[str] = []
team_id: str | None = None
class TeamNewResponse(BaseModel):
team_id: str
class TeamInfoParams(BaseModel):
team_id: str
class TeamData(BaseModel):
team_alias: str | None = None
models: list[str] = []
members_with_roles: list[TeamMemberEntry] = []
class TeamInfoResponse(BaseModel):
team_id: str
team_info: TeamData
class TeamMemberAddBody(BaseModel):
team_id: str
member: TeamMemberEntry
class TeamMemberDeleteBody(BaseModel):
team_id: str
user_id: str
class TeamDeleteBody(BaseModel):
team_ids: list[str]
UserRole = Literal["proxy_admin", "proxy_admin_viewer", "internal_user", "internal_user_viewer"]
class UserNewBody(BaseModel):
user_email: str
user_role: UserRole
user_id: str | None = None
class UserNewResponse(BaseModel):
user_id: str
class UserInfoParams(BaseModel):
user_id: str
class UserData(BaseModel):
user_id: str | None = None
user_email: str | None = None
user_role: str | None = None
class UserInfoResponse(BaseModel):
user_id: str
user_info: UserData
class UserDeleteBody(BaseModel):
user_ids: list[str]
class UserListParams(BaseModel):
user_ids: str
class UserListResponse(BaseModel):
total: int
class OrgNewBody(BaseModel):
organization_alias: str
models: list[str] = []
class OrgNewResponse(BaseModel):
organization_id: str
class OrgInfoParams(BaseModel):
organization_id: str
class OrgInfoResponse(BaseModel):
organization_id: str
organization_alias: str | None = None
models: list[str] = []
class OrgDeleteBody(BaseModel):
organization_ids: list[str]