scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-08 03:08:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 115

test(e2e): tag management tests with Subject metadata and record management client steps (#44962)

Browse source 
* test(e2e): add enum values, auto-discovering label gates and secret hiding for e2e metadata

* test(e2e): tag management tests with Subject metadata and record management client steps

* docs(e2e): name every markerless harness test file that carries no Subject

* test(e2e): keep the step discovery comprehensions to one for clause

* test(e2e): keep the prompt out of the chat_status step so polled retries collapse
This commit is contained in:
ryan-crabbe-berri 2026-10-07 10:32:09 -07:00 • committed by GitHub
parent 0ae55bdf7a
commit d7cdc88c66
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 545 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
tests/e2e/management/jwt_actors.py
View file

@ -5,6 +5,7 @@ from typing import Final, Literal
from e2e_config import unique_marker
from e2e_http import NoBody, unwrap
from e2e_metadata import step
from idp import ADMIN_CLIENT_ID, TESTS_CLIENT_ID, Identity, Keycloak
from lifecycle import ResourceManager
from management.management_client import ManagementClient
@ -53,6 +54,7 @@ class Actor:
profile: ActorProfile
tenants: tuple[Tenant, ...]
@step("Get a JWT from the identity provider for the actor in the {self.role} role")
def mint_caller(self, idp: Keycloak) -> Caller:
return Caller(
credential=idp.access_token(
@ -74,6 +76,7 @@ class ActorFactory:
if self.bootstrap.proxy.caller is not None:
raise ValueError("Actor bootstrap requires a separately held master client")
@step("Generate a virtual key as the proxy admin")
def key(self, tenant: Tenant | None = None, *, user_id: str | None = None) -> KeyGenerateResponse:
created: Final = unwrap(
self.bootstrap.generate_key(
@ -87,6 +90,7 @@ class ActorFactory:
self.resources.defer(lambda: self.bootstrap.delete_key_strict(created.key, missing_ok=True))
return created
@step("Create an organization, a team in it and a matching identity provider group")
def tenant(self) -> Tenant:
marker: Final = unique_marker()
organization_id: Final = self.bootstrap.create_org(OrgNewBody(organization_alias=f"e2e-organization-{marker}"))
@ -118,6 +122,7 @@ class ActorFactory:
self.resources.defer(lambda: self.idp.with_strict_cleanup().delete_group(group_id))
return Tenant(organization_id=organization_id, team_id=team_id, group_id=group_id)
@step("Create an actor in the {role} role, with its identity provider user, internal user and any tenant memberships")
def create(
self, role: ActorRole, *, tenants: tuple[Tenant, ...] = (), profile: ActorProfile = "database_role"
) -> Actor:

68
tests/e2e/management/management_client.py
View file

@ -24,6 +24,7 @@ from e2e_http import (
retry_attempts,
unwrap,
)
from e2e_metadata import STEP_FRAMES, step
from models import (
AuditLogPage,
AuditLogParams,
@ -116,9 +117,11 @@ class ManagementClient:
def with_caller(self, caller: Caller) -> ManagementClient:
return replace(self, proxy=self.proxy.with_caller(caller))
@step("Generate a virtual key limited to the LLM API routes")
def llm_only_key(self) -> str:
return self.proxy.generate_key(KeyGenerateBody(models=[], allowed_routes=["llm_api_routes"]))
@step("Generate a virtual key with {body}")
def generate_key(self, body: KeyGenerateBody, *, caller_key: str | None = None) -> Result[KeyGenerateResponse]:
"""POST /key/generate. `caller_key` is who is creating the key: the master
key by default, or a virtual key (an admin filling in Create New Key on the
@ -133,6 +136,7 @@ class ManagementClient:
response_type=KeyGenerateResponse,
)
@step("Update the virtual key's settings with /key/update")
def update_key(self, body: KeyUpdateBody, *, caller_key: str | None = None) -> Result[NoBody]:
"""POST /key/update. `caller_key` is who is editing: the master key by
default, or a virtual key (the dashboard edits under the session key its
@ -152,16 +156,22 @@ class ManagementClient:
case UnknownApiError(body=error_body) if any(
marker in error_body.lower() for marker in _TRANSIENT_BACKEND_MARKERS
):
warnings.warn(f"Transient backend response on attempt {attempt + 1}", RuntimeWarning, stacklevel=2)
warnings.warn(
f"Transient backend response on attempt {attempt + 1}",
RuntimeWarning,
stacklevel=2 + STEP_FRAMES,
)
time.sleep(0.5 * (attempt + 1))
continue
case _:
break
return last
@step("Set the virtual key's models to [{models}]")
def update_key_models(self, key: str, models: list[str]) -> None:
_ = unwrap(self.update_key(KeyUpdateBody(key=key, models=models)))
@step("Delete the virtual key with the alias {key_alias}")
def delete_key_by_alias(self, key_alias: str) -> None:
_ = unwrap(
self.proxy.transport.post(
@ -172,6 +182,7 @@ class ManagementClient:
)
)
@step("Read the key's deletion entries from the /audit log")
def key_deleted_audit_logs(self, token_hash: str) -> AuditLogPage:
return unwrap(
self.proxy.transport.get(
@ -187,6 +198,7 @@ class ManagementClient:
)
)
@step("Read the key's settings back from /key/info")
def key_info_as(self, key: str, *, caller_key: str | None = None) -> Result[KeyInfoResponse]:
return self.proxy.transport.get(
"/key/info",
@ -195,6 +207,7 @@ class ManagementClient:
response_type=KeyInfoResponse,
)
@step("Delete the virtual key")
def delete_key_strict(self, key: str, *, caller_key: str | None = None, missing_ok: bool = False) -> None:
"""Strict delete for the act phase of a test: a failed delete is a hard
failure, unlike the warn-only ProxyClient.delete_key used at teardown."""
@ -208,6 +221,7 @@ class ManagementClient:
return
_ = unwrap(result)
@step("Delete the deployment")
def delete_model_strict(self, model_id: str) -> None:
"""Strict delete for the act phase of a test: a failed delete is a hard
failure, unlike the warn-only ProxyClient.delete_model used at teardown."""
@ -220,6 +234,7 @@ class ManagementClient:
)
)
@step("Run Test Connection on {body.litellm_params.model} in {body.mode} mode with /health/test_connection")
def connection_test(self, body: ConnectionTestBody) -> Result[ConnectionTestResponse]:
"""POST /health/test_connection, the call behind the Admin UI's Test
Connection button, probing the live provider with the supplied params."""
@ -231,6 +246,7 @@ class ManagementClient:
timeout=120.0,
)
@step("Block the virtual key")
def block_key(self, key: str) -> None:
_ = unwrap(
self.proxy.transport.post(
@ -240,6 +256,7 @@ class ManagementClient:
response_type=NoBody,
)
)
@step("Regenerate the virtual key with /key/regenerate")
def regenerate_key(self, key: str, *, grace_period: str | None = None) -> str:
return unwrap(
self.proxy.transport.post(
@ -250,6 +267,7 @@ class ManagementClient:
)
).key
@step("Reset the virtual key's spend to {reset_to}")
def reset_key_spend(self, key: str, reset_to: float) -> KeyResetSpendResponse:
return unwrap(
self.proxy.transport.post(
@ -260,6 +278,7 @@ class ManagementClient:
)
)
@step("List the keys with the alias {key_alias} from /key/list")
def key_list(self, key_alias: str, *, caller_key: str | None = None) -> Result[KeyListResponse]:
"""GET /key/list, the Virtual Keys page's own inventory call. `caller_key` is
who is asking: the master key by default, or a virtual key."""
@ -271,9 +290,11 @@ class ManagementClient:
response_type=KeyListResponse,
)
@step("Count the keys with the alias {key_alias} in /key/list")
def key_alias_count(self, key_alias: str) -> int:
return unwrap(self.key_list(key_alias)).total_count
@step("Sign in to the Admin UI with /v2/login")
def dashboard_login(self, username: str, password: str) -> DashboardSession:
"""POST /v2/login, the call the Admin UI's sign-in form makes.
@ -297,6 +318,7 @@ class ManagementClient:
redirect_url=response.redirect_url,
)
@step("Create a team with {body}")
def create_team(self, body: TeamNewBody) -> str:
team_id = unwrap(
self.proxy.transport.post(
@ -309,6 +331,7 @@ class ManagementClient:
self._wait_for_team(team_id)
return team_id
@step("Update a team with {body}")
def update_team(self, body: TeamUpdateBody) -> None:
last: Result[NoBody] | None = None
for attempt in range(retry_attempts(5)):
@ -324,7 +347,11 @@ class ManagementClient:
case UnknownApiError(body=body_text) if (
"connecting to redis" in body_text.lower() or "name resolution" in body_text.lower()
):
warnings.warn(f"Transient backend response on attempt {attempt + 1}", RuntimeWarning, stacklevel=2)
warnings.warn(
f"Transient backend response on attempt {attempt + 1}",
RuntimeWarning,
stacklevel=2 + STEP_FRAMES,
)
time.sleep(0.5 * (attempt + 1))
continue
case _:
@ -332,6 +359,7 @@ class ManagementClient:
assert last is not None
raise AssertionError(last)
@step("Delete the team")
def delete_team(self, team_id: str) -> None:
_ = self.proxy.transport.post(
"/team/delete",
@ -340,6 +368,7 @@ class ManagementClient:
response_type=NoBody,
)
@step("Read the team back from /team/info")
def team_info(self, team_id: str) -> TeamData:
return unwrap(
self.proxy.transport.get(
@ -350,6 +379,7 @@ class ManagementClient:
)
).team_info
@step("List the teams from /team/list")
def team_list_ids(self) -> tuple[str, ...]:
return tuple(
entry.team_id
@ -363,6 +393,7 @@ class ManagementClient:
).root
)
@step("Check whether /team/info finds the team")
def team_info_status(self, team_id: str) -> ProbeResult:
return self.proxy.transport.probe(
"/team/info", params=TeamInfoParams(team_id=team_id), headers=self.proxy.management_headers()
@ -386,6 +417,7 @@ class ManagementClient:
assert last is not None
raise AssertionError(last)
@step("Add a user to the team with /team/member_add")
def add_team_member(self, team_id: str, user_id: str) -> None:
last: Result[NoBody] | None = None
for attempt in range(retry_attempts(_TEAM_READY_ATTEMPTS)):
@ -402,7 +434,9 @@ class ManagementClient:
_TEAM_READY_ATTEMPTS
):
warnings.warn(
"Retrying team membership while the team becomes available", RuntimeWarning, stacklevel=2
"Retrying team membership while the team becomes available",
RuntimeWarning,
stacklevel=2 + STEP_FRAMES,
)
time.sleep(_TEAM_READY_SLEEP_SECONDS)
continue
@ -411,6 +445,7 @@ class ManagementClient:
assert last is not None
raise AssertionError(last)
@step("Add a roster of members to the team with /team/member_add")
def add_team_members(self, team_id: str, members: list[TeamMemberEntry]) -> None:
"""Bulk form of /team/member_add: `member` accepts a list, so one call
seeds a whole roster the way an admin import does."""
@ -423,6 +458,7 @@ class ManagementClient:
)
)
@step("Try to delete the team with /team/delete")
def delete_team_status(self, team_id: str) -> StreamingResponse:
"""POST /team/delete judged by HTTP outcome: the raw status and body, so a
test can assert on what a caller actually sees when the delete fails."""
@ -432,6 +468,7 @@ class ManagementClient:
json=TeamDeleteBody(team_ids=[team_id]),
)
@step("Remove a user from the team with /team/member_delete")
def delete_team_member(self, team_id: str, user_id: str) -> None:
_ = unwrap(
self.proxy.transport.post(
@ -442,6 +479,7 @@ class ManagementClient:
)
)
@step("Create an internal user with {body}")
def create_user(self, body: UserNewBody) -> str:
return unwrap(
self.proxy.transport.post(
@ -452,6 +490,7 @@ class ManagementClient:
)
).user_id
@step("Create the end user {user_id}")
def create_customer(self, user_id: str) -> str:
_ = unwrap(
self.proxy.transport.post(
@ -463,6 +502,7 @@ class ManagementClient:
)
return user_id
@step("Read the end user {end_user_id} back from /customer/info")
def customer_info(self, end_user_id: str) -> CustomerResponse:
return unwrap(
self.proxy.transport.get(
@ -473,6 +513,7 @@ class ManagementClient:
)
)
@step("Delete the end user {user_id}")
def delete_customer(self, user_id: str) -> None:
_ = self.proxy.transport.post(
"/customer/delete",
@ -481,6 +522,7 @@ class ManagementClient:
response_type=NoBody,
)
@step("Update an internal user with {body}")
def update_user(self, body: UserUpdateBody) -> None:
_ = unwrap(
self.proxy.transport.post(
@ -491,6 +533,7 @@ class ManagementClient:
)
)
@step("Delete the internal user")
def delete_user(self, user_id: str) -> None:
_ = self.proxy.transport.post(
"/user/delete",
@ -499,6 +542,7 @@ class ManagementClient:
response_type=NoBody,
)
@step("Delete the internal user")
def delete_user_strict(self, user_id: str) -> None:
"""Strict delete for the act phase of a test: a failed delete is a hard
failure, unlike the warn-only delete_user used at teardown."""
@ -511,6 +555,7 @@ class ManagementClient:
)
)
@step("Read the user back from /user/info")
def user_info(self, user_id: str | None = None) -> UserInfoResponse:
return unwrap(
self.proxy.transport.get(
@ -521,6 +566,7 @@ class ManagementClient:
)
)
@step("Count the matching users in /user/list")
def user_count(self, user_id: str) -> int:
return unwrap(
self.proxy.transport.get(
@ -531,6 +577,7 @@ class ManagementClient:
)
).total
@step("List the matching users from /user/list")
def user_list_ids(self, user_id: str) -> tuple[str, ...]:
listing = unwrap(
self.proxy.transport.get(
@ -542,6 +589,7 @@ class ManagementClient:
)
return tuple(row.user_id for row in listing.users)
@step("Create an organization with {body}")
def create_org(self, body: OrgNewBody) -> str:
return unwrap(
self.proxy.transport.post(
@ -552,6 +600,7 @@ class ManagementClient:
)
).organization_id
@step("Update an organization with {body}")
def update_org(self, body: OrgUpdateBody) -> None:
_ = unwrap(
self.proxy.transport.patch(
@ -562,6 +611,7 @@ class ManagementClient:
)
)
@step("Delete the organization")
def delete_org(self, organization_id: str) -> None:
_ = self.proxy.transport.delete(
"/organization/delete",
@ -570,6 +620,7 @@ class ManagementClient:
response_type=NoBody,
)
@step("Read the organization back from /organization/info")
def org_info(self, organization_id: str) -> OrgInfoResponse:
return unwrap(
self.proxy.transport.get(
@ -580,6 +631,7 @@ class ManagementClient:
)
)
@step("Check whether /organization/info finds the organization")
def org_info_status(self, organization_id: str) -> ProbeResult:
return self.proxy.transport.probe(
"/organization/info",
@ -587,6 +639,7 @@ class ManagementClient:
headers=self.proxy.management_headers(),
)
@step("Create a tag with {body}")
def create_tag(self, body: TagNewBody) -> None:
_ = unwrap(
self.proxy.transport.post(
@ -597,6 +650,7 @@ class ManagementClient:
)
)
@step("Delete the tag {name}")
def delete_tag(self, name: str) -> None:
_ = self.proxy.transport.post(
"/tag/delete",
@ -605,6 +659,7 @@ class ManagementClient:
response_type=NoBody,
)
@step("List the tags from /tag/list")
def tag_list(self) -> tuple[TagListEntry, ...]:
return tuple(
unwrap(
@ -617,6 +672,7 @@ class ManagementClient:
).root
)
@step("Create an MCP server named {body.alias}")
def create_mcp_server(self, body: McpServerCreateBody) -> McpServerRow:
return unwrap(
self.proxy.transport.post(
@ -627,6 +683,7 @@ class ManagementClient:
)
)
@step("Update the MCP server's settings with PUT /v1/mcp/server")
def update_mcp_server(self, body: McpServerUpdateBody) -> McpServerRow:
"""PUT /v1/mcp/server, the call behind the dashboard's Save Changes: a partial
update where a field left unset keeps its stored value and None clears it."""
@ -639,6 +696,7 @@ class ManagementClient:
)
)
@step("Delete the MCP server")
def delete_mcp_server(self, server_id: str) -> Result[NoBody]:
"""DELETE /v1/mcp/server/{server_id}. Returns the outcome so the act phase can
unwrap it while a deferred teardown can ignore an already-deleted server."""
@ -649,6 +707,7 @@ class ManagementClient:
response_type=NoBody,
)
@step("Send a /chat/completions request to {model}")
def chat_status(self, key: str, model: str, content: str) -> StreamingResponse:
return self.proxy.transport.send(
"/chat/completions",
@ -656,12 +715,15 @@ class ManagementClient:
json=ChatBody(model=model, messages=[ChatMessage(role="user", content=content)], max_tokens=16),
)
@step("Try to generate a virtual key with {body}, calling with a virtual key")
def key_generate_status(self, key: str, body: KeyGenerateBody) -> StreamingResponse:
return self.proxy.transport.send("/key/generate", headers=self.proxy.transport.bearer(key), json=body)
@step("Try to create a team with {body}, calling with a virtual key")
def team_new_status(self, key: str, body: TeamNewBody) -> StreamingResponse:
return self.proxy.transport.send("/team/new", headers=self.proxy.transport.bearer(key), json=body)
@step("Try to create an internal user with {body}, calling with a virtual key")
def user_new_status(self, key: str, body: UserNewBody) -> StreamingResponse:
return self.proxy.transport.send("/user/new", headers=self.proxy.transport.bearer(key), json=body)

14
tests/e2e/management/test_budget_customer_user_org_e2e.py
View file

@ -23,6 +23,7 @@ from pydantic import BaseModel, Field, RootModel
from e2e_config import unique_marker
from e2e_http import NoBody, Success, UnauthorizedError, UnknownApiError, is_ok, unwrap
from e2e_metadata import Domain, Route, Subject, meta
from lifecycle import ResourceManager
from management_client import ManagementClient
from models import KeyGenerateBody, ModelBudgetEntry, OrgInfoParams, OrgNewBody, UserNewBody
@ -152,6 +153,7 @@ _UPDATED_MAX_BUDGET = 91.25
class TestBudgetManagement:
@pytest.mark.covers("mgmt.budget.list.happy_path")
@meta(Subject(domain=Domain.SPEND_BUDGETS, route=Route.BUDGET_MANAGEMENT))
def test_created_budget_appears_in_budget_list(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -164,6 +166,7 @@ class TestBudgetManagement:
)
@pytest.mark.covers("mgmt.budget.update.accepts_model_max_budget")
@meta(Subject(domain=Domain.SPEND_BUDGETS, route=Route.BUDGET_MANAGEMENT))
def test_update_accepts_per_model_budgets_including_punctuated_names(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -213,6 +216,7 @@ class TestBudgetManagement:
)
@pytest.mark.covers("mgmt.budget.update.persists")
@meta(Subject(domain=Domain.SPEND_BUDGETS, route=Route.BUDGET_MANAGEMENT))
def test_update_max_budget_persists_to_budget_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -247,6 +251,7 @@ class TestBudgetManagement:
)
@pytest.mark.covers("mgmt.budget.new.admin_only")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.BUDGET_MANAGEMENT))
def test_new_is_refused_for_a_non_admin_key(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -352,6 +357,7 @@ class TestBudgetListV1:
"""
@pytest.mark.covers("mgmt.budget.list_v1.happy_path")
@meta(Subject(domain=Domain.SPEND_BUDGETS, route=Route.BUDGET_MANAGEMENT))
def test_sorts_pages_and_filters_the_budgets_it_created(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -392,6 +398,7 @@ class TestBudgetListV1:
assert [row.tpm_limit for row in limits] == [60000, 60000, 60000]
@pytest.mark.covers("mgmt.budget.list_v1.happy_path")
@meta(Subject(domain=Domain.SPEND_BUDGETS, route=Route.BUDGET_MANAGEMENT))
def test_is_null_finds_the_budget_left_uncapped(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -409,11 +416,13 @@ class TestBudgetListV1:
assert [row.max_budget for row in _list_budgets(client, found).data] == [None]
@pytest.mark.covers("mgmt.budget.list_v1.happy_path")
@meta(Subject(domain=Domain.SPEND_BUDGETS, route=Route.BUDGET_MANAGEMENT))
def test_refuses_a_sort_field_and_a_parameter_it_does_not_support(self, client: ManagementClient) -> None:
assert _list_status(client, BudgetPageParams(sort="budget_duration")) == 400
assert _list_status(client, BudgetPageParams(not_a_parameter="b-1")) == 400
@pytest.mark.covers("mgmt.budget.list_v1.admin_only")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.BUDGET_MANAGEMENT))
def test_is_refused_for_a_non_admin_key(self, client: ManagementClient, resources: ResourceManager) -> None:
key = client.proxy.generate_key(KeyGenerateBody())
resources.defer(lambda: client.proxy.delete_key(key))
@ -482,6 +491,7 @@ def _customer_info(client: ManagementClient, route: str, user_id: str) -> Custom
class TestCustomerManagement:
@pytest.mark.covers("mgmt.customer.new.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.CUSTOMER_MANAGEMENT))
def test_new_persists_to_customer_info(self, client: ManagementClient, resources: ResourceManager) -> None:
customer_id = f"e2e-mgmt-cust-{unique_marker()}"
created = _create_customer(
@ -495,6 +505,7 @@ class TestCustomerManagement:
)
@pytest.mark.covers("mgmt.customer.delete.persists")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.CUSTOMER_MANAGEMENT))
def test_delete_removes_the_customer(self, client: ManagementClient, resources: ResourceManager) -> None:
"""The teardown's deferred delete fires again on the already-deleted customer
by design: it is the safety net if this test fails before the in-body delete,
@ -524,6 +535,7 @@ class TestCustomerManagement:
_ = _poll(client, gone, f"customer {customer_id} still resolved on /customer/info after /customer/delete")
@pytest.mark.covers("mgmt.end_user.new.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.CUSTOMER_MANAGEMENT))
def test_end_user_new_persists_to_end_user_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -542,6 +554,7 @@ class TestCustomerManagement:
class TestUserManagement:
@pytest.mark.covers("mgmt.user.info.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.USER_MANAGEMENT))
def test_new_user_is_readable_via_user_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -586,6 +599,7 @@ class OrgInfoMembersResponse(BaseModel):
class TestOrganizationMembership:
@pytest.mark.covers("mgmt.organization.member_add.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.ORGANIZATION_MANAGEMENT))
def test_member_add_records_membership(
self, client: ManagementClient, resources: ResourceManager
) -> None:

12
tests/e2e/management/test_config_misc_endpoints_e2e.py
View file

@ -32,6 +32,7 @@ from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import NoBody, Success, unwrap, unwrap_status
from e2e_metadata import Domain, Route, Subject, meta
from lifecycle import ResourceManager
from management_client import ManagementClient
from models import KeyGenerateBody, LiteLLMParamsBody, TeamNewBody
@ -231,6 +232,7 @@ class McpServerResponse(BaseModel):
class TestInventoryRoutes:
@pytest.mark.covers("mgmt.callback.list.happy_path")
@meta(Subject(domain=Domain.OBSERVABILITY))
def test_callbacks_list_reports_active_logging_callbacks(self, client: ManagementClient) -> None:
listing = unwrap(
client.proxy.transport.get(
@ -247,6 +249,7 @@ class TestInventoryRoutes:
)
@pytest.mark.covers("mgmt.tool_management.list.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT))
def test_tool_list_returns_catalog_with_consistent_total(self, client: ManagementClient) -> None:
listing = unwrap(
client.proxy.transport.get(
@ -261,6 +264,7 @@ class TestInventoryRoutes:
)
@pytest.mark.covers("mgmt.workflow.list.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT))
def test_workflow_runs_list_returns_consistent_count(self, client: ManagementClient) -> None:
listing = unwrap(
client.proxy.transport.get(
@ -275,6 +279,7 @@ class TestInventoryRoutes:
)
@pytest.mark.covers("mgmt.credential_migration.check.happy_path")
@meta(Subject(domain=Domain.DEPLOY_OPS))
def test_credential_migration_check_reports_residual_scan(self, client: ManagementClient) -> None:
report = unwrap(
client.proxy.transport.get(
@ -295,6 +300,7 @@ class TestInventoryRoutes:
class TestCostEstimate:
@pytest.mark.covers("mgmt.cost_tracking.estimate.happy_path")
@meta(Subject(domain=Domain.COST_MAP))
def test_estimate_computes_cost_from_token_counts(self, client: ManagementClient) -> None:
estimate = unwrap(
client.proxy.transport.post(
@ -325,6 +331,7 @@ class TestCostEstimate:
class TestComplianceRoutes:
@pytest.mark.covers("mgmt.compliance.gdpr.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT))
def test_gdpr_check_derives_verdict_from_the_request(self, client: ManagementClient) -> None:
result = unwrap(
client.proxy.transport.post(
@ -356,6 +363,7 @@ class TestComplianceRoutes:
class TestFallbackManagement:
@pytest.mark.covers("mgmt.fallback_management.update.happy_path")
@meta(Subject(domain=Domain.ROUTING))
def test_create_persists_and_is_read_back(self, client: ManagementClient, resources: ResourceManager) -> None:
primary = f"e2e-fallback-primary-{unique_marker()}"
secondary = f"e2e-fallback-secondary-{unique_marker()}"
@ -410,6 +418,7 @@ class TestFallbackManagement:
class TestJwtKeyMapping:
@pytest.mark.covers("mgmt.jwt_key_mapping.new.happy_path")
@meta(Subject(domain=Domain.PROXY_AUTH))
def test_new_persists_mapping_and_is_read_back(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -461,6 +470,7 @@ class TestJwtKeyMapping:
class TestRouterSettings:
@pytest.mark.covers("mgmt.router_settings.update.happy_path")
@meta(Subject(domain=Domain.ROUTING, route=Route.PROXY_CONFIG))
def test_config_update_persists_router_setting_to_get(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -531,6 +541,7 @@ class TestRouterSettings:
class TestMcpServerSubmission:
@pytest.mark.covers("mgmt.mcp_server.register.happy_path")
@meta(Subject(domain=Domain.MCP, route=Route.MCP))
def test_register_submits_pending_server(self, client: ManagementClient, resources: ResourceManager) -> None:
"""A non-admin, team-scoped key submits an MCP server for review; the proxy
stores it as pending_review without loading it into the runtime registry."""
@ -564,6 +575,7 @@ class TestMcpServerSubmission:
)
@pytest.mark.covers("mgmt.mcp_server.approve.persists")
@meta(Subject(domain=Domain.MCP, route=Route.MCP))
def test_approve_activates_submission_and_persists(
self, client: ManagementClient, resources: ResourceManager
) -> None:

10
tests/e2e/management/test_jwt_management_e2e.py
View file

@ -7,6 +7,7 @@ from typing import Final, Literal
import pytest
from e2e_config import CHEAP_OPENAI_MODEL, PROXY_BASE_URL, unique_marker
from e2e_http import UnauthorizedError, UnknownApiError, unwrap
from e2e_metadata import Domain, Route, Subject, meta
from idp import ADMIN_CLIENT_ID, Identity, Keycloak, token_claims
from lifecycle import ResourceManager
from management.jwt_actors import ActorFactory, ActorRole
@ -32,6 +33,7 @@ class TestJwtManagement:
),
)
@pytest.mark.covers("mgmt.user.jwt.database_roles")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.USER_MANAGEMENT))
def test_actor_subject_and_database_role(self, actor_factory: ActorFactory, role: ActorRole) -> None:
tenants: Final = (
(actor_factory.tenant(),) if role in ("organization_admin", "team_admin", "team_member") else ()
@ -70,6 +72,7 @@ class TestJwtManagement:
} == {(actor.identity.user_id, "org_admin" if role == "organization_admin" else "internal_user")}
@pytest.mark.covers("mgmt.key.jwt.viewer_denied")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.KEY_MANAGEMENT))
def test_admin_viewer_reads_but_cannot_update(self, actor_factory: ActorFactory) -> None:
actor: Final = actor_factory.create("proxy_admin_viewer")
viewer: Final = actor_factory.bootstrap.with_caller(actor.mint_caller(actor_factory.idp))
@ -83,6 +86,7 @@ class TestJwtManagement:
assert actor_factory.bootstrap.proxy.key_info(key).key_alias == alias
@pytest.mark.covers("mgmt.user.oidc.identity_mapping")
@meta(Subject(domain=Domain.PROXY_AUTH))
def test_oidc_browser_profile_identity_mapping(self, actor_factory: ActorFactory) -> None:
actor: Final = actor_factory.create("internal_user")
idp: Final = actor_factory.idp.with_strict_cleanup()
@ -100,6 +104,7 @@ class TestJwtManagement:
@pytest.mark.covers("mgmt.key.jwt.lifecycle")
@pytest.mark.parametrize("credential_kind", ("direct_jwt", "virtual_key"))
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.KEY_MANAGEMENT))
def test_admin_creates_reads_updates_clears_and_deletes_a_key(
self,
actor_factory: ActorFactory,
@ -142,6 +147,7 @@ class TestJwtManagement:
assert unwrap(bound.key_list(updated_alias)).total_count == 0
@pytest.mark.covers("mgmt.team.jwt.tenant_isolation")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.KEY_MANAGEMENT))
def test_two_actor_sets_keep_tenants_and_keys_isolated(self, actor_factory: ActorFactory) -> None:
first: Final = actor_factory.tenant()
second: Final = actor_factory.tenant()
@ -163,6 +169,7 @@ class TestJwtManagement:
assert tuple(actor.identity.groups for actor in actors) == ((first.team_id,), (second.team_id,))
@pytest.mark.covers("mgmt.team.jwt.multiple_memberships")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_multi_group_actor_keeps_exact_memberships(self, actor_factory: ActorFactory) -> None:
tenants: Final = (actor_factory.tenant(), actor_factory.tenant())
actor: Final = actor_factory.create("team_member", tenants=tenants, profile="group_scoped")
@ -177,6 +184,7 @@ class TestJwtManagement:
} == {(actor.identity.user_id, "user")}
@pytest.mark.covers("mgmt.user.jwt.cleanup")
@meta(Subject(domain=Domain.PROXY_AUTH))
def test_successful_actor_cleanup_removes_owned_state(self, actor_factory: ActorFactory) -> None:
resources: Final = ResourceManager(client=actor_factory.bootstrap.proxy, strict_cleanup=True)
factory: Final = ActorFactory(bootstrap=actor_factory.bootstrap, idp=actor_factory.idp, resources=resources)
@ -197,6 +205,7 @@ class TestJwtManagement:
@pytest.mark.parametrize("stage", ("group", "user"))
@pytest.mark.covers("mgmt.user.jwt.partial_cleanup")
@meta(Subject(domain=Domain.PROXY_AUTH))
def test_partial_setup_removes_previously_created_identities(
self,
actor_factory: ActorFactory,
@ -236,6 +245,7 @@ class TestJwtManagement:
idp.assert_absent("users", identity.user_id)
@pytest.mark.covers("mgmt.key.jwt.member_denied", "mgmt.key.jwt.other_team_denied")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.KEY_MANAGEMENT))
def test_member_cannot_write_and_another_team_cannot_read_the_key(
self, client: ManagementClient, idp: Keycloak, jwt_identity: Identity, resources: ResourceManager
) -> None:

22
tests/e2e/management/test_key_lifecycle_e2e.py
View file

@ -23,6 +23,7 @@ import pytest
from e2e_config import unique_marker
from e2e_http import Result, StreamingResponse, Success, unwrap
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from lifecycle import ResourceManager
from management_client import MODEL_ACCESS_DENIED_MARKER, ManagementClient
from models import (
@ -188,6 +189,7 @@ def _assert_chat_rejected_everywhere(client: ManagementClient, key: str, model:
class TestKeyLifecycle:
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.KEY_MANAGEMENT))
def test_create_echoes_every_field_written(
self, client: ManagementClient, resources: ResourceManager, mock_deployment: str
) -> None:
@ -206,6 +208,7 @@ class TestKeyLifecycle:
):
assert observed == wanted, f"/key/generate echoed {field}={observed!r}, sent {wanted!r}"
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.KEY_MANAGEMENT))
def test_read_reflects_the_create_on_every_replica(
self, client: ManagementClient, resources: ResourceManager, mock_deployment: str
) -> None:
@ -221,6 +224,7 @@ class TestKeyLifecycle:
)
@pytest.mark.covers("mgmt.key.update.preserves_unrelated_fields")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.KEY_MANAGEMENT))
def test_partial_update_changes_only_the_named_field(
self, client: ManagementClient, resources: ResourceManager, mock_deployment: str
) -> None:
@ -238,6 +242,7 @@ class TestKeyLifecycle:
)
@pytest.mark.covers("mgmt.key.update.clear_persists")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.KEY_MANAGEMENT))
def test_explicit_null_clears_the_budget_and_its_reset_time(
self, client: ManagementClient, resources: ResourceManager, mock_deployment: str
) -> None:
@ -258,6 +263,14 @@ class TestKeyLifecycle:
info, created.written.model_copy(update={"max_budget": None, "budget_duration": None}), replica
)
@meta(
Subject(
domain=Domain.PROXY_AUTH,
providers=(Provider.OPENAI,),
models=(BACKING_MODEL,),
mode=Mode.NONSTREAM,
)
)
def test_key_serves_its_model_and_is_denied_others(
self, client: ManagementClient, resources: ResourceManager, mock_deployment: str
) -> None:
@ -274,6 +287,15 @@ class TestKeyLifecycle:
f"403 body must be a model-access denial, got: {denied.body[:300]}"
)
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.OPENAI,),
models=(BACKING_MODEL,),
mode=Mode.NONSTREAM,
)
)
def test_delete_revokes_info_and_chat_on_every_replica(
self, client: ManagementClient, resources: ResourceManager, mock_deployment: str
) -> None:

29
tests/e2e/management/test_key_management_e2e.py
View file

@ -19,6 +19,7 @@ import pytest
from e2e_config import unique_marker
from e2e_http import NoBody, StreamingResponse, unwrap
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from lifecycle import ResourceManager
from management_client import ManagementClient
from models import (
@ -31,6 +32,7 @@ pytestmark = pytest.mark.e2e
TINY_BUDGET = 3e-6
SPEND_MODEL = "claude-haiku-4-5"
SYNTHETIC_BACKEND: Final = "openai/synthetic-detachment"
class KeyToggleBlockBody(BaseModel):
@ -161,13 +163,22 @@ def project_resources(client: ManagementClient) -> Iterator[ResourceManager]:
class TestKeyManagementRoutes:
@pytest.mark.covers("mgmt.key.update.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.OPENAI,),
models=(SYNTHETIC_BACKEND,),
mode=Mode.NONSTREAM,
)
)
def test_project_detachment_preserves_key_scope_and_refreshes_auth(
self, client: ManagementClient, project_resources: ResourceManager
) -> None:
resources: Final = project_resources
name: Final = f"e2e-detach-{unique_marker()}"
model_id: Final = client.proxy.create_model(
name, LiteLLMParamsBody(model="openai/synthetic-detachment", api_key="synthetic", mock_response="orbit")
name, LiteLLMParamsBody(model=SYNTHETIC_BACKEND, api_key="synthetic", mock_response="orbit")
)
resources.defer(lambda: client.proxy.delete_model(model_id))
org_id: Final = client.create_org(OrgNewBody(organization_alias=name, models=[name]))
@ -222,6 +233,7 @@ class TestKeyManagementRoutes:
assert denied.status_code in (401, 403), denied.body
@pytest.mark.covers("mgmt.key.info.persists")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.KEY_MANAGEMENT))
def test_info_reflects_the_fields_the_key_was_created_with(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -246,6 +258,7 @@ class TestKeyManagementRoutes:
assert info.rpm_limit == 141414, f"/key/info reports rpm_limit {info.rpm_limit}, configured 141414"
@pytest.mark.covers("mgmt.key.unblock.persists")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.KEY_MANAGEMENT))
def test_unblock_flips_key_info_blocked_back(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -266,6 +279,7 @@ class TestKeyManagementRoutes:
)
@pytest.mark.covers("mgmt.key.health.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.KEY_MANAGEMENT))
def test_health_reports_the_calling_key_healthy(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -285,6 +299,7 @@ class TestKeyManagementRoutes:
)
@pytest.mark.covers("mgmt.key.bulk_update.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.KEY_MANAGEMENT))
def test_bulk_update_applies_max_budget_to_target_key(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -314,6 +329,15 @@ class TestKeyManagementRoutes:
)
@pytest.mark.covers("other.key_mgmt.spend_reset.resets_to_value")
@meta(
Subject(
domain=Domain.SPEND_BUDGETS,
route=Route.KEY_MANAGEMENT,
providers=(Provider.ANTHROPIC,),
models=(SPEND_MODEL,),
mode=Mode.NONSTREAM,
)
)
def test_reset_spend_zeroes_recorded_spend_and_lifts_the_budget_block(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -340,6 +364,7 @@ class TestKeyManagementRoutes:
_ = _poll(client, call_allowed_again, "the key stayed budget-blocked after its spend was reset to 0")
@pytest.mark.covers("mgmt.key.generate.admin_only")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.KEY_MANAGEMENT))
def test_generate_forbidden_for_non_admin_key(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -355,6 +380,7 @@ class TestKeyManagementRoutes:
)
@pytest.mark.covers("mgmt.key.delete.admin_only")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.KEY_MANAGEMENT))
def test_delete_forbidden_for_non_admin_key(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -374,6 +400,7 @@ class TestKeyManagementRoutes:
)
@pytest.mark.covers("mgmt.key.update.admin_only")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.KEY_MANAGEMENT))
def test_update_forbidden_for_non_admin_key(
self, client: ManagementClient, resources: ResourceManager
) -> None:

308
tests/e2e/management/test_management_e2e.py
View file

@ -18,6 +18,7 @@ import pytest
from e2e_config import POLL_INTERVAL, POLL_TIMEOUT, UI_PASSWORD, UI_USERNAME, unique_marker
from e2e_http import StreamingResponse, Success, unwrap
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from lifecycle import ResourceManager
from management_client import (
DASHBOARD_SESSION_TEAM_ID,
@ -47,6 +48,8 @@ from proxy_client import Converged, await_converged
pytestmark = pytest.mark.e2e
GEMINI_MODEL: Final = "gemini-2.5-flash"
OPENAI_MODEL: Final = "gpt-5.5"
REGENERATE_GRACE_PERIOD = "15s"
REGENERATE_GRACE_SECONDS = 15.0
TEAM_DELETE_POOL_OVERFLOW_MEMBERS = 250
@ -130,6 +133,15 @@ def _poll_model_access_granted(client: ManagementClient, key: str, model: str) -
class TestKeyRoutes:
@pytest.mark.covers("mgmt.key.generate.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.GEMINI,),
models=(GEMINI_MODEL,),
mode=Mode.NONSTREAM,
)
)
def test_generate_persists_to_key_info_and_scopes_chat(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -137,12 +149,12 @@ class TestKeyRoutes:
key = _generate_key(
client,
resources,
KeyGenerateBody(models=["gemini-2.5-flash"], key_alias=alias, tpm_limit=424242, rpm_limit=424243),
KeyGenerateBody(models=[GEMINI_MODEL], key_alias=alias, tpm_limit=424242, rpm_limit=424243),
)
info = client.proxy.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"], (
assert info.models == [GEMINI_MODEL], (
f"/key/info reports models {info.models}, configured ['gemini-2.5-flash']"
)
assert info.tpm_limit == 424242, (
@ -152,49 +164,73 @@ class TestKeyRoutes:
f"/key/info reports rpm_limit {info.rpm_limit}, configured 424243"
)
_poll_chat_ok(client, key, "gemini-2.5-flash")
_poll_chat_ok(client, key, GEMINI_MODEL)
_assert_model_denied(
client.chat_status(key, "gpt-5.5", f"say hi {unique_marker()}"), "gpt-5.5"
client.chat_status(key, OPENAI_MODEL, f"say hi {unique_marker()}"), OPENAI_MODEL
)
@pytest.mark.covers("mgmt.key.update.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.GEMINI, Provider.OPENAI),
models=(GEMINI_MODEL, OPENAI_MODEL),
mode=Mode.NONSTREAM,
)
)
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")
key = _generate_key(client, resources, KeyGenerateBody(models=[GEMINI_MODEL]))
_poll_chat_ok(client, key, GEMINI_MODEL)
_assert_model_denied(
client.chat_status(key, "gpt-5.5", f"say hi {unique_marker()}"), "gpt-5.5"
client.chat_status(key, OPENAI_MODEL, f"say hi {unique_marker()}"), OPENAI_MODEL
)
client.update_key_models(key, ["gpt-5.5"])
client.update_key_models(key, [OPENAI_MODEL])
info = client.proxy.key_info(key)
assert info.models == ["gpt-5.5"], (
assert info.models == [OPENAI_MODEL], (
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")
_poll_model_access_granted(client, key, OPENAI_MODEL)
_poll_chat_denied(client, key, GEMINI_MODEL)
@pytest.mark.covers("mgmt.key.delete.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.GEMINI,),
models=(GEMINI_MODEL,),
mode=Mode.NONSTREAM,
)
)
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")
key = _generate_key(client, resources, KeyGenerateBody(models=[GEMINI_MODEL]))
_poll_chat_ok(client, key, GEMINI_MODEL)
client.delete_key_strict(key)
def rejected() -> bool | None:
outcome = client.chat_status(key, "gemini-2.5-flash", f"say hi {unique_marker()}")
outcome = client.chat_status(key, GEMINI_MODEL, 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")
@pytest.mark.covers("mgmt.key.list.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
)
)
def test_created_key_appears_in_key_list_inventory(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -214,8 +250,14 @@ class TestKeyRoutes:
@pytest.mark.covers("mgmt.key.block.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
)
)
def test_block_persists_to_key_info(self, client: ManagementClient, resources: ResourceManager) -> None:
key = _generate_key(client, resources, KeyGenerateBody(models=["gemini-2.5-flash"]))
key = _generate_key(client, resources, KeyGenerateBody(models=[GEMINI_MODEL]))
assert not client.proxy.key_info(key).blocked, "/key/info reports the key blocked before /key/block ran"
client.block_key(key)
@ -233,6 +275,15 @@ class TestDashboardKeyRoutes:
are the same routes the API-surface tests cover with a different caller."""
@pytest.mark.covers("mgmt.key.generate.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.GEMINI,),
models=(GEMINI_MODEL,),
mode=Mode.NONSTREAM,
)
)
def test_creating_a_key_from_the_dashboard_persists_and_works(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -260,7 +311,7 @@ class TestDashboardKeyRoutes:
def dashboard_creates_the_key() -> str | None:
match client.generate_key(
KeyGenerateBody(models=["gemini-2.5-flash"], key_alias=alias, tpm_limit=100),
KeyGenerateBody(models=[GEMINI_MODEL], key_alias=alias, tpm_limit=100),
caller_key=session.session_key,
):
case Success(data=created):
@ -280,7 +331,7 @@ class TestDashboardKeyRoutes:
f"/key/info reports key_alias {created_info.key_alias!r} for the key the dashboard created, "
f"expected {alias!r}"
)
assert created_info.models == ["gemini-2.5-flash"], (
assert created_info.models == [GEMINI_MODEL], (
f"/key/info reports models {created_info.models} for the key the dashboard created"
)
assert created_info.tpm_limit == 100, (
@ -301,10 +352,19 @@ class TestDashboardKeyRoutes:
"would render no keys",
)
_poll_chat_ok(client, created, "gemini-2.5-flash")
_assert_model_denied(client.chat_status(created, "gpt-5.5", f"say hi {unique_marker()}"), "gpt-5.5")
_poll_chat_ok(client, created, GEMINI_MODEL)
_assert_model_denied(client.chat_status(created, OPENAI_MODEL, f"say hi {unique_marker()}"), OPENAI_MODEL)
@pytest.mark.covers("mgmt.key.update.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.GEMINI, Provider.OPENAI),
models=(GEMINI_MODEL, OPENAI_MODEL),
mode=Mode.NONSTREAM,
)
)
def test_editing_a_key_from_the_dashboard_persists_and_is_enforced(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -312,17 +372,17 @@ class TestDashboardKeyRoutes:
target = _generate_key(
client,
resources,
KeyGenerateBody(models=["gemini-2.5-flash"], key_alias=alias, tpm_limit=100, rpm_limit=200),
KeyGenerateBody(models=[GEMINI_MODEL], key_alias=alias, tpm_limit=100, rpm_limit=200),
)
_poll_chat_ok(client, target, "gemini-2.5-flash")
_assert_model_denied(client.chat_status(target, "gpt-5.5", f"say hi {unique_marker()}"), "gpt-5.5")
_poll_chat_ok(client, target, GEMINI_MODEL)
_assert_model_denied(client.chat_status(target, OPENAI_MODEL, f"say hi {unique_marker()}"), OPENAI_MODEL)
session = client.dashboard_login(UI_USERNAME, UI_PASSWORD)
resources.defer(lambda: client.proxy.delete_key(session.session_key))
def dashboard_saves_the_edit() -> bool | None:
match client.update_key(
KeyUpdateBody(key=target, models=["gpt-5.5"], tpm_limit=300, rpm_limit=400),
KeyUpdateBody(key=target, models=[OPENAI_MODEL], tpm_limit=300, rpm_limit=400),
caller_key=session.session_key,
):
case Success():
@ -337,7 +397,7 @@ class TestDashboardKeyRoutes:
)
info = client.proxy.key_info(target)
assert info.models == ["gpt-5.5"], (
assert info.models == [OPENAI_MODEL], (
f"/key/info reports models {info.models} after the dashboard edit to ['gpt-5.5']"
)
assert info.tpm_limit == 300, f"/key/info reports tpm_limit {info.tpm_limit} after the dashboard edit to 300"
@ -346,29 +406,38 @@ class TestDashboardKeyRoutes:
f"the dashboard edit renamed the key to {info.key_alias!r}, it should still be {alias!r}"
)
_poll_model_access_granted(client, target, "gpt-5.5")
_poll_chat_denied(client, target, "gemini-2.5-flash")
_poll_model_access_granted(client, target, OPENAI_MODEL)
_poll_chat_denied(client, target, GEMINI_MODEL)
class TestKeyRegeneration:
@pytest.mark.covers("mgmt.key.regenerate.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.OPENAI,),
models=(OPENAI_MODEL,),
mode=Mode.NONSTREAM,
)
)
def test_regenerate_rotates_to_a_working_new_key(
self, client: ManagementClient, resources: ResourceManager
) -> None:
old_key = _generate_key(client, resources, KeyGenerateBody(models=["gpt-5.5"]))
old_key = _generate_key(client, resources, KeyGenerateBody(models=[OPENAI_MODEL]))
new_key = client.regenerate_key(old_key)
resources.defer(lambda: client.proxy.delete_key(new_key))
assert new_key != old_key, "regenerate returned the same key string, so no rotation happened"
def new_accepted() -> bool | None:
outcome = client.chat_status(new_key, "gpt-5.5", f"say hi {unique_marker()}")
outcome = client.chat_status(new_key, OPENAI_MODEL, f"say hi {unique_marker()}")
return True if outcome.status_code != 401 else None
_ = _poll(client, new_accepted, "regenerated key was never accepted at auth (still 401) at the deadline")
def old_rejected() -> bool | None:
outcome = client.chat_status(old_key, "gpt-5.5", f"say hi {unique_marker()}")
outcome = client.chat_status(old_key, OPENAI_MODEL, f"say hi {unique_marker()}")
return True if outcome.status_code == 401 else None
_ = _poll(
@ -376,10 +445,19 @@ class TestKeyRegeneration:
)
@pytest.mark.covers("other.key_mgmt.regenerate.grace_period_honored")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
providers=(Provider.OPENAI,),
models=(OPENAI_MODEL,),
mode=Mode.NONSTREAM,
)
)
def test_regenerate_with_grace_period_keeps_old_key_until_revoked(
self, client: ManagementClient, resources: ResourceManager
) -> None:
old_key = _generate_key(client, resources, KeyGenerateBody(models=["gpt-5.5"]))
old_key = _generate_key(client, resources, KeyGenerateBody(models=[OPENAI_MODEL]))
new_key = client.regenerate_key(old_key, grace_period=REGENERATE_GRACE_PERIOD)
resources.defer(lambda: client.proxy.delete_key(new_key))
@ -387,7 +465,7 @@ class TestKeyRegeneration:
assert new_key != old_key, "regenerate returned the same key string, so no rotation happened"
def old_accepted() -> bool | None:
outcome = client.chat_status(old_key, "gpt-5.5", f"say hi {unique_marker()}")
outcome = client.chat_status(old_key, OPENAI_MODEL, f"say hi {unique_marker()}")
return True if outcome.ok else None
_ = _poll(client, old_accepted, "old key was rejected 401 inside its grace period at the deadline")
@ -396,7 +474,7 @@ class TestKeyRegeneration:
)
def old_rejected() -> bool | None:
outcome = client.chat_status(old_key, "gpt-5.5", f"say hi {unique_marker()}")
outcome = client.chat_status(old_key, OPENAI_MODEL, f"say hi {unique_marker()}")
return True if outcome.status_code == 401 else None
_ = _poll(
@ -408,15 +486,21 @@ class TestKeyRegeneration:
class TestTeamRoutes:
@pytest.mark.covers("mgmt.team.new.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TEAM_MANAGEMENT,
)
)
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"])
team_id = _create_team(client, resources, alias, [GEMINI_MODEL])
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"], (
assert info.models == [GEMINI_MODEL], (
f"/team/info reports models {info.models}, configured ['gemini-2.5-flash']"
)
@ -427,8 +511,14 @@ class TestTeamRoutes:
)
@pytest.mark.covers("mgmt.team.update.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TEAM_MANAGEMENT,
)
)
def test_update_persists_to_team_info(self, client: ManagementClient, resources: ResourceManager) -> None:
team_id = _create_team(client, resources, f"e2e-mgmt-team-{unique_marker()}", ["gemini-2.5-flash"])
team_id = _create_team(client, resources, f"e2e-mgmt-team-{unique_marker()}", [GEMINI_MODEL])
updated_alias = f"e2e-mgmt-team-updated-{unique_marker()}"
client.update_team(TeamUpdateBody(team_id=team_id, team_alias=updated_alias))
@ -438,11 +528,17 @@ class TestTeamRoutes:
_ = _poll(client, reflected, f"/team/info never reflected team_alias {updated_alias!r} after /team/update")
@pytest.mark.covers("mgmt.team.list.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TEAM_MANAGEMENT,
)
)
def test_created_team_appears_in_team_list(
self, client: ManagementClient, resources: ResourceManager
) -> None:
alias = f"e2e-mgmt-team-{unique_marker()}"
team_id = _create_team(client, resources, alias, ["gemini-2.5-flash"])
team_id = _create_team(client, resources, alias, [GEMINI_MODEL])
_ = _poll(
client,
@ -451,17 +547,26 @@ class TestTeamRoutes:
)
@pytest.mark.covers("mgmt.team.delete.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TEAM_MANAGEMENT,
providers=(Provider.OPENAI,),
models=(OPENAI_MODEL,),
mode=Mode.NONSTREAM,
)
)
def test_delete_persists_and_revokes_team_bound_key(
self, client: ManagementClient, resources: ResourceManager
) -> None:
"""The teardown's deferred delete_team/delete_key fire again on the already-
deleted team and key by design: both are warn-only no-ops, and the deferred
cleanup must survive this test failing before the in-body delete."""
team_id = _create_team(client, resources, f"e2e-mgmt-team-{unique_marker()}", ["gpt-5.5"])
team_id = _create_team(client, resources, f"e2e-mgmt-team-{unique_marker()}", [OPENAI_MODEL])
key = _generate_key(client, resources, KeyGenerateBody(team_id=team_id))
def accepted() -> bool | None:
outcome = client.chat_status(key, "gpt-5.5", f"say hi {unique_marker()}")
outcome = client.chat_status(key, OPENAI_MODEL, f"say hi {unique_marker()}")
return True if outcome.status_code != 401 else None
_ = _poll(client, accepted, "team-bound key was never accepted at auth before team deletion")
@ -474,7 +579,7 @@ class TestTeamRoutes:
)
def rejected() -> bool | None:
outcome = client.chat_status(key, "gpt-5.5", f"say hi {unique_marker()}")
outcome = client.chat_status(key, OPENAI_MODEL, f"say hi {unique_marker()}")
return True if outcome.status_code == 401 else None
_ = _poll(
@ -482,6 +587,12 @@ class TestTeamRoutes:
)
@pytest.mark.covers("mgmt.team.delete.membership_larger_than_db_pool")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TEAM_MANAGEMENT,
)
)
def test_team_delete_succeeds_for_team_larger_than_db_pool(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -519,6 +630,12 @@ class TestTeamRoutes:
)
@pytest.mark.covers("mgmt.team.member_add.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TEAM_MANAGEMENT,
)
)
def test_member_add_and_delete_persist_to_team_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -527,7 +644,7 @@ class TestTeamRoutes:
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"])
team_id = _create_team(client, resources, f"e2e-mgmt-team-{unique_marker()}", [GEMINI_MODEL])
client.add_team_member(team_id, user_id)
member = next(
@ -545,6 +662,12 @@ class TestTeamRoutes:
class TestUserRoutes:
@pytest.mark.covers("mgmt.user.new.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.USER_MANAGEMENT,
)
)
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"))
@ -556,6 +679,12 @@ class TestUserRoutes:
)
@pytest.mark.covers("mgmt.user.update.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.USER_MANAGEMENT,
)
)
def test_update_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"))
@ -572,6 +701,12 @@ class TestUserRoutes:
f"/user/info reports user_role {info.user_role!r} after /user/update to 'internal_user_viewer'"
)
@pytest.mark.covers("mgmt.user.delete.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.USER_MANAGEMENT,
)
)
def test_delete_removes_the_user_from_inventory(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -594,6 +729,12 @@ class TestUserRoutes:
_ = _poll(client, removed, f"user {user_id} still present in /user/list after /user/delete at the deadline")
@pytest.mark.covers("mgmt.user.list.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.USER_MANAGEMENT,
)
)
def test_created_users_appear_in_user_list(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -616,22 +757,34 @@ class TestUserRoutes:
class TestOrganizationRoutes:
@pytest.mark.covers("mgmt.organization.new.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.ORGANIZATION_MANAGEMENT,
)
)
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"]))
org_id = client.create_org(OrgNewBody(organization_alias=alias, models=[GEMINI_MODEL]))
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"], (
assert info.models == [GEMINI_MODEL], (
f"/organization/info reports models {info.models}, configured ['gemini-2.5-flash']"
)
@pytest.mark.covers("mgmt.organization.update.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.ORGANIZATION_MANAGEMENT,
)
)
def test_update_alias_persists_to_organization_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -650,6 +803,12 @@ class TestOrganizationRoutes:
)
@pytest.mark.covers("mgmt.organization.delete.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.ORGANIZATION_MANAGEMENT,
)
)
def test_delete_removes_from_organization_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -674,6 +833,12 @@ class TestOrganizationRoutes:
class TestTagRoutes:
@pytest.mark.covers("mgmt.tag.new.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TAG_MANAGEMENT,
)
)
def test_new_persists_to_tag_list(self, client: ManagementClient, resources: ResourceManager) -> None:
name = f"e2e-mgmt-tag-{unique_marker()}"
description = "Tag for spend categorization"
@ -704,6 +869,12 @@ def _model_entry(client: ManagementClient, model_name: str) -> ModelInfoEntry |
class TestModelRoutes:
@pytest.mark.covers("mgmt.model.update.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.MODEL_MANAGEMENT,
)
)
def test_update_persists_input_cost_to_model_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -747,6 +918,12 @@ class TestModelRoutes:
)
@pytest.mark.covers("mgmt.model.delete.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.MODEL_MANAGEMENT,
)
)
def test_delete_removes_from_model_info_catalog(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -769,6 +946,12 @@ class TestModelRoutes:
_ = _poll(client, absent, f"{model_name} still present in /model/info after /model/delete at the deadline")
@pytest.mark.covers("mgmt.model.add.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.MODEL_MANAGEMENT,
)
)
def test_new_persists_to_model_info_catalog(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -797,6 +980,11 @@ def _assert_route_forbidden(route: str, outcome: StreamingResponse) -> None:
class TestManagementRoutePermissions:
@pytest.mark.covers("other.auth.virtual_key.route_permission_enforced")
@meta(
Subject(
domain=Domain.PROXY_AUTH,
)
)
def test_llm_only_key_forbidden_from_management_writes(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -832,6 +1020,12 @@ class TestManagementRoutePermissions:
class TestCustomer:
@pytest.mark.covers("mgmt.end_user.new.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.CUSTOMER_MANAGEMENT,
)
)
def test_customer_create_persists_to_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -896,6 +1090,12 @@ def _generate_response(
class TestKeyDeletionAuditLog:
@pytest.mark.covers("mgmt.key.delete.audit_logged")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
)
)
def test_key_delete_by_key_writes_audit_row(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -908,6 +1108,12 @@ class TestKeyDeletionAuditLog:
_assert_single_deleted_row(_await_deleted_audit_rows(client, token), token)
@pytest.mark.covers("mgmt.key.delete.audit_logged")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.KEY_MANAGEMENT,
)
)
def test_key_delete_by_alias_writes_audit_row(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -921,6 +1127,12 @@ class TestKeyDeletionAuditLog:
_assert_single_deleted_row(_await_deleted_audit_rows(client, token), token)
@pytest.mark.covers("mgmt.team.member_delete.audit_logs_keys")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TEAM_MANAGEMENT,
)
)
def test_team_member_delete_writes_audit_row_for_member_keys(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -940,6 +1152,12 @@ class TestKeyDeletionAuditLog:
_assert_single_deleted_row(_await_deleted_audit_rows(client, token), token)
@pytest.mark.covers("mgmt.team.delete.audit_logs_keys")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TEAM_MANAGEMENT,
)
)
def test_team_delete_writes_audit_row_for_team_keys(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -953,6 +1171,12 @@ class TestKeyDeletionAuditLog:
_assert_single_deleted_row(_await_deleted_audit_rows(client, token), token)
@pytest.mark.covers("mgmt.user.delete.audit_logs_keys")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.USER_MANAGEMENT,
)
)
def test_user_delete_writes_audit_row_for_user_keys(
self, client: ManagementClient, resources: ResourceManager
) -> None:

61
tests/e2e/management/test_mcp_lifecycle_e2e.py
View file

@ -19,6 +19,7 @@ from typing import Final
import pytest
from e2e_config import unique_marker
from e2e_http import unwrap
from e2e_metadata import Domain, Route, Subject, meta
from lifecycle import ResourceManager
from management_client import ManagementClient
from models import (
@ -88,6 +89,12 @@ def _listed_server_everywhere(client: ManagementClient, server_id: str) -> Mappi
class TestMcpServerLifecycle:
@pytest.mark.covers("mgmt.mcp_server.new.persists")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_create_persists_every_field_on_every_replica(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -107,6 +114,12 @@ class TestMcpServerLifecycle:
)
)
@pytest.mark.covers("mgmt.mcp_server.list.persists")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_created_server_is_listed_with_every_field(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -116,6 +129,12 @@ class TestMcpServerLifecycle:
_assert_server_matches(row, body, where=f"GET /v1/mcp/server on {replica}")
@pytest.mark.covers("mgmt.mcp_server.update.preserves_unrelated_fields")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_updating_only_the_alias_keeps_every_other_field_on_every_replica(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -133,6 +152,12 @@ class TestMcpServerLifecycle:
)
@pytest.mark.covers("mgmt.mcp_server.update.clear_persists")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_clearing_the_description_with_null_reads_back_null(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -149,6 +174,12 @@ class TestMcpServerLifecycle:
)
@pytest.mark.covers("mgmt.mcp_server.delete.persists")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_delete_removes_the_server_from_every_replica(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -199,6 +230,12 @@ def _toolset_everywhere(
class TestMcpToolsetLifecycle:
@pytest.mark.covers("mgmt.mcp_toolset.new.persists")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_create_persists_both_tools_under_the_exact_names_written(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -221,6 +258,12 @@ class TestMcpToolsetLifecycle:
)
@pytest.mark.covers("mgmt.mcp_toolset.update.preserves_unrelated_fields")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_updating_only_the_description_keeps_the_tools_and_name(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -238,6 +281,12 @@ class TestMcpToolsetLifecycle:
)
@pytest.mark.covers("mgmt.mcp_toolset.update.persists")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_updating_the_tools_to_one_entry_reads_back_exactly_that_entry(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -256,6 +305,12 @@ class TestMcpToolsetLifecycle:
)
@pytest.mark.covers("mgmt.mcp_toolset.update.clear_persists")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_clearing_the_description_with_null_reads_back_null(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -273,6 +328,12 @@ class TestMcpToolsetLifecycle:
)
@pytest.mark.covers("mgmt.mcp_toolset.delete.persists")
@meta(
Subject(
domain=Domain.MCP,
route=Route.MCP,
)
)
def test_delete_removes_the_toolset_from_every_replica(
self, client: ManagementClient, resources: ResourceManager
) -> None:

37
tests/e2e/management/test_model_tag_accessgroup_e2e.py
View file

@ -22,6 +22,7 @@ from pydantic import BaseModel, ConfigDict, RootModel
from e2e_config import unique_marker
from e2e_http import NoBody, unwrap
from e2e_metadata import Domain, Route, Subject, meta
from lifecycle import ResourceManager
from management_client import ManagementClient
from models import KeyGenerateBody, LiteLLMParamsBody, ModelInfoBody, ModelNewBody
@ -216,6 +217,12 @@ def _model_blocked_flag(client: ManagementClient, model_id: str) -> bool | None:
class TestModelRoutes:
@pytest.mark.covers("mgmt.model.add.admin_only")
@meta(
Subject(
domain=Domain.PROXY_AUTH,
route=Route.MODEL_MANAGEMENT,
)
)
def test_non_admin_key_cannot_add_global_model(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -248,6 +255,12 @@ class TestModelRoutes:
)
@pytest.mark.covers("mgmt.model.block.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.MODEL_MANAGEMENT,
)
)
def test_block_then_unblock_persists_to_model_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -281,6 +294,12 @@ class TestModelRoutes:
class TestTagRoutes:
@pytest.mark.covers("mgmt.tag.list.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TAG_MANAGEMENT,
)
)
def test_tag_list_reports_created_tag(self, client: ManagementClient, resources: ResourceManager) -> None:
name = f"e2e-mgmt-tag-{unique_marker()}"
description = "coverage: tag inventory"
@ -301,6 +320,12 @@ class TestTagRoutes:
)
@pytest.mark.covers("mgmt.tag.delete.persists")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.TAG_MANAGEMENT,
)
)
def test_tag_delete_removes_from_list(self, client: ManagementClient, resources: ResourceManager) -> None:
"""The teardown's deferred delete fires again on the already-deleted tag by
design: it is the safety net if this test fails before the in-body delete,
@ -326,6 +351,12 @@ class TestTagRoutes:
class TestModelAccessGroupRoutes:
@pytest.mark.covers("mgmt.access_group.new.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.MODEL_MANAGEMENT,
)
)
def test_new_access_group_tags_the_deployment(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -356,6 +387,12 @@ class TestModelAccessGroupRoutes:
)
@pytest.mark.covers("mgmt.access_group.info.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.MODEL_MANAGEMENT,
)
)
def test_access_group_info_reports_membership(
self, client: ManagementClient, resources: ResourceManager
) -> None:

10
tests/e2e/management/test_model_test_connection_e2e.py
View file

@ -23,6 +23,7 @@ import time
import pytest
from e2e_http import unwrap
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from management_client import ManagementClient
from models import ConnectionTestBody, ConnectionTestResponse, LiteLLMParamsBody
@ -50,6 +51,15 @@ def _probe_mantle(client: ManagementClient) -> ConnectionTestResponse:
class TestModelTestConnection:
@pytest.mark.covers("mgmt.model.test_connection.happy_path")
@meta(
Subject(
domain=Domain.MANAGEMENT,
route=Route.HEALTH,
providers=(Provider.BEDROCK_MANTLE,),
models=(MANTLE_RESPONSES_BACKEND,),
mode=Mode.NONSTREAM,
)
)
def test_bedrock_mantle_responses_connection_succeeds(self, client: ManagementClient) -> None:
for attempt in range(1, PROBE_ATTEMPTS + 1):
response = _probe_mantle(client)

15
tests/e2e/management/test_team_management_e2e.py
View file

@ -27,6 +27,7 @@ import pytest
from pydantic import BaseModel
from e2e_config import settle_propagation, unique_marker
from e2e_metadata import Domain, Route, Subject, meta
from e2e_http import NoBody, PartialBody, StreamingResponse, unwrap
from lifecycle import ResourceManager
from management_client import ManagementClient
@ -241,6 +242,7 @@ def _member_delete_status(client: ManagementClient, key: str, team_id: str, user
class TestTeamManagementRoutes:
@pytest.mark.covers("mgmt.team.info.happy_path")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.TEAM_MANAGEMENT))
def test_info_returns_created_team_fields(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -257,6 +259,7 @@ class TestTeamManagementRoutes:
)
@pytest.mark.covers("mgmt.team.block.persists")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.TEAM_MANAGEMENT))
def test_block_then_unblock_persists_to_team_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -278,6 +281,7 @@ class TestTeamManagementRoutes:
)
@pytest.mark.covers("mgmt.team.member_update.persists")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.TEAM_MANAGEMENT))
def test_member_update_persists_role_and_budget(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -302,6 +306,7 @@ class TestTeamManagementRoutes:
)
@pytest.mark.covers("mgmt.team.member_delete.persists")
@meta(Subject(domain=Domain.MANAGEMENT, route=Route.TEAM_MANAGEMENT))
def test_member_delete_persists_to_team_info(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -320,6 +325,7 @@ class TestTeamManagementRoutes:
)
@pytest.mark.covers("mgmt.team.new.admin_only")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_new_is_denied_to_non_admin_keys(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -334,6 +340,7 @@ class TestTeamManagementRoutes:
)
@pytest.mark.covers("mgmt.team.member_add.member_forbidden")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_member_add_forbidden_to_plain_member(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -348,6 +355,7 @@ class TestTeamManagementRoutes:
)
@pytest.mark.covers("mgmt.team.member_delete.member_forbidden")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_member_delete_forbidden_to_plain_member(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -458,6 +466,7 @@ class TestTeamAdminWithNoEditableFields:
"""No proxy admin has enabled a team field for team admins, which is how every proxy starts."""
@pytest.mark.covers("mgmt.team.update.team_admin_forbidden_until_enabled")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_team_admin_cannot_change_any_team_setting(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -483,6 +492,7 @@ class TestTeamAdminWithTpmLimitEnabled:
"""A proxy admin has enabled tpm_limit, so a team admin may change that setting and no other."""
@pytest.mark.covers("mgmt.team.update.team_admin_limited_to_enabled_fields")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_team_admin_saves_the_settings_form_with_a_new_tpm_limit(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -522,6 +532,7 @@ class TestTeamAdminWithTpmLimitEnabled:
pytest.param(TeamSettingsChange(metadata=TeamCustomMetadata(cost_center="team-admin")), id="metadata"),
],
)
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_team_admin_cannot_change_a_setting_that_is_not_enabled(
self, client: ManagementClient, resources: ResourceManager, change: TeamSettingsChange
) -> None:
@ -548,6 +559,7 @@ class TestTeamAdminWithTpmLimitEnabled:
)
@pytest.mark.covers("mgmt.team.update.team_admin_resend_keeps_budget_reset")
@meta(Subject(domain=Domain.SPEND_BUDGETS, route=Route.TEAM_MANAGEMENT))
def test_team_admin_resending_the_budget_settings_keeps_the_next_budget_reset(
self, client: ManagementClient, resources: ResourceManager
) -> None:
@ -613,6 +625,7 @@ class TestTeamAdminWithRpmLimitAndMaxBudgetEnabled:
"current_budget",
[pytest.param(_TEAM_MAX_BUDGET, id="lower"), pytest.param(None, id="first-budget")],
)
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_team_admin_saves_a_new_rpm_limit_and_a_tighter_budget(
self, client: ManagementClient, resources: ResourceManager, current_budget: float | None
) -> None:
@ -649,6 +662,7 @@ class TestTeamAdminWithRpmLimitAndMaxBudgetEnabled:
pytest.param(None, "Only a proxy admin can remove", id="remove"),
],
)
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_team_admin_cannot_raise_or_remove_the_budget(
self, client: ManagementClient, resources: ResourceManager, max_budget: float | None, refusal: str
) -> None:
@ -670,6 +684,7 @@ class TestTeamAdminWithRpmLimitAndMaxBudgetEnabled:
)
@pytest.mark.covers("mgmt.team.update.team_admin_cannot_grow_budget")
@meta(Subject(domain=Domain.PROXY_AUTH, route=Route.TEAM_MANAGEMENT))
def test_team_admin_cannot_raise_an_org_team_budget_under_the_org_cap(
self, client: ManagementClient, resources: ResourceManager
) -> None: