From 0f0c1bc80b64afa387dc06bd6c0761678e4c9447 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 14 Oct 2026 06:29:20 +0000 Subject: [PATCH] test(integration): cover teams endpoint contracts (#44705) * test(integration): cover teams endpoint contracts Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): address review on teams contract tests Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): drain shared upstream observations at fixture setup Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): prove team block, member and bulk permission contracts at runtime Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): assert email admin adds and whole membership rows for team members Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): drive key update grant and revoke, Z-suffixed temp budget expiry, link LIT-9257 Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): close review gaps in teams endpoint contracts Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): harden team member endpoint contracts Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: mateo Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../test_team_member_permissions.py | 331 ++++++- .../authorization/test_team_scoped_models.py | 149 ++- .../integration/management/test_team_block.py | 226 +++++ .../management/test_team_member_lifecycle.py | 925 +++++++++++++++++ .../spend/test_team_member_spend.py | 932 +++++++++++++++++- 5 files changed, 2557 insertions(+), 6 deletions(-) create mode 100644 tests/integration/management/test_team_block.py create mode 100644 tests/integration/management/test_team_member_lifecycle.py diff --git a/tests/integration/authorization/test_team_member_permissions.py b/tests/integration/authorization/test_team_member_permissions.py index 24ad98b486d..5d29aceb10c 100644 --- a/tests/integration/authorization/test_team_member_permissions.py +++ b/tests/integration/authorization/test_team_member_permissions.py @@ -1,15 +1,81 @@ +import json +from collections.abc import Iterator from dataclasses import dataclass from hashlib import sha256 from typing import Final import httpx -from integration._support.client import Gateway, Scenario, object_value, string_value -from integration._support.database import read_rows -from pydantic import JsonValue +import pytest +from integration._support.client import ( + Gateway, + Scenario, + delete_key_if_present, + object_value, + string_value, +) +from integration._support.database import read_rows, write_rows +from pydantic import BaseModel, JsonValue PERMISSION_ERROR: Final = "team_member_permission_error" +class _ObservedRequest(BaseModel): + body: dict[str, JsonValue] + + +class _Observations(BaseModel): + requests: list[_ObservedRequest] + + +class _TeamPermissions(BaseModel): + team_id: str + team_member_permissions: list[str] | None = None + + +class _PermissionsList(BaseModel): + team_id: str + team_member_permissions: list[str] + + +class _TeamPermissionRow(BaseModel): + team_id: str + team_member_permissions: list[str] | None = None + + +class _BulkPermissionsUpdate(BaseModel): + message: str + teams_updated: int + permissions_appended: list[str] | None = None + + +class _GeneratedKey(BaseModel): + key: str + + +class _Error(BaseModel): + type: str + + +class _ErrorResponse(BaseModel): + error: _Error + + +class _HTTPErrorDetail(BaseModel): + error: str + + +class _HTTPErrorResponse(BaseModel): + detail: _HTTPErrorDetail + + +@pytest.fixture +def upstream(gateway: Gateway) -> Iterator[httpx.Client]: + with httpx.Client(base_url=gateway.upstream_url, timeout=15, trust_env=False) as client: + response: Final = client.get("/__observations") + assert response.status_code == 200, response.text + yield client + + @dataclass(frozen=True, slots=True) class Member: team_id: str @@ -31,6 +97,15 @@ def _team_key_row(key: str) -> list[dict[str, JsonValue]]: ) +def _update_team_key(proxy: Gateway, member: Member, owner: str) -> httpx.Response: + return proxy.request( + "POST", + "/key/update", + {"key": member.team_key, "team_id": member.team_id, "metadata": {"owner": owner}}, + key=member.member_key, + ) + + def _team_key_count(team_id: str) -> int: return len(read_rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE team_id = %s', (team_id,))) @@ -41,6 +116,256 @@ def _refused(response: httpx.Response, status: int, error_type: str | None = Non assert object_value(response.json()["error"])["type"] == error_type, response.text +def _observed_models(upstream: httpx.Client) -> tuple[str, ...]: + response: Final = upstream.get("/__observations") + assert response.status_code == 200, response.text + observations: Final = _Observations.model_validate_json(response.text) + return tuple(string_value(request.body["model"]) for request in observations.requests) + + +def _permissions(gateway: Gateway, team_id: str) -> _PermissionsList: + response: Final = gateway.request("GET", "/team/permissions_list", params={"team_id": team_id}) + assert response.status_code == 200, response.text + return _PermissionsList.model_validate_json(response.text) + + +def _team_permission_rows(team_ids: tuple[str, ...]) -> tuple[_TeamPermissionRow, ...]: + return tuple( + _TeamPermissionRow.model_validate(row) + for row in read_rows( + 'SELECT team_id, team_member_permissions FROM "LiteLLM_TeamTable" WHERE team_id = ANY(%s) ORDER BY team_id', + (list(team_ids),), + ) + ) + + +def _restore_team_permissions(team_id: str, permissions: list[str] | None) -> None: + if permissions is None: + write_rows( + 'UPDATE "LiteLLM_TeamTable" SET team_member_permissions = NULL WHERE team_id = %s', + (team_id,), + ) + return + write_rows( + 'UPDATE "LiteLLM_TeamTable" SET team_member_permissions = ARRAY(SELECT jsonb_array_elements_text(%s::jsonb)) WHERE team_id = %s', + (json.dumps(permissions), team_id), + ) + + +def test_permissions_update_persists_requested_member_permissions(gateway: Gateway, upstream: httpx.Client) -> None: + with gateway.scenario() as scenario: + team: Final = scenario.team() + requested: Final = ["/key/generate", "/key/info"] + updated: Final = gateway.request( + "POST", + "/team/permissions_update", + {"team_id": team, "team_member_permissions": requested}, + ) + assert updated.status_code == 200, updated.text + response: Final = _TeamPermissions.model_validate_json(updated.text) + assert response.team_id == team, updated.text + assert response.team_member_permissions == requested, updated.text + readback: Final = _permissions(gateway, team) + assert readback.team_member_permissions == requested, repr(readback) + assert read_rows( + 'SELECT team_member_permissions FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [{"team_member_permissions": requested}], updated.text + observed: Final = _observed_models(upstream) + assert observed == (), repr(observed) + + +def test_permissions_update_grants_and_revokes_key_generate_and_key_update_for_a_warmed_member_on_both_proxies( + gateway: Gateway, peer: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + team: Final = scenario.team() + member: Final = scenario.member(team) + member_key: Final = scenario.key(user_id=member) + team_key: Final = scenario.key(team_id=team, metadata={"owner": "team"}) + holder: Final = Member(team, team_key, member_key) + denied_gateway: Final = gateway.request("POST", "/key/generate", {"team_id": team}, key=member_key) + _refused(denied_gateway, 401, PERMISSION_ERROR) + denied_peer: Final = peer.request("POST", "/key/generate", {"team_id": team}, key=member_key) + _refused(denied_peer, 401, PERMISSION_ERROR) + _refused(_update_team_key(gateway, holder, "member before grant"), 401, PERMISSION_ERROR) + _refused(_update_team_key(peer, holder, "member before grant"), 401, PERMISSION_ERROR) + assert _team_key_row(team_key) == [{"team_id": team, "metadata": {"owner": "team"}}] + + granted_permissions: Final = ["/key/generate", "/key/info", "/key/update"] + granted: Final = gateway.request( + "POST", + "/team/permissions_update", + {"team_id": team, "team_member_permissions": granted_permissions}, + ) + assert granted.status_code == 200, granted.text + added: Final = _permissions(gateway, team) + assert added.team_member_permissions == granted_permissions, repr(added) + assert read_rows( + 'SELECT team_member_permissions FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [{"team_member_permissions": granted_permissions}], granted.text + generated_gateway: Final = gateway.request("POST", "/key/generate", {"team_id": team}, key=member_key) + assert generated_gateway.status_code == 200, generated_gateway.text + gateway_key: Final = _GeneratedKey.model_validate_json(generated_gateway.text).key + scenario.cleanups.callback(delete_key_if_present, gateway, gateway_key) + generated_peer: Final = peer.request("POST", "/key/generate", {"team_id": team}, key=member_key) + assert generated_peer.status_code == 200, generated_peer.text + peer_key: Final = _GeneratedKey.model_validate_json(generated_peer.text).key + scenario.cleanups.callback(delete_key_if_present, gateway, peer_key) + updated_gateway: Final = _update_team_key(gateway, holder, "member via gateway") + assert updated_gateway.status_code == 200, updated_gateway.text + assert _team_key_row(team_key) == [{"team_id": team, "metadata": {"owner": "member via gateway"}}] + updated_peer: Final = _update_team_key(peer, holder, "member via peer") + assert updated_peer.status_code == 200, updated_peer.text + assert _team_key_row(team_key) == [{"team_id": team, "metadata": {"owner": "member via peer"}}] + + revoked_permissions: Final = ["/key/info"] + revoked: Final = gateway.request( + "POST", + "/team/permissions_update", + {"team_id": team, "team_member_permissions": revoked_permissions}, + ) + assert revoked.status_code == 200, revoked.text + revoked_readback: Final = _permissions(gateway, team) + assert revoked_readback.team_member_permissions == revoked_permissions, repr(revoked_readback) + assert read_rows( + 'SELECT team_member_permissions FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [{"team_member_permissions": revoked_permissions}], revoked.text + _refused(_update_team_key(gateway, holder, "member after revoke"), 401, PERMISSION_ERROR) + _refused(_update_team_key(peer, holder, "member after revoke"), 401, PERMISSION_ERROR) + assert _team_key_row(team_key) == [{"team_id": team, "metadata": {"owner": "member via peer"}}] + revoked_gateway: Final = gateway.request("POST", "/key/generate", {"team_id": team}, key=member_key) + _refused(revoked_gateway, 401, PERMISSION_ERROR) + revoked_peer: Final = peer.request("POST", "/key/generate", {"team_id": team}, key=member_key) + _refused(revoked_peer, 401, PERMISSION_ERROR) + observed: Final = _observed_models(upstream) + assert observed == (), repr(observed) + + +def test_permissions_bulk_update_appends_only_to_the_listed_teams( + gateway: Gateway, peer: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + listed: Final = _member(scenario, ["/key/info"]) + team_two: Final = scenario.team(team_member_permissions=["/key/info", "/key/update"]) + team_three_permission: Final = "/team/daily/activity" + team_three: Final = _member(scenario, [team_three_permission]) + unlisted: Final = _member(scenario, ["/key/info"]) + team_ids: Final = (listed.team_id, team_two, team_three.team_id, unlisted.team_id) + for proxy in (gateway, peer): + for member in (listed, team_three, unlisted): + _refused(_update_team_key(proxy, member, "member before bulk"), 401, PERMISSION_ERROR) + before_non_admin: Final = _team_permission_rows(team_ids) + caller: Final = scenario.user(user_role="internal_user") + caller_key: Final = scenario.key(user_id=caller, allowed_routes=["/team/permissions_bulk_update"]) + denied: Final = gateway.request( + "POST", + "/team/permissions_bulk_update", + {"permissions": ["/key/update"], "team_ids": [listed.team_id, team_two]}, + key=caller_key, + ) + assert denied.status_code == 403, denied.text + assert _HTTPErrorResponse.model_validate_json(denied.text) == _HTTPErrorResponse( + detail=_HTTPErrorDetail(error="Only proxy admins can bulk-update team permissions") + ), denied.text + after_denied: Final = _team_permission_rows(team_ids) + assert after_denied == before_non_admin, repr(after_denied) + + selected: Final = gateway.request( + "POST", + "/team/permissions_bulk_update", + {"permissions": ["/key/update"], "team_ids": [listed.team_id, team_two, team_three.team_id]}, + ) + assert selected.status_code == 200, selected.text + selected_response: Final = _BulkPermissionsUpdate.model_validate_json(selected.text) + assert selected_response == _BulkPermissionsUpdate( + message="Team permissions updated successfully", + teams_updated=2, + permissions_appended=["/key/update"], + ), selected.text + selected_rows: Final = _team_permission_rows(team_ids) + assert tuple(sorted(selected_rows, key=lambda row: row.team_id)) == tuple( + sorted( + ( + _TeamPermissionRow(team_id=listed.team_id, team_member_permissions=["/key/info", "/key/update"]), + _TeamPermissionRow(team_id=team_two, team_member_permissions=["/key/info", "/key/update"]), + _TeamPermissionRow( + team_id=team_three.team_id, + team_member_permissions=["/key/update", team_three_permission], + ), + _TeamPermissionRow(team_id=unlisted.team_id, team_member_permissions=["/key/info"]), + ), + key=lambda row: row.team_id, + ) + ), repr(selected_rows) + listed_readback: Final = _permissions(gateway, listed.team_id) + assert listed_readback.team_member_permissions == ["/key/info", "/key/update"], repr(listed_readback) + team_three_readback: Final = _permissions(gateway, team_three.team_id) + assert team_three_readback.team_member_permissions == ["/key/update", team_three_permission], repr( + team_three_readback + ) + unlisted_readback: Final = _permissions(gateway, unlisted.team_id) + assert unlisted_readback.team_member_permissions == ["/key/info"], repr(unlisted_readback) + + gateway_update: Final = _update_team_key(gateway, listed, "member via gateway") + assert gateway_update.status_code == 200, gateway_update.text + assert _team_key_row(listed.team_key) == [ + {"team_id": listed.team_id, "metadata": {"owner": "member via gateway"}} + ], gateway_update.text + peer_update: Final = _update_team_key(peer, listed, "member via peer") + assert peer_update.status_code == 200, peer_update.text + assert _team_key_row(listed.team_key) == [ + {"team_id": listed.team_id, "metadata": {"owner": "member via peer"}} + ], peer_update.text + for proxy in (gateway, peer): + _refused(_update_team_key(proxy, unlisted, "unlisted member"), 401, PERMISSION_ERROR) + assert _team_key_row(unlisted.team_key) == [{"team_id": unlisted.team_id, "metadata": {"owner": "team"}}] + + all_before: Final = tuple( + _TeamPermissionRow.model_validate(row) + for row in read_rows( + 'SELECT team_id, team_member_permissions FROM "LiteLLM_TeamTable" ORDER BY team_id', + (), + ) + ) + for row in all_before: + if row.team_id not in team_ids and "/key/update" not in (row.team_member_permissions or []): + scenario.cleanups.callback(_restore_team_permissions, row.team_id, row.team_member_permissions) + missing: Final = tuple( + row.team_id for row in all_before if "/key/update" not in (row.team_member_permissions or []) + ) + all_updated: Final = gateway.request( + "POST", + "/team/permissions_bulk_update", + {"permissions": ["/key/update"], "apply_to_all_teams": True}, + ) + assert all_updated.status_code == 200, all_updated.text + all_response: Final = _BulkPermissionsUpdate.model_validate_json(all_updated.text) + assert all_response == _BulkPermissionsUpdate( + message="Team permissions updated successfully", + teams_updated=len(missing), + permissions_appended=["/key/update"], + ), all_updated.text + all_after: Final = tuple( + _TeamPermissionRow.model_validate(row) + for row in read_rows( + 'SELECT team_id, team_member_permissions FROM "LiteLLM_TeamTable" ORDER BY team_id', + (), + ) + ) + assert tuple(row.team_id for row in all_after) == tuple(row.team_id for row in all_before), repr(all_after) + assert all("/key/update" in (row.team_member_permissions or []) for row in all_after), repr(all_after) + unlisted_after_all: Final = _update_team_key(peer, unlisted, "unlisted member after all teams") + assert unlisted_after_all.status_code == 200, unlisted_after_all.text + assert _team_key_row(unlisted.team_key) == [ + {"team_id": unlisted.team_id, "metadata": {"owner": "unlisted member after all teams"}} + ], unlisted_after_all.text + observed: Final = _observed_models(upstream) + assert observed == (), repr(observed) + + def test_default_member_permissions_only_allow_reading_team_keys(gateway: Gateway) -> None: with gateway.scenario() as scenario: member: Final = _member(scenario, None) diff --git a/tests/integration/authorization/test_team_scoped_models.py b/tests/integration/authorization/test_team_scoped_models.py index 1323c1ee004..2982fbdb4eb 100644 --- a/tests/integration/authorization/test_team_scoped_models.py +++ b/tests/integration/authorization/test_team_scoped_models.py @@ -4,8 +4,38 @@ from typing import Final import httpx import pytest -from integration._support.client import Gateway, object_value, string_value -from pydantic import JsonValue +from integration._support.client import Gateway, Scenario, object_value, string_value +from integration._support.database import read_rows +from pydantic import BaseModel, JsonValue + + +class _ObservedRequest(BaseModel): + body: dict[str, JsonValue] + + +class _Observations(BaseModel): + requests: list[_ObservedRequest] + + +class _TeamState(BaseModel): + team_id: str + models: list[str] | None = None + + +class _TeamModelRow(BaseModel): + models: list[str] | None = None + + +class _TeamInfo(BaseModel): + team_info: _TeamState + + +class _ProxyError(BaseModel): + type: str + + +class _ErrorResponse(BaseModel): + error: _ProxyError @pytest.fixture @@ -27,6 +57,13 @@ def _calls_to(observed: list[JsonValue], provider_model: str) -> int: return sum(object_value(object_value(request)["body"]).get("model") == provider_model for request in observed) +def _observed_provider_models(upstream: httpx.Client) -> tuple[str, ...]: + observed: Final = upstream.get("/__observations") + assert observed.status_code == 200, observed.text + response: Final = _Observations.model_validate_json(observed.text) + return tuple(string_value(request.body["model"]) for request in response.requests) + + def _chat(gateway: Gateway, model: str, key: str) -> httpx.Response: return gateway.request( "POST", @@ -36,6 +73,28 @@ def _chat(gateway: Gateway, model: str, key: str) -> httpx.Response: ) +def _team_model(scenario: Scenario) -> tuple[str, str]: + provider_model: Final = f"team-model-{uuid.uuid4().hex}" + return scenario.model(model=f"openai/{provider_model}"), provider_model + + +def _team_info(gateway: Gateway, team_id: str) -> _TeamInfo: + response: Final = gateway.request("GET", "/team/info", params={"team_id": team_id}) + assert response.status_code == 200, response.text + return _TeamInfo.model_validate_json(response.text) + + +def _stored_team_models(team_id: str) -> _TeamModelRow: + rows: Final = read_rows('SELECT models FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team_id,)) + assert len(rows) == 1, repr(rows) + return _TeamModelRow.model_validate(rows[0]) + + +def _model_denied(response: httpx.Response) -> None: + assert response.status_code == 403, response.text + assert _ErrorResponse.model_validate_json(response.text).error.type == "team_model_access_denied", response.text + + def _ids(listing: dict[str, JsonValue]) -> set[JsonValue]: data: Final = listing["data"] assert isinstance(data, list) @@ -112,3 +171,89 @@ def test_team_model_alias_routes_a_team_key_to_its_target( assert unaliased.json()["error"]["type"] == "key_model_access_denied" after_refusal: Final = _observed_requests(upstream) assert _calls_to(after_refusal, provider_model) == 0, after_refusal + + +def test_team_model_add_and_delete_change_what_a_warmed_team_key_can_call( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model_one, provider_one = _team_model(scenario) + model_two, provider_two = _team_model(scenario) + model_three, provider_three = _team_model(scenario) + team: Final = scenario.team(models=[model_one]) + key: Final = scenario.key(team_id=team) + + first: Final = _chat(gateway, model_one, key) + assert first.status_code == 200, first.text + initially_denied: Final = _chat(gateway, model_two, key) + _model_denied(initially_denied) + observed_before_add: Final = _observed_provider_models(upstream) + assert observed_before_add == (provider_one,), repr(observed_before_add) + + added: Final = gateway.request("POST", "/team/model/add", {"team_id": team, "models": [model_two, model_three]}) + assert added.status_code == 200, added.text + added_state: Final = _TeamState.model_validate_json(added.text) + expected_added: Final = sorted((model_one, model_two, model_three)) + assert sorted(added_state.models or ()) == expected_added, added.text + added_info: Final = _team_info(gateway, team) + assert sorted(added_info.team_info.models or ()) == expected_added, repr(added_info) + added_row: Final = _stored_team_models(team) + assert sorted(added_row.models or ()) == expected_added, repr(added_row) + + second: Final = _chat(gateway, model_two, key) + assert second.status_code == 200, second.text + third: Final = _chat(gateway, model_three, key) + assert third.status_code == 200, third.text + deleted: Final = gateway.request("POST", "/team/model/delete", {"team_id": team, "models": [model_two]}) + assert deleted.status_code == 200, deleted.text + deleted_state: Final = _TeamState.model_validate_json(deleted.text) + expected_remaining: Final = sorted((model_one, model_three)) + assert sorted(deleted_state.models or ()) == expected_remaining, deleted.text + deleted_info: Final = _team_info(gateway, team) + assert sorted(deleted_info.team_info.models or ()) == expected_remaining, repr(deleted_info) + deleted_row: Final = _stored_team_models(team) + assert sorted(deleted_row.models or ()) == expected_remaining, repr(deleted_row) + + denied_after_delete: Final = _chat(gateway, model_two, key) + _model_denied(denied_after_delete) + still_allowed_one: Final = _chat(gateway, model_one, key) + assert still_allowed_one.status_code == 200, still_allowed_one.text + still_allowed_three: Final = _chat(gateway, model_three, key) + assert still_allowed_three.status_code == 200, still_allowed_three.text + observed_after_add_and_delete: Final = _observed_provider_models(upstream) + assert observed_before_add + observed_after_add_and_delete == ( + provider_one, + provider_two, + provider_three, + provider_one, + provider_three, + ), repr((observed_before_add, observed_after_add_and_delete)) + + +def test_team_model_add_on_an_unrestricted_team_keeps_every_other_model( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model_one, provider_one = _team_model(scenario) + model_two, provider_two = _team_model(scenario) + team: Final = scenario.team(models=[]) + key: Final = scenario.key(team_id=team) + + before_add: Final = _chat(gateway, model_two, key) + assert before_add.status_code == 200, before_add.text + + added: Final = gateway.request("POST", "/team/model/add", {"team_id": team, "models": [model_one]}) + assert added.status_code == 200, added.text + expected: Final = sorted(("all-proxy-models", model_one)) + added_state: Final = _TeamState.model_validate_json(added.text) + assert sorted(added_state.models or ()) == expected, added.text + added_info: Final = _team_info(gateway, team) + assert sorted(added_info.team_info.models or ()) == expected, repr(added_info) + added_row: Final = _stored_team_models(team) + assert sorted(added_row.models or ()) == expected, repr(added_row) + still_unrestricted: Final = _chat(gateway, model_two, key) + assert still_unrestricted.status_code == 200, still_unrestricted.text + added_model: Final = _chat(gateway, model_one, key) + assert added_model.status_code == 200, added_model.text + observed_after_add: Final = _observed_provider_models(upstream) + assert observed_after_add == (provider_two, provider_two, provider_one), repr(observed_after_add) diff --git a/tests/integration/management/test_team_block.py b/tests/integration/management/test_team_block.py new file mode 100644 index 00000000000..37c9da33c06 --- /dev/null +++ b/tests/integration/management/test_team_block.py @@ -0,0 +1,226 @@ +import uuid +from collections.abc import Iterator +from typing import Final + +import httpx +import pytest +from pydantic import BaseModel, JsonValue + +from tests.integration._support.client import Gateway, string_value +from tests.integration._support.database import read_rows + + +class _ObservedRequest(BaseModel): + body: dict[str, JsonValue] + + +class _Observations(BaseModel): + requests: list[_ObservedRequest] + + +class _TeamState(BaseModel): + team_id: str + blocked: bool + + +class _TeamInfo(BaseModel): + team_info: _TeamState + + +class _ProxyError(BaseModel): + type: str + + +class _ErrorResponse(BaseModel): + error: _ProxyError + + +@pytest.fixture +def upstream(gateway: Gateway) -> Iterator[httpx.Client]: + with httpx.Client(base_url=gateway.upstream_url, timeout=15, trust_env=False) as client: + response: Final = client.get("/__observations") + assert response.status_code == 200, response.text + yield client + + +def _chat(proxy: Gateway, model: str, key: str, text: str) -> httpx.Response: + return proxy.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": text}]}, + key=key, + ) + + +def _models(client: httpx.Client) -> tuple[str, ...]: + response: Final = client.get("/__observations") + assert response.status_code == 200, response.text + observations: Final = _Observations.model_validate_json(response.text) + return tuple(string_value(request.body["model"]) for request in observations.requests) + + +def test_team_block_and_unblock_update_endpoint_info_and_db_state( + gateway: Gateway, peer: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + provider_model: Final = f"team-block-{uuid.uuid4().hex}" + model: Final = scenario.model(model=f"openai/{provider_model}") + team: Final = scenario.team(models=[model]) + key: Final = scenario.key(team_id=team) + + primary_before: Final = _chat(gateway, model, key, "team block primary warm") + assert primary_before.status_code == 200, primary_before.text + peer_before: Final = _chat(peer, model, key, "team block peer warm") + assert peer_before.status_code == 200, peer_before.text + observed_before_block: Final = _models(upstream) + assert observed_before_block == (provider_model, provider_model), repr(observed_before_block) + + blocked_response: Final = gateway.request("POST", "/team/block", {"team_id": team}) + assert blocked_response.status_code == 200, blocked_response.text + blocked: Final = _TeamState.model_validate_json(blocked_response.text) + assert blocked.blocked is True, blocked_response.text + + info_response: Final = gateway.request("GET", "/team/info", params={"team_id": team}) + assert info_response.status_code == 200, info_response.text + info: Final = _TeamInfo.model_validate_json(info_response.text) + assert info.team_info.blocked is True, info_response.text + assert info.team_info.team_id == team, info_response.text + assert read_rows( + 'SELECT blocked FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [{"blocked": True}], info_response.text + + cold_key: Final = scenario.key(team_id=team) + cold_denied: Final = _chat(gateway, model, cold_key, "cold team key after block") + assert cold_denied.status_code == 401, cold_denied.text + assert _ErrorResponse.model_validate_json(cold_denied.text).error.type == "auth_error", cold_denied.text + after_cold_denied: Final = _models(upstream) + assert after_cold_denied == (), repr(after_cold_denied) + + unblocked_response: Final = gateway.request("POST", "/team/unblock", {"team_id": team}) + assert unblocked_response.status_code == 200, unblocked_response.text + restored_key: Final = scenario.key(team_id=team) + restored: Final = _chat(gateway, model, restored_key, "new team key after unblock") + assert restored.status_code == 200, restored.text + after_unblock: Final = _models(upstream) + assert after_unblock == (provider_model,), repr(after_unblock) + unblocked: Final = _TeamState.model_validate_json(unblocked_response.text) + assert unblocked.blocked is False, unblocked_response.text + unblocked_info_response: Final = gateway.request("GET", "/team/info", params={"team_id": team}) + assert unblocked_info_response.status_code == 200, unblocked_info_response.text + unblocked_info: Final = _TeamInfo.model_validate_json(unblocked_info_response.text) + assert unblocked_info.team_info.blocked is False, unblocked_info_response.text + assert read_rows( + 'SELECT blocked FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [{"blocked": False}], unblocked_info_response.text + + assert observed_before_block + after_cold_denied + after_unblock == ( + provider_model, + provider_model, + provider_model, + ), repr((observed_before_block, after_cold_denied, after_unblock)) + + +def test_unblocking_a_team_restores_warmed_keys_on_both_proxies( + gateway: Gateway, peer: Gateway, upstream: httpx.Client +) -> None: + pytest.skip("BUG: SLP-605 /team/unblock does not restore warmed team-key access after blocked auth is cached") + + with gateway.scenario() as scenario: + provider_model: Final = f"team-block-{uuid.uuid4().hex}" + model: Final = scenario.model(model=f"openai/{provider_model}") + team: Final = scenario.team(models=[model]) + key: Final = scenario.key(team_id=team) + + primary_before: Final = _chat(gateway, model, key, "team unblock primary warm") + assert primary_before.status_code == 200, primary_before.text + peer_before: Final = _chat(peer, model, key, "team unblock peer warm") + assert peer_before.status_code == 200, peer_before.text + observed_before_block: Final = _models(upstream) + assert observed_before_block == (provider_model, provider_model), repr(observed_before_block) + + blocked_response: Final = gateway.request("POST", "/team/block", {"team_id": team}) + assert blocked_response.status_code == 200, blocked_response.text + cold_key: Final = scenario.key(team_id=team) + cold_denied: Final = _chat(gateway, model, cold_key, "cold team key after block") + assert cold_denied.status_code == 401, cold_denied.text + assert _ErrorResponse.model_validate_json(cold_denied.text).error.type == "auth_error", cold_denied.text + after_cold_denied: Final = _models(upstream) + assert after_cold_denied == (), repr(after_cold_denied) + + unblocked_response: Final = gateway.request("POST", "/team/unblock", {"team_id": team}) + assert unblocked_response.status_code == 200, unblocked_response.text + + primary_after: Final = _chat(gateway, model, key, "team unblock primary") + assert primary_after.status_code == 200, primary_after.text + peer_after: Final = _chat(peer, model, key, "team unblock peer") + assert peer_after.status_code == 200, peer_after.text + after_unblock: Final = _models(upstream) + assert after_unblock == (provider_model, provider_model), repr(after_unblock) + + +def test_warmed_team_keys_are_refused_on_both_proxies_after_block( + gateway: Gateway, peer: Gateway, upstream: httpx.Client +) -> None: + pytest.skip("BUG: SLP-605 blocking a team does not refresh warmed team-key auth state") + + with gateway.scenario() as scenario: + provider_model: Final = f"team-block-{uuid.uuid4().hex}" + model: Final = scenario.model(model=f"openai/{provider_model}") + team: Final = scenario.team(models=[model]) + key: Final = scenario.key(team_id=team) + + primary_before: Final = _chat(gateway, model, key, "team block primary warm") + assert primary_before.status_code == 200, primary_before.text + peer_before: Final = _chat(peer, model, key, "team block peer warm") + assert peer_before.status_code == 200, peer_before.text + observed_before_block: Final = _models(upstream) + assert observed_before_block == (provider_model, provider_model), repr(observed_before_block) + + blocked_response: Final = gateway.request("POST", "/team/block", {"team_id": team}) + assert blocked_response.status_code == 200, blocked_response.text + + primary_denied: Final = _chat(gateway, model, key, "blocked team primary") + peer_denied: Final = _chat(peer, model, key, "blocked team peer") + after_block: Final = _models(upstream) + assert (primary_denied.status_code, peer_denied.status_code) == (401, 401), ( + f"primary={primary_denied.text}; peer={peer_denied.text}" + ) + assert _ErrorResponse.model_validate_json(primary_denied.text).error.type == "auth_error", primary_denied.text + assert _ErrorResponse.model_validate_json(peer_denied.text).error.type == "auth_error", peer_denied.text + assert after_block == (), repr(after_block) + assert observed_before_block + after_block == (provider_model, provider_model), repr( + (observed_before_block, after_block) + ) + + +def test_a_new_team_key_is_refused_on_a_peer_that_cached_the_team_before_block( + gateway: Gateway, peer: Gateway, upstream: httpx.Client +) -> None: + pytest.skip("BUG: SLP-605 a peer proxy that cached the team before /team/block admits a never-seen team key") + + with gateway.scenario() as scenario: + provider_model: Final = f"team-block-{uuid.uuid4().hex}" + model: Final = scenario.model(model=f"openai/{provider_model}") + team: Final = scenario.team(models=[model]) + warm_key: Final = scenario.key(team_id=team) + + peer_before: Final = _chat(peer, model, warm_key, "team block peer warm") + assert peer_before.status_code == 200, peer_before.text + observed_before_block: Final = _models(upstream) + assert observed_before_block == (provider_model,), repr(observed_before_block) + + blocked_response: Final = gateway.request("POST", "/team/block", {"team_id": team}) + assert blocked_response.status_code == 200, blocked_response.text + assert read_rows( + 'SELECT blocked FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [{"blocked": True}], blocked_response.text + + cold_key: Final = scenario.key(team_id=team) + peer_denied: Final = _chat(peer, model, cold_key, "new team key on peer after block") + after_block: Final = _models(upstream) + assert peer_denied.status_code == 401, peer_denied.text + assert _ErrorResponse.model_validate_json(peer_denied.text).error.type == "auth_error", peer_denied.text + assert after_block == (), repr(after_block) diff --git a/tests/integration/management/test_team_member_lifecycle.py b/tests/integration/management/test_team_member_lifecycle.py new file mode 100644 index 00000000000..f2b8abaa9f1 --- /dev/null +++ b/tests/integration/management/test_team_member_lifecycle.py @@ -0,0 +1,925 @@ +import json +import uuid +from collections.abc import Iterator +from hashlib import sha256 +from typing import Final + +import httpx +import pytest +from pydantic import BaseModel, JsonValue + +from tests.integration._support.client import ( + Gateway, + Scenario, + delete_key_if_present, + string_value, +) +from tests.integration._support.database import read_rows, write_rows + + +class _ObservedRequest(BaseModel): + body: dict[str, JsonValue] + + +class _Observations(BaseModel): + requests: list[_ObservedRequest] + + +class _RosterMember(BaseModel): + role: str + user_id: str | None = None + user_email: str | None = None + + +class _TeamRecord(BaseModel): + team_id: str + models: list[str] | None = None + members_with_roles: list[_RosterMember] + + +class _BudgetDetails(BaseModel): + max_budget: float | None = None + allowed_models: list[str] | None = None + + +class _TeamMembership(BaseModel): + team_id: str + user_id: str + budget_id: str | None = None + spend: float | None = None + total_spend: float | None = None + litellm_budget_table: _BudgetDetails | None = None + + +class _TeamInfo(BaseModel): + team_info: _TeamRecord + team_memberships: list[_TeamMembership] | None = None + + +class _UserRecord(BaseModel): + user_id: str + user_email: str | None = None + teams: list[str] | None = None + + +class _MemberAddResponse(BaseModel): + team_id: str + updated_users: list[_UserRecord] + updated_team_memberships: list[_TeamMembership] + + +class _MemberUpdateResponse(BaseModel): + team_id: str + user_id: str + max_budget_in_team: float | None = None + + +class _GeneratedKey(BaseModel): + key: str + + +class _ProxyError(BaseModel): + type: str + + +class _ErrorResponse(BaseModel): + error: _ProxyError + + +class _HTTPErrorDetail(BaseModel): + error: str + + +class _HTTPErrorResponse(BaseModel): + detail: _HTTPErrorDetail + + +class _BulkAddResult(BaseModel): + user_id: str | None = None + user_email: str | None = None + success: bool + error: str | None = None + + +class _BulkAddResponse(BaseModel): + team_id: str + results: list[_BulkAddResult] + total_requested: int + successful_additions: int + failed_additions: int + + +class _BulkDeleteResult(BaseModel): + user_id: str | None = None + user_email: str | None = None + success: bool + error: str | None = None + + +class _BulkDeleteResponse(BaseModel): + data: tuple[_BulkDeleteResult, ...] + + +@pytest.fixture +def upstream(gateway: Gateway) -> Iterator[httpx.Client]: + with httpx.Client(base_url=gateway.upstream_url, timeout=15, trust_env=False) as client: + response: Final = client.get("/__observations") + assert response.status_code == 200, response.text + yield client + + +def _model(scenario: Scenario) -> tuple[str, str]: + provider_model: Final = f"team-member-{uuid.uuid4().hex}" + return scenario.model(model=f"openai/{provider_model}"), provider_model + + +def _key(scenario: Scenario, **fields: JsonValue) -> str: + response: Final = scenario.gateway.request("POST", "/key/generate", fields) + assert response.status_code == 200, response.text + token: Final = _GeneratedKey.model_validate_json(response.text).key + scenario.cleanups.callback(delete_key_if_present, scenario.gateway, token) + return token + + +def _team_info(gateway: Gateway, team_id: str) -> _TeamInfo: + response: Final = gateway.request("GET", "/team/info", params={"team_id": team_id}) + assert response.status_code == 200, response.text + return _TeamInfo.model_validate_json(response.text) + + +def _models(upstream: httpx.Client) -> tuple[str, ...]: + response: Final = upstream.get("/__observations") + assert response.status_code == 200, response.text + observations: Final = _Observations.model_validate_json(response.text) + return tuple(string_value(request.body["model"]) for request in observations.requests) + + +def _chat(proxy: Gateway, model: str, key: str, text: str) -> httpx.Response: + return proxy.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": text}]}, + key=key, + ) + + +def _refused(response: httpx.Response, status: int, error_type: str) -> None: + assert response.status_code == status, response.text + assert _ErrorResponse.model_validate_json(response.text).error.type == error_type, response.text + + +def _remove_default_user(gateway: Gateway, team_id: str) -> None: + response: Final = gateway.request("POST", "/team/member_delete", {"team_id": team_id, "user_id": "default_user_id"}) + assert response.status_code == 200, response.text + + +def _roster_member(user_id: str, user_email: str | None, role: str = "user") -> dict[str, JsonValue]: + return {"role": role, "user_id": user_id, "user_email": user_email} + + +@pytest.mark.parametrize("identifier", ("user_id", "user_email"), ids=("by-id", "by-email")) +def test_member_delete_revokes_the_removed_members_warmed_key( + gateway: Gateway, upstream: httpx.Client, identifier: str +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _model(scenario) + team: Final = scenario.team(models=[model]) + leaver_email: Final = f"member-leaver-{uuid.uuid4().hex}@integration.test" + leaver: Final = scenario.user(user_email=leaver_email) + _remove_default_user(gateway, team) + stayer: Final = scenario.member(team) + added: Final = gateway.request( + "POST", + "/team/member_add", + { + "team_id": team, + "member": ( + {"role": "user", "user_id": leaver} + if identifier == "user_id" + else {"role": "user", "user_email": leaver_email} + ), + }, + ) + assert added.status_code == 200, added.text + leaver_key: Final = _key(scenario, team_id=team, user_id=leaver) + stayer_key: Final = _key(scenario, team_id=team, user_id=stayer) + leaver_warm: Final = _chat(gateway, model, leaver_key, "warm leaver " + uuid.uuid4().hex) + assert leaver_warm.status_code == 200, leaver_warm.text + stayer_warm: Final = _chat(gateway, model, stayer_key, "warm stayer " + uuid.uuid4().hex) + assert stayer_warm.status_code == 200, stayer_warm.text + observed_before_delete: Final = _models(upstream) + assert observed_before_delete == (provider_model, provider_model), repr(observed_before_delete) + + delete_body: Final = ( + {"team_id": team, "user_id": leaver} + if identifier == "user_id" + else {"team_id": team, "user_email": leaver_email} + ) + deleted: Final = gateway.request("POST", "/team/member_delete", delete_body) + assert deleted.status_code == 200, deleted.text + + info: Final = _team_info(gateway, team) + stayer_roster: Final = [_roster_member(stayer, None)] + assert info.team_info.members_with_roles == [_RosterMember.model_validate(stayer_roster[0])], repr(info) + assert read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,)) == [ + {"members_with_roles": stayer_roster} + ], deleted.text + assert ( + read_rows( + 'SELECT user_id, team_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s AND user_id = %s', + (team, leaver), + ) + == [] + ), deleted.text + assert read_rows( + 'SELECT user_id, team_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s AND user_id = %s', + (team, stayer), + ) == [{"user_id": stayer, "team_id": team}], deleted.text + leaver_hash: Final = sha256(leaver_key.encode()).hexdigest() + assert ( + read_rows( + 'SELECT token FROM "LiteLLM_VerificationToken" WHERE token = %s', + (leaver_hash,), + ) + == [] + ), deleted.text + assert read_rows( + 'SELECT token FROM "LiteLLM_DeletedVerificationToken" WHERE token = %s', + (leaver_hash,), + ) == [{"token": leaver_hash}], deleted.text + + refused: Final = _chat(gateway, model, leaver_key, "deleted leaver " + uuid.uuid4().hex) + _refused(refused, 401, "token_not_found_in_db") + after_refused: Final = _models(upstream) + assert after_refused == (), repr(after_refused) + stayer_after: Final = _chat(gateway, model, stayer_key, "stayer after delete " + uuid.uuid4().hex) + assert stayer_after.status_code == 200, stayer_after.text + observed_after_delete: Final = _models(upstream) + assert observed_before_delete + after_refused + observed_after_delete == ( + provider_model, + provider_model, + provider_model, + ), repr((observed_before_delete, after_refused, observed_after_delete)) + + +def test_member_delete_by_email_removes_a_legacy_email_only_roster_entry( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _model(scenario) + team: Final = scenario.team(models=[model]) + email: Final = f"legacy-member-{uuid.uuid4().hex}@integration.test" + user: Final = scenario.user(user_email=email) + _remove_default_user(gateway, team) + stayer: Final = scenario.member(team) + added: Final = gateway.request( + "POST", "/team/member_add", {"team_id": team, "member": {"role": "user", "user_id": user}} + ) + assert added.status_code == 200, added.text + leaver_key: Final = _key(scenario, team_id=team, user_id=user) + stayer_key: Final = _key(scenario, team_id=team, user_id=stayer) + leaver_warm: Final = _chat(gateway, model, leaver_key, "warm legacy leaver " + uuid.uuid4().hex) + assert leaver_warm.status_code == 200, leaver_warm.text + observed_before_delete: Final = _models(upstream) + assert observed_before_delete == (provider_model,), repr(observed_before_delete) + + stayer_roster: Final = _roster_member(stayer, None) + legacy_roster: Final = {"role": "user", "user_id": None, "user_email": email} + write_rows( + 'UPDATE "LiteLLM_TeamTable" SET members_with_roles = %s::jsonb WHERE team_id = %s', + (json.dumps([legacy_roster, stayer_roster]), team), + ) + assert read_rows('SELECT teams FROM "LiteLLM_UserTable" WHERE user_id = %s', (user,)) == [{"teams": [team]}], ( + added.text + ) + leaver_hash: Final = sha256(leaver_key.encode()).hexdigest() + + deleted: Final = gateway.request("POST", "/team/member_delete", {"team_id": team, "user_email": email}) + assert deleted.status_code == 200, deleted.text + info: Final = _team_info(gateway, team) + assert info.team_info.members_with_roles == [_RosterMember.model_validate(stayer_roster)], repr(info) + assert read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,)) == [ + {"members_with_roles": [stayer_roster]} + ], deleted.text + assert read_rows('SELECT teams FROM "LiteLLM_UserTable" WHERE user_id = %s', (user,)) == [{"teams": []}], ( + deleted.text + ) + assert read_rows( + 'SELECT user_id, team_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s ORDER BY user_id', + (team,), + ) == [{"user_id": stayer, "team_id": team}], deleted.text + assert read_rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token = %s', (leaver_hash,)) == [], ( + deleted.text + ) + assert read_rows( + 'SELECT token FROM "LiteLLM_DeletedVerificationToken" WHERE token = %s', + (leaver_hash,), + ) == [{"token": leaver_hash}], deleted.text + + refused: Final = _chat(gateway, model, leaver_key, "deleted legacy leaver " + uuid.uuid4().hex) + _refused(refused, 401, "token_not_found_in_db") + after_refused: Final = _models(upstream) + assert after_refused == (), repr(after_refused) + served: Final = _chat(gateway, model, stayer_key, "stayer after legacy delete " + uuid.uuid4().hex) + assert served.status_code == 200, served.text + observed: Final = _models(upstream) + assert observed_before_delete + after_refused + observed == (provider_model, provider_model), repr( + (observed_before_delete, after_refused, observed) + ) + + +def test_member_delete_by_case_variant_email_removes_member_and_refuses_warmed_key( + gateway: Gateway, upstream: httpx.Client +) -> None: + pytest.skip("BUG: case-variant email does not match the member during team deletion") + + with gateway.scenario() as scenario: + model, provider_model = _model(scenario) + team: Final = scenario.team(models=[model]) + email: Final = f"case-delete-{uuid.uuid4().hex}@integration.test" + user: Final = scenario.user(user_email=email) + _remove_default_user(gateway, team) + stayer: Final = scenario.member(team) + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_email": email}}, + ) + assert added.status_code == 200, added.text + leaver_key: Final = _key(scenario, team_id=team, user_id=user) + warmed: Final = _chat(gateway, model, leaver_key, f"warm case-variant leaver {uuid.uuid4().hex}") + assert warmed.status_code == 200, warmed.text + observed_before_delete: Final = _models(upstream) + assert observed_before_delete == (provider_model,), repr(observed_before_delete) + + deleted: Final = gateway.request( + "POST", + "/team/member_delete", + {"team_id": team, "user_email": email.upper()}, + ) + assert deleted.status_code == 200, deleted.text + info: Final = _team_info(gateway, team) + stayer_roster: Final = _roster_member(stayer, None) + assert info.team_info.members_with_roles == [_RosterMember.model_validate(stayer_roster)], repr(info) + assert read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,)) == [ + {"members_with_roles": [stayer_roster]} + ], deleted.text + assert read_rows( + 'SELECT user_id, team_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s AND user_id = %s', + (team, user), + ) == [], deleted.text + + refused: Final = _chat(gateway, model, leaver_key, f"deleted case-variant leaver {uuid.uuid4().hex}") + _refused(refused, 401, "token_not_found_in_db") + after_refused: Final = _models(upstream) + assert after_refused == (), repr(after_refused) + assert observed_before_delete + after_refused == (provider_model,), repr( + (observed_before_delete, after_refused) + ) + + +def test_member_update_by_case_variant_email_updates_the_member( + gateway: Gateway, +) -> None: + pytest.skip("BUG: case-variant email does not match the member during team update") + + with gateway.scenario() as scenario: + team: Final = scenario.team() + email: Final = f"case-update-{uuid.uuid4().hex}@integration.test" + user: Final = scenario.user(user_email=email) + _remove_default_user(gateway, team) + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_email": email}}, + ) + assert added.status_code == 200, added.text + + response: Final = gateway.request( + "POST", + "/team/member_update", + {"team_id": team, "user_email": email.upper(), "max_budget_in_team": 12}, + ) + assert response.status_code == 200, response.text + updated: Final = _MemberUpdateResponse.model_validate_json(response.text) + assert updated.user_id == user, response.text + assert updated.max_budget_in_team == 12, response.text + + membership_rows: Final = read_rows( + 'SELECT user_id, team_id, budget_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s AND user_id = %s', + (team, user), + ) + assert len(membership_rows) == 1, membership_rows + budget_id: Final = membership_rows[0]["budget_id"] + assert isinstance(budget_id, str), membership_rows + budget_rows: Final = read_rows( + 'SELECT budget_id, max_budget FROM "LiteLLM_BudgetTable" WHERE budget_id = %s', + (budget_id,), + ) + assert budget_rows == [{"budget_id": budget_id, "max_budget": 12.0}], budget_rows + + info: Final = _team_info(gateway, team) + assert info.team_memberships is not None, repr(info) + updated_membership: Final = next(member for member in info.team_memberships if member.user_id == user) + assert updated_membership.budget_id == budget_id, repr(updated_membership) + assert updated_membership.litellm_budget_table is not None, repr(updated_membership) + assert updated_membership.litellm_budget_table.max_budget == 12, repr(updated_membership) + + +def test_member_add_by_email_provisions_a_new_user_and_resolves_an_existing_one( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + team: Final = scenario.team() + _remove_default_user(gateway, team) + existing_email: Final = f"existing-member-{uuid.uuid4().hex}@integration.test" + existing_user: Final = scenario.user(user_email=existing_email) + existing_email_variant: Final = existing_email.upper() + existing_added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_email": existing_email_variant}}, + ) + assert existing_added.status_code == 200, existing_added.text + existing_response: Final = _MemberAddResponse.model_validate_json(existing_added.text) + assert tuple(user.user_id for user in existing_response.updated_users) == (existing_user,), existing_added.text + assert read_rows( + 'SELECT user_id, user_email FROM "LiteLLM_UserTable" WHERE LOWER(user_email) = LOWER(%s)', + (existing_email,), + ) == [{"user_id": existing_user, "user_email": existing_email}], existing_added.text + + new_email: Final = f"provisioned-member-{uuid.uuid4().hex}@integration.test" + new_added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_email": new_email}}, + ) + assert new_added.status_code == 200, new_added.text + new_response: Final = _MemberAddResponse.model_validate_json(new_added.text) + assert len(new_response.updated_users) == 1, new_added.text + provisioned: Final = new_response.updated_users[0] + assert provisioned.user_email == new_email, new_added.text + scenario.cleanups.callback(scenario.delete_user, provisioned.user_id) + assert read_rows( + 'SELECT user_id, user_email, teams FROM "LiteLLM_UserTable" WHERE user_email = %s', + (new_email,), + ) == [{"user_id": provisioned.user_id, "user_email": new_email, "teams": [team]}], new_added.text + info: Final = _team_info(gateway, team) + assert info.team_info.members_with_roles == [ + _RosterMember(role="user", user_id=existing_user, user_email=existing_email_variant), + _RosterMember(role="user", user_id=provisioned.user_id, user_email=new_email), + ], repr(info) + observed: Final = _models(upstream) + assert observed == (), repr(observed) + + +def test_member_add_list_mixing_id_and_email_applies_budget_and_allowed_models( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model_one, provider_one = _model(scenario) + model_two, provider_two = _model(scenario) + team: Final = scenario.team(models=[model_one, model_two]) + _remove_default_user(gateway, team) + user_one: Final = scenario.user(user_role="internal_user") + email_two: Final = f"mixed-member-{uuid.uuid4().hex}@integration.test" + user_two: Final = scenario.user(user_email=email_two, user_role="internal_user") + added: Final = gateway.request( + "POST", + "/team/member_add", + { + "team_id": team, + "member": [ + {"role": "user", "user_id": user_one}, + {"role": "user", "user_email": email_two}, + ], + "max_budget_in_team": 1, + "allowed_models": [model_one], + }, + ) + assert added.status_code == 200, added.text + response: Final = _MemberAddResponse.model_validate_json(added.text) + assert tuple(user.user_id for user in response.updated_users) == (user_one, user_two), added.text + assert tuple(member.user_id for member in response.updated_team_memberships) == (user_one, user_two), added.text + assert read_rows( + "SELECT tm.user_id, b.max_budget, b.allowed_models " + 'FROM "LiteLLM_TeamMembership" tm ' + 'JOIN "LiteLLM_BudgetTable" b ON b.budget_id = tm.budget_id ' + "WHERE tm.team_id = %s ORDER BY tm.user_id", + (team,), + ) == [ + {"user_id": user_id, "max_budget": 1.0, "allowed_models": [model_one]} + for user_id in sorted((user_one, user_two)) + ], added.text + info: Final = _team_info(gateway, team) + membership_limits: Final = tuple( + sorted( + ( + ( + membership.user_id, + membership.litellm_budget_table.max_budget + if membership.litellm_budget_table is not None + else None, + membership.litellm_budget_table.allowed_models + if membership.litellm_budget_table is not None + else None, + ) + for membership in info.team_memberships or () + ), + key=lambda member: member[0], + ) + ) + assert membership_limits == tuple( + sorted(((user_one, 1, [model_one]), (user_two, 1, [model_one])), key=lambda member: member[0]) + ), added.text + + key_one: Final = _key(scenario, team_id=team, user_id=user_one) + key_two: Final = _key(scenario, team_id=team, user_id=user_two) + allowed_one: Final = _chat(gateway, model_one, key_one, "mixed member one allowed") + assert allowed_one.status_code == 200, allowed_one.text + denied_one: Final = _chat(gateway, model_two, key_one, "mixed member one denied") + _refused(denied_one, 403, "team_model_access_denied") + after_denied_one: Final = _models(upstream) + allowed_two: Final = _chat(gateway, model_one, key_two, "mixed member two allowed") + assert allowed_two.status_code == 200, allowed_two.text + denied_two: Final = _chat(gateway, model_two, key_two, "mixed member two denied") + _refused(denied_two, 403, "team_model_access_denied") + after_denied_two: Final = _models(upstream) + assert after_denied_one == (provider_one,), repr(after_denied_one) + assert after_denied_two == (provider_one,), repr(after_denied_two) + assert provider_two not in after_denied_two, repr(after_denied_two) + assert after_denied_one + after_denied_two == (provider_one, provider_one), repr( + (after_denied_one, after_denied_two) + ) + + +def test_bulk_member_add_adds_every_listed_member_and_reports_failures( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _model(scenario) + team: Final = scenario.team(models=[model]) + _remove_default_user(gateway, team) + user_one: Final = scenario.user(user_role="internal_user") + user_two: Final = scenario.user(user_role="internal_user") + email_three: Final = f"bulk-member-{uuid.uuid4().hex}@integration.test" + user_three: Final = scenario.user(user_email=email_three, user_role="internal_user") + email_four: Final = f"bulk-admin-{uuid.uuid4().hex}@integration.test" + user_four: Final = scenario.user(user_email=email_four, user_role="internal_user") + email_five: Final = f"bulk-admin-provisioned-{uuid.uuid4().hex}@integration.test" + members: Final = [ + {"role": "user", "user_id": user_one}, + {"role": "user", "user_id": user_two}, + {"role": "user", "user_email": email_three}, + {"role": "admin", "user_email": email_four}, + {"role": "admin", "user_email": email_five}, + ] + added: Final = gateway.request( + "POST", + "/team/bulk_member_add", + {"team_id": team, "members": members, "max_budget_in_team": 1}, + ) + assert added.status_code == 200, added.text + provisioned_rows: Final = read_rows( + 'SELECT user_id FROM "LiteLLM_UserTable" WHERE user_email = %s', + (email_five,), + ) + assert len(provisioned_rows) == 1, added.text + user_five: Final = string_value(provisioned_rows[0]["user_id"]) + scenario.cleanups.callback(scenario.delete_user, user_five) + response: Final = _BulkAddResponse.model_validate_json(added.text) + assert response.team_id == team, added.text + assert response.total_requested == 5, added.text + assert response.successful_additions == 5, added.text + assert response.failed_additions == 0, added.text + assert response.results == [ + _BulkAddResult(user_id=user_one, success=True), + _BulkAddResult(user_id=user_two, success=True), + _BulkAddResult(user_id=user_three, user_email=email_three, success=True), + _BulkAddResult(user_id=user_four, user_email=email_four, success=True), + _BulkAddResult(user_email=email_five, success=True), + ], added.text + assert read_rows( + 'SELECT user_id, team_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s ORDER BY user_id', + (team,), + ) == [ + {"user_id": user_id, "team_id": team} + for user_id in sorted((user_one, user_two, user_three, user_four, user_five)) + ], added.text + assert read_rows( + 'SELECT m.user_id, b.max_budget FROM "LiteLLM_TeamMembership" m ' + 'JOIN "LiteLLM_BudgetTable" b ON b.budget_id = m.budget_id WHERE m.team_id = %s ORDER BY m.user_id', + (team,), + ) == [ + {"user_id": user_id, "max_budget": 1.0} + for user_id in sorted((user_one, user_two, user_three, user_four, user_five)) + ], added.text + assert read_rows( + 'SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [ + { + "members_with_roles": [ + _roster_member(user_one, None), + _roster_member(user_two, None), + _roster_member(user_three, email_three), + _roster_member(user_four, email_four, role="admin"), + _roster_member(user_five, email_five, role="admin"), + ] + } + ], added.text + + write_rows( + 'UPDATE "LiteLLM_TeamMembership" SET spend = 2 WHERE team_id = %s AND user_id = %s', + (team, user_three), + ) + keys: Final = tuple(_key(scenario, team_id=team, user_id=user) for user in (user_one, user_two, user_three)) + calls: Final = tuple(_chat(gateway, model, key, f"bulk member {index}") for index, key in enumerate(keys)) + assert tuple(response.status_code for response in calls[:2]) == (200, 200), tuple( + response.text for response in calls + ) + _refused(calls[2], 422, "budget_exceeded") + duplicate: Final = gateway.request( + "POST", + "/team/bulk_member_add", + {"team_id": team, "members": members, "max_budget_in_team": 1}, + ) + assert duplicate.status_code == 200, duplicate.text + failed: Final = _BulkAddResponse.model_validate_json(duplicate.text) + assert failed.total_requested == 5, duplicate.text + assert failed.successful_additions == 0, duplicate.text + assert failed.failed_additions == 5, duplicate.text + assert tuple(result.success for result in failed.results) == (False,) * 5, duplicate.text + assert all(result.error is not None for result in failed.results), duplicate.text + assert read_rows( + 'SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [ + { + "members_with_roles": [ + _roster_member(user_one, None), + _roster_member(user_two, None), + _roster_member(user_three, email_three), + _roster_member(user_four, email_four, role="admin"), + _roster_member(user_five, email_five, role="admin"), + ] + } + ], duplicate.text + + admin_personal_key: Final = _key(scenario, user_id=user_four) + provisioned_admin_key: Final = _key(scenario, user_id=user_five) + member_personal_key: Final = _key(scenario, user_id=user_one) + probe: Final = scenario.user() + admin_added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_id": probe}}, + key=admin_personal_key, + ) + assert admin_added.status_code == 200, admin_added.text + probe_five: Final = scenario.user() + provisioned_admin_added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_id": probe_five}}, + key=provisioned_admin_key, + ) + assert provisioned_admin_added.status_code == 200, provisioned_admin_added.text + refused_probe: Final = scenario.user() + member_denied: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_id": refused_probe}}, + key=member_personal_key, + ) + assert member_denied.status_code == 403, member_denied.text + denied_response: Final = _HTTPErrorResponse.model_validate_json(member_denied.text) + assert "User not proxy admin OR team admin" in denied_response.detail.error, member_denied.text + assert read_rows( + 'SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) == [ + { + "members_with_roles": [ + _roster_member(user_one, None), + _roster_member(user_two, None), + _roster_member(user_three, email_three), + _roster_member(user_four, email_four, role="admin"), + _roster_member(user_five, email_five, role="admin"), + _roster_member(probe, None), + _roster_member(probe_five, None), + ] + } + ], member_denied.text + observed: Final = _models(upstream) + assert observed == (provider_model, provider_model), repr(observed) + + +def test_bulk_member_add_all_users_is_proxy_admin_only_and_adds_every_user( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + team: Final = scenario.team() + _remove_default_user(gateway, team) + caller: Final = scenario.user(user_role="internal_user") + caller_key: Final = _key(scenario, user_id=caller, allowed_routes=["/team/bulk_member_add"]) + denied: Final = gateway.request( + "POST", "/team/bulk_member_add", {"team_id": team, "all_users": True}, key=caller_key + ) + assert denied.status_code == 403, denied.text + assert _HTTPErrorResponse.model_validate_json(denied.text) == _HTTPErrorResponse( + detail=_HTTPErrorDetail( + error="`all_users=true` is restricted to PROXY_ADMIN. Org/team admins must specify explicit member lists." + ) + ), denied.text + assert read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,)) == [ + {"members_with_roles": []} + ], denied.text + assert read_rows('SELECT user_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s', (team,)) == [], denied.text + + users: Final = tuple( + (string_value(row["user_id"]), row["user_email"]) + for row in read_rows('SELECT user_id, user_email FROM "LiteLLM_UserTable" ORDER BY user_id', ()) + ) + assert caller in {user_id for user_id, _ in users}, repr(users) + added: Final = gateway.request("POST", "/team/bulk_member_add", {"team_id": team, "all_users": True}) + assert added.status_code == 200, added.text + response: Final = _BulkAddResponse.model_validate_json(added.text) + assert response.team_id == team, added.text + assert response.total_requested == len(users), added.text + assert response.successful_additions == len(users), added.text + assert response.failed_additions == 0, added.text + assert sorted( + ((result.user_id, result.user_email, result.success) for result in response.results), + key=lambda result: result[0] or "", + ) == [(user_id, user_email, True) for user_id, user_email in users], added.text + assert read_rows( + 'SELECT user_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s ORDER BY user_id', + (team,), + ) == [{"user_id": user_id} for user_id, _ in users], added.text + roster: Final = read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,)) + assert len(roster) == 1, added.text + roster_members: Final = roster[0]["members_with_roles"] + assert isinstance(roster_members, list), repr(roster) + assert sorted( + (_RosterMember.model_validate(member) for member in roster_members), + key=lambda member: member.user_id or "", + ) == [ + _RosterMember.model_validate(_roster_member(user_id, string_value(user_email) if user_email else None)) + for user_id, user_email in users + ], added.text + observed: Final = _models(upstream) + assert observed == (), repr(observed) + + +def test_bulk_delete_removes_exactly_the_named_members_and_revokes_their_access( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _model(scenario) + team: Final = scenario.team(models=[model]) + _remove_default_user(gateway, team) + admin: Final = scenario.member(team, role="admin") + email_member: Final = f"bulk-delete-member-{uuid.uuid4().hex}@integration.test" + member: Final = scenario.user(user_email=email_member, user_role="internal_user") + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_email": email_member}}, + ) + assert added.status_code == 200, added.text + stayer: Final = scenario.member(team) + admin_key: Final = _key(scenario, team_id=team, user_id=admin) + member_key: Final = _key(scenario, team_id=team, user_id=member) + stayer_key: Final = _key(scenario, team_id=team, user_id=stayer) + personal_admin_key: Final = _key(scenario, user_id=admin) + admin_warm: Final = _chat(gateway, model, admin_key, "bulk delete warm admin") + assert admin_warm.status_code == 200, admin_warm.text + member_warm: Final = _chat(gateway, model, member_key, "bulk delete warm member") + assert member_warm.status_code == 200, member_warm.text + stayer_warm: Final = _chat(gateway, model, stayer_key, "bulk delete warm stayer") + assert stayer_warm.status_code == 200, stayer_warm.text + observed_before_delete: Final = _models(upstream) + assert observed_before_delete == (provider_model, provider_model, provider_model), repr(observed_before_delete) + + probe_user: Final = scenario.user() + admin_route_before: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_id": probe_user}}, + key=personal_admin_key, + ) + assert admin_route_before.status_code == 200, admin_route_before.text + probe_removed: Final = gateway.request("POST", "/team/member_delete", {"team_id": team, "user_id": probe_user}) + assert probe_removed.status_code == 200, probe_removed.text + deleted: Final = gateway.request( + "POST", + f"/management/v1/teams/{team}/members/bulk_delete", + { + "members": [ + {"user_id": admin}, + {"user_email": email_member}, + {"user_id": admin}, + ] + }, + ) + assert deleted.status_code == 200, deleted.text + results: Final = _BulkDeleteResponse.model_validate_json(deleted.text).data + assert results == ( + _BulkDeleteResult(user_id=admin, success=True), + _BulkDeleteResult(user_email=email_member, success=True), + _BulkDeleteResult(user_id=admin, success=False, error="Duplicate member in request"), + ), deleted.text + team_info: Final = _team_info(gateway, team) + assert team_info.team_info.members_with_roles == [ + _RosterMember(role="user", user_id=stayer, user_email=None) + ], repr(team_info) + assert read_rows( + 'SELECT user_id, team_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s ORDER BY user_id', + (team,), + ) == [{"user_id": stayer, "team_id": team}], deleted.text + for removed_key in (admin_key, member_key): + removed_hash: Final = sha256(removed_key.encode()).hexdigest() + assert ( + read_rows( + 'SELECT token FROM "LiteLLM_VerificationToken" WHERE token = %s', + (removed_hash,), + ) + == [] + ), deleted.text + + denied_admin: Final = _chat(gateway, model, admin_key, "deleted admin key") + _refused(denied_admin, 401, "token_not_found_in_db") + observed_after_admin_denied: Final = _models(upstream) + denied_member: Final = _chat(gateway, model, member_key, "deleted member key") + _refused(denied_member, 401, "token_not_found_in_db") + observed_after_member_denied: Final = _models(upstream) + denied_admin_route: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_id": probe_user}}, + key=personal_admin_key, + ) + assert denied_admin_route.status_code == 403, denied_admin_route.text + denied_admin_response: Final = _HTTPErrorResponse.model_validate_json(denied_admin_route.text) + assert "User not proxy admin OR team admin" in denied_admin_response.detail.error, denied_admin_route.text + stayer_after: Final = _chat(gateway, model, stayer_key, "stayer after bulk delete") + assert stayer_after.status_code == 200, stayer_after.text + observed_after_stayer: Final = _models(upstream) + assert observed_after_admin_denied == (), repr(observed_after_admin_denied) + assert observed_after_member_denied == (), repr(observed_after_member_denied) + assert ( + observed_before_delete + observed_after_admin_denied + observed_after_member_denied + observed_after_stayer + == ( + provider_model, + provider_model, + provider_model, + provider_model, + ) + ) + + +def test_bulk_delete_by_case_variant_email_removes_member_and_refuses_warmed_key( + gateway: Gateway, upstream: httpx.Client +) -> None: + pytest.skip("BUG: bulk member deletion does not match case-variant member email") + with gateway.scenario() as scenario: + model, provider_model = _model(scenario) + team: Final = scenario.team(models=[model]) + _remove_default_user(gateway, team) + email: Final = f"bulk-case-delete-{uuid.uuid4().hex}@integration.test" + user: Final = scenario.user(user_email=email) + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"role": "user", "user_email": email}}, + ) + assert added.status_code == 200, added.text + member_key: Final = _key(scenario, team_id=team, user_id=user) + warmed: Final = _chat(gateway, model, member_key, f"warm bulk case-variant member {uuid.uuid4().hex}") + assert warmed.status_code == 200, warmed.text + observed_before_delete: Final = _models(upstream) + assert observed_before_delete == (provider_model,), repr(observed_before_delete) + + deleted: Final = gateway.request( + "POST", + f"/management/v1/teams/{team}/members/bulk_delete", + {"members": [{"user_email": email.upper()}]}, + ) + assert deleted.status_code == 200, deleted.text + results: Final = _BulkDeleteResponse.model_validate_json(deleted.text).data + assert results == (_BulkDeleteResult(user_email=email.upper(), success=True),), deleted.text + info: Final = _team_info(gateway, team) + assert info.team_info.members_with_roles == [], repr(info) + assert read_rows( + 'SELECT user_id, team_id FROM "LiteLLM_TeamMembership" WHERE team_id = %s AND user_id = %s', + (team, user), + ) == [], deleted.text + + refused: Final = _chat(gateway, model, member_key, f"deleted bulk case-variant member {uuid.uuid4().hex}") + _refused(refused, 401, "token_not_found_in_db") + after_refused: Final = _models(upstream) + assert after_refused == (), repr(after_refused) + assert observed_before_delete + after_refused == (provider_model,), repr( + (observed_before_delete, after_refused) + ) diff --git a/tests/integration/spend/test_team_member_spend.py b/tests/integration/spend/test_team_member_spend.py index cf1dac25793..343a3be1b49 100644 --- a/tests/integration/spend/test_team_member_spend.py +++ b/tests/integration/spend/test_team_member_spend.py @@ -1,14 +1,231 @@ import os import uuid +from collections.abc import Callable, Iterator +from datetime import datetime, timedelta, timezone from typing import Final import httpx import pytest -from integration._support.client import Gateway, eventually, object_value +from integration._support.client import Gateway, Scenario, eventually, object_value, string_value from integration._support.database import read_rows +from pydantic import BaseModel, JsonValue from redis import Redis +def _python_isoformat(value: datetime) -> str: + return value.isoformat() + + +def _javascript_iso_string(value: datetime) -> str: + return value.strftime("%Y-%m-%dT%H:%M:%S.000Z") + + +class _ObservedRequest(BaseModel): + body: dict[str, JsonValue] + + +class _Observations(BaseModel): + requests: list[_ObservedRequest] + + +class _ProxyError(BaseModel): + type: str + + +class _ErrorResponse(BaseModel): + error: _ProxyError + + +class _BudgetRecord(BaseModel): + budget_id: str + max_budget: float | None = None + tpm_limit: int | None = None + rpm_limit: int | None = None + budget_duration: str | None = None + allowed_models: list[str] | None = None + temp_budget_increase: float | None = None + temp_budget_expiry: datetime | None = None + + +class _MemberBudgetState(BaseModel): + user_id: str + team_id: str + budget_id: str | None = None + spend: float | None = None + max_budget: float | None = None + tpm_limit: int | None = None + rpm_limit: int | None = None + budget_duration: str | None = None + allowed_models: list[str] | None = None + temp_budget_increase: float | None = None + temp_budget_expiry: datetime | None = None + + +class _TeamRosterMember(BaseModel): + role: str + user_id: str | None = None + user_email: str | None = None + + +class _TeamState(BaseModel): + team_id: str + models: list[str] | None = None + members_with_roles: tuple[_TeamRosterMember, ...] = () + metadata: dict[str, JsonValue] | None = None + + +class _TeamMembershipInfo(BaseModel): + user_id: str + team_id: str + budget_id: str | None = None + spend: float | None = None + budget_source: str | None = None + litellm_budget_table: _BudgetRecord | None = None + + +class _TeamInfo(BaseModel): + team_info: _TeamState + team_memberships: list[_TeamMembershipInfo] + + +class _MemberMe(BaseModel): + user_id: str + team_id: str + budget_id: str | None = None + role: str | None = None + litellm_budget_table: _BudgetRecord | None = None + + +class _MemberUpdateResponse(BaseModel): + team_id: str + user_id: str + user_email: str | None = None + max_budget_in_team: float | None = None + tpm_limit: int | None = None + rpm_limit: int | None = None + budget_duration: str | None = None + allowed_models: list[str] | None = None + temp_budget_increase: float | None = None + temp_budget_expiry: datetime | None = None + + +class _BulkBudgetResult(BaseModel): + user_id: str | None = None + user_email: str | None = None + success: bool + error: str | None = None + budget_id: str | None = None + max_budget: float | None = None + max_budget_source: str | None = None + tpm_limit: int | None = None + rpm_limit: int | None = None + budget_duration: str | None = None + allowed_models: tuple[str, ...] | None = None + + +class _BulkBudgetUpdateResponse(BaseModel): + data: tuple[_BulkBudgetResult, ...] + + +@pytest.fixture +def upstream(gateway: Gateway) -> Iterator[httpx.Client]: + with httpx.Client(base_url=gateway.upstream_url, timeout=15, trust_env=False) as client: + response: Final = client.get("/__observations") + assert response.status_code == 200, response.text + yield client + + +def _priced_model(scenario: Scenario) -> tuple[str, str]: + provider_model: Final = f"team-member-spend-{uuid.uuid4().hex}" + model: Final = scenario.model( + model=f"openai/{provider_model}", + input_cost_per_token=0.001, + output_cost_per_token=0.002, + ) + return model, provider_model + + +def _free_model(scenario: Scenario) -> tuple[str, str]: + provider_model: Final = f"team-member-free-{uuid.uuid4().hex}" + model: Final = scenario.model( + model=f"openai/{provider_model}", + input_cost_per_token=0, + output_cost_per_token=0, + ) + return model, provider_model + + +def _observed_models(upstream: httpx.Client) -> tuple[str, ...]: + response: Final = upstream.get("/__observations") + assert response.status_code == 200, response.text + observations: Final = _Observations.model_validate_json(response.text) + return tuple(string_value(request.body["model"]) for request in observations.requests) + + +def _chat(gateway: Gateway, model: str, key: str, text: str) -> httpx.Response: + return gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": text}]}, + key=key, + ) + + +def _team_info(gateway: Gateway, team_id: str) -> _TeamInfo: + response: Final = gateway.request("GET", "/team/info", params={"team_id": team_id}) + assert response.status_code == 200, response.text + return _TeamInfo.model_validate_json(response.text) + + +def _member_me(gateway: Gateway, team_id: str, key: str) -> _MemberMe: + response: Final = gateway.request("GET", f"/team/{team_id}/members/me", key=key) + assert response.status_code == 200, response.text + return _MemberMe.model_validate_json(response.text) + + +def _budget_state(team_id: str, user_id: str) -> tuple[_MemberBudgetState, ...]: + return tuple( + _MemberBudgetState.model_validate(row) + for row in read_rows( + "SELECT tm.user_id, tm.team_id, tm.budget_id, tm.spend, b.max_budget, b.tpm_limit, b.rpm_limit, " + "b.budget_duration, b.allowed_models, b.temp_budget_increase, " + "b.temp_budget_expiry::text AS temp_budget_expiry " + 'FROM "LiteLLM_TeamMembership" tm LEFT JOIN "LiteLLM_BudgetTable" b ON b.budget_id = tm.budget_id ' + "WHERE tm.team_id = %s AND tm.user_id = %s", + (team_id, user_id), + ) + ) + + +def _budget_record(budget_id: str) -> tuple[_BudgetRecord, ...]: + return tuple( + _BudgetRecord.model_validate(row) + for row in read_rows( + "SELECT budget_id, max_budget, tpm_limit, rpm_limit, budget_duration, allowed_models, " + 'temp_budget_increase, temp_budget_expiry FROM "LiteLLM_BudgetTable" WHERE budget_id = %s', + (budget_id,), + ) + ) + + +def _team_default_budget_id(team_id: str) -> str: + rows: Final = read_rows('SELECT metadata FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team_id,)) + assert len(rows) == 1, rows + metadata: Final = object_value(rows[0]["metadata"]) + return string_value(metadata["team_member_budget_id"]) + + +def _member_update(gateway: Gateway, body: dict[str, JsonValue]) -> _MemberUpdateResponse: + response: Final = gateway.request("POST", "/team/member_update", body) + assert response.status_code == 200, response.text + return _MemberUpdateResponse.model_validate_json(response.text) + + +def _budget_refused(response: httpx.Response) -> None: + assert response.status_code == 422, response.text + assert _ErrorResponse.model_validate_json(response.text).error.type == "budget_exceeded", response.text + + @pytest.mark.covers("spend.team_member.member_without_budget_gets_membership_row_and_spend") def test_member_added_without_any_budget_is_charged_on_its_membership_row(gateway: Gateway) -> None: with gateway.scenario() as scenario: @@ -97,3 +314,716 @@ def test_member_over_budget_is_blocked_when_redis_counter_reads_stale_low(gatewa assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text assert upstream.get("/__observations").json()["requests"] == [] assert float(cache.get(counter_key)) == pytest.approx(0.06), denied.text + + +def test_member_update_lowering_the_budget_below_spend_blocks_the_next_call( + gateway: Gateway, peer: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _priced_model(scenario) + team: Final = scenario.team(models=[model]) + user: Final = scenario.user() + added: Final = gateway.request( + "POST", "/team/member_add", {"team_id": team, "member": {"user_id": user, "role": "user"}} + ) + assert added.status_code == 200, added.text + initial_budget: Final = _member_update( + gateway, + {"team_id": team, "user_id": user, "max_budget_in_team": 1}, + ) + assert initial_budget == _MemberUpdateResponse(team_id=team, user_id=user, max_budget_in_team=1), initial_budget + key: Final = scenario.key(team_id=team, user_id=user) + personal_key: Final = scenario.key(user_id=user) + + first: Final = _chat(gateway, model, key, f"accrued member spend {uuid.uuid4().hex}") + assert first.status_code == 200, first.text + eventually( + lambda: _budget_state(team, user), + lambda values: len(values) == 1 and values[0].spend is not None and values[0].spend >= 0.06, + seconds=70, + ) + peer_first: Final = _chat(peer, model, key, f"accrued member spend on peer {uuid.uuid4().hex}") + assert peer_first.status_code == 200, peer_first.text + observed_before_update: Final = _observed_models(upstream) + assert observed_before_update == (provider_model, provider_model), repr(observed_before_update) + accrued: Final = eventually( + lambda: _budget_state(team, user), + lambda values: len(values) == 1 and values[0].spend is not None and values[0].spend >= 0.12, + seconds=70, + ) + assert accrued[0].spend == pytest.approx(0.12), accrued + cached_membership: Final = _member_me(gateway, team, key) + assert cached_membership.litellm_budget_table is not None, repr(cached_membership) + assert cached_membership.litellm_budget_table.max_budget == 1, repr(cached_membership) + + updated: Final = _member_update( + gateway, + { + "team_id": team, + "user_id": user, + "max_budget_in_team": 0.0001, + "budget_duration": "30d", + "tpm_limit": 1000, + "rpm_limit": 100, + }, + ) + assert updated == _MemberUpdateResponse( + team_id=team, + user_id=user, + max_budget_in_team=0.0001, + budget_duration="30d", + tpm_limit=1000, + rpm_limit=100, + ), repr(updated) + denied: Final = _chat(gateway, model, key, f"lowered member budget {uuid.uuid4().hex}") + _budget_refused(denied) + peer_denied: Final = _chat(peer, model, key, f"lowered member budget on peer {uuid.uuid4().hex}") + _budget_refused(peer_denied) + observed_after_denial: Final = _observed_models(upstream) + assert observed_after_denial == (), repr(observed_after_denial) + assert observed_before_update + observed_after_denial == (provider_model, provider_model), observed_after_denial + + state: Final = _budget_state(team, user) + assert len(state) == 1, state + assert state[0].user_id == user, state + assert state[0].team_id == team, state + assert state[0].spend == pytest.approx(0.12), state + assert state[0].max_budget == 0.0001, state + assert state[0].tpm_limit == 1000, state + assert state[0].rpm_limit == 100, state + assert state[0].budget_duration == "30d", state + team_info: Final = _team_info(gateway, team) + listed: Final = next(member for member in team_info.team_memberships if member.user_id == user) + assert listed.budget_id == state[0].budget_id, repr(listed) + assert listed.litellm_budget_table is not None, repr(listed) + assert listed.litellm_budget_table.max_budget == 0.0001, repr(listed) + assert listed.litellm_budget_table.tpm_limit == 1000, repr(listed) + assert listed.litellm_budget_table.rpm_limit == 100, repr(listed) + assert listed.litellm_budget_table.budget_duration == "30d", repr(listed) + member_info: Final = _member_me(gateway, team, personal_key) + assert member_info.budget_id == state[0].budget_id, repr(member_info) + assert member_info.litellm_budget_table is not None, repr(member_info) + assert member_info.litellm_budget_table.max_budget == 0.0001, repr(member_info) + assert member_info.litellm_budget_table.tpm_limit == 1000, repr(member_info) + assert member_info.litellm_budget_table.rpm_limit == 100, repr(member_info) + assert member_info.litellm_budget_table.budget_duration == "30d", repr(member_info) + + +@pytest.mark.parametrize( + "spell_expiry", + [ + pytest.param(_python_isoformat, id="python-isoformat"), + pytest.param(_javascript_iso_string, id="javascript-toisostring"), + ], +) +@pytest.mark.parametrize( + "budget_source", + [ + pytest.param("own-budget", id="own-budget"), + pytest.param("team-default", id="team-default"), + ], +) +def test_member_update_temp_budget_increase_readmits_a_blocked_member( + gateway: Gateway, + upstream: httpx.Client, + spell_expiry: Callable[[datetime], str], + budget_source: str, +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _priced_model(scenario) + team: Final = ( + scenario.team(models=[model], team_member_budget=0.05) + if budget_source == "team-default" + else scenario.team(models=[model]) + ) + user: Final = scenario.user() + member_add_body: Final = {"team_id": team, "member": {"user_id": user, "role": "user"}} + added: Final = gateway.request( + "POST", + "/team/member_add", + member_add_body | ({"max_budget_in_team": 0.05} if budget_source == "own-budget" else {}), + ) + assert added.status_code == 200, added.text + default_budget_id: Final = _team_default_budget_id(team) if budget_source == "team-default" else None + default_budget_before: Final = ( + _budget_record(default_budget_id) if default_budget_id is not None else () + ) + if default_budget_id is not None: + assert len(default_budget_before) == 1, repr(default_budget_before) + assert default_budget_before[0].max_budget == 0.05, default_budget_before + assert default_budget_before[0].temp_budget_increase is None, default_budget_before + assert default_budget_before[0].temp_budget_expiry is None, default_budget_before + initial_state: Final = _budget_state(team, user) + assert len(initial_state) == 1, initial_state + if budget_source == "own-budget": + assert initial_state[0].max_budget == 0.05, initial_state + else: + assert default_budget_id is not None + assert initial_state[0].budget_id == default_budget_id, initial_state + key: Final = scenario.key(team_id=team, user_id=user) + + first: Final = _chat(gateway, model, key, f"initial member budget {uuid.uuid4().hex}") + assert first.status_code == 200, first.text + observed_before_block: Final = _observed_models(upstream) + assert observed_before_block == (provider_model,), repr(observed_before_block) + accrued: Final = eventually( + lambda: _budget_state(team, user), + lambda values: len(values) == 1 and values[0].spend is not None and values[0].spend >= 0.06, + seconds=70, + ) + assert accrued[0].spend == pytest.approx(0.06), accrued + blocked: Final = _chat(gateway, model, key, f"exceeded member budget {uuid.uuid4().hex}") + _budget_refused(blocked) + observed_after_block: Final = _observed_models(upstream) + assert observed_after_block == (), repr(observed_after_block) + + expiry: Final = (datetime.now(timezone.utc) + timedelta(days=1)).replace(microsecond=0) + updated: Final = _member_update( + gateway, + { + "team_id": team, + "user_id": user, + "temp_budget_increase": 1.0, + "temp_budget_expiry": spell_expiry(expiry), + }, + ) + assert updated.temp_budget_expiry is not None, repr(updated) + assert updated.model_copy( + update={"temp_budget_expiry": updated.temp_budget_expiry.replace(tzinfo=timezone.utc)} + ) == _MemberUpdateResponse( + team_id=team, + user_id=user, + temp_budget_increase=1.0, + temp_budget_expiry=expiry, + ), repr(updated) + state: Final = _budget_state(team, user) + assert len(state) == 1, state + if budget_source == "own-budget": + assert state[0].max_budget == 0.05, state + else: + assert default_budget_id is not None + assert state[0].budget_id != default_budget_id, state + assert state[0].max_budget is None, state + default_budget_after: Final = _budget_record(default_budget_id) + assert default_budget_after == default_budget_before, repr(default_budget_after) + assert default_budget_after[0].max_budget == 0.05, default_budget_after + assert default_budget_after[0].temp_budget_increase is None, default_budget_after + assert default_budget_after[0].temp_budget_expiry is None, default_budget_after + assert state[0].temp_budget_increase == 1.0, state + assert state[0].temp_budget_expiry is not None, state + assert state[0].temp_budget_expiry.replace(tzinfo=timezone.utc) == expiry, state + team_info: Final = _team_info(gateway, team) + listed: Final = next(member for member in team_info.team_memberships if member.user_id == user) + assert listed.budget_id == state[0].budget_id, repr(listed) + assert listed.litellm_budget_table is not None, repr(listed) + assert listed.litellm_budget_table.budget_id == state[0].budget_id, repr(listed) + assert listed.litellm_budget_table.temp_budget_increase == 1.0, repr(listed) + assert listed.litellm_budget_table.temp_budget_expiry is not None, repr(listed) + assert listed.litellm_budget_table.temp_budget_expiry.replace(tzinfo=timezone.utc) == expiry, repr(listed) + member_info: Final = _member_me(gateway, team, key) + assert member_info.budget_id == state[0].budget_id, repr(member_info) + assert member_info.litellm_budget_table is not None, repr(member_info) + assert member_info.litellm_budget_table.budget_id == state[0].budget_id, repr(member_info) + assert member_info.litellm_budget_table.temp_budget_increase == 1.0, repr(member_info) + assert member_info.litellm_budget_table.temp_budget_expiry is not None, repr(member_info) + assert member_info.litellm_budget_table.temp_budget_expiry.replace(tzinfo=timezone.utc) == expiry, repr( + member_info + ) + + readmitted: Final = _chat(gateway, model, key, f"temporary member budget {uuid.uuid4().hex}") + assert readmitted.status_code == 200, readmitted.text + observed_after_readmission: Final = _observed_models(upstream) + assert observed_after_readmission == (provider_model,), repr(observed_after_readmission) + assert observed_before_block + observed_after_readmission == (provider_model, provider_model), repr( + (observed_before_block, observed_after_readmission) + ) + + +def test_member_rpm_limit_update_changes_runtime_admission(gateway: Gateway, upstream: httpx.Client) -> None: + pytest.skip("BUG: member_update leaves a warmed key using the old RPM limit") + with gateway.scenario() as scenario: + model, provider_model = _free_model(scenario) + team: Final = scenario.team(models=[model]) + user: Final = scenario.user() + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"user_id": user, "role": "user"}}, + ) + assert added.status_code == 200, added.text + key: Final = scenario.key(team_id=team, user_id=user) + + warm: Final = _chat(gateway, model, key, f"warm member key before rpm limit {uuid.uuid4().hex}") + assert warm.status_code == 200, warm.text + updated: Final = _member_update(gateway, {"team_id": team, "user_id": user, "rpm_limit": 1}) + assert updated.rpm_limit == 1, repr(updated) + + admitted: Final = _chat(gateway, model, key, f"member rpm admitted request {uuid.uuid4().hex}") + assert admitted.status_code == 200, admitted.text + limited: Final = _chat(gateway, model, key, f"member rpm limited request {uuid.uuid4().hex}") + assert limited.status_code == 429, limited.text + assert _ErrorResponse.model_validate_json(limited.text).error.type == "throttling_error", limited.text + observed: Final = _observed_models(upstream) + assert observed == (provider_model, provider_model), repr(observed) + + +def test_member_rpm_limit_update_changes_runtime_admission_for_cold_key( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _free_model(scenario) + team: Final = scenario.team(models=[model]) + user: Final = scenario.user() + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"user_id": user, "role": "user"}}, + ) + assert added.status_code == 200, added.text + key: Final = scenario.key(team_id=team, user_id=user) + + updated: Final = _member_update(gateway, {"team_id": team, "user_id": user, "rpm_limit": 1}) + assert updated.rpm_limit == 1, repr(updated) + + admitted: Final = _chat(gateway, model, key, f"cold member rpm admitted request {uuid.uuid4().hex}") + assert admitted.status_code == 200, admitted.text + limited: Final = _chat(gateway, model, key, f"cold member rpm limited request {uuid.uuid4().hex}") + assert limited.status_code == 429, limited.text + assert _ErrorResponse.model_validate_json(limited.text).error.type == "throttling_error", limited.text + observed: Final = _observed_models(upstream) + assert observed == (provider_model,), repr(observed) + + +def test_member_tpm_limit_update_changes_runtime_admission(gateway: Gateway, upstream: httpx.Client) -> None: + pytest.skip("BUG: member_update leaves a warmed key using the old TPM limit") + with gateway.scenario() as scenario: + model, provider_model = _free_model(scenario) + team: Final = scenario.team(models=[model]) + user: Final = scenario.user() + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"user_id": user, "role": "user"}}, + ) + assert added.status_code == 200, added.text + key: Final = scenario.key(team_id=team, user_id=user) + + warm: Final = _chat(gateway, model, key, f"warm member key before tpm limit {uuid.uuid4().hex}") + assert warm.status_code == 200, warm.text + updated: Final = _member_update(gateway, {"team_id": team, "user_id": user, "tpm_limit": 30}) + assert updated.tpm_limit == 30, repr(updated) + + admitted: Final = _chat(gateway, model, key, f"first member tpm-limited request {uuid.uuid4().hex}") + assert admitted.status_code == 200, admitted.text + limited: Final = _chat(gateway, model, key, f"second member tpm-limited request {uuid.uuid4().hex}") + assert limited.status_code == 429, limited.text + assert _ErrorResponse.model_validate_json(limited.text).error.type == "throttling_error", limited.text + observed: Final = _observed_models(upstream) + assert observed == (provider_model, provider_model), repr(observed) + + +def test_member_tpm_limit_update_changes_runtime_admission_for_cold_key( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _free_model(scenario) + team: Final = scenario.team(models=[model]) + user: Final = scenario.user() + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"user_id": user, "role": "user"}}, + ) + assert added.status_code == 200, added.text + key: Final = scenario.key(team_id=team, user_id=user) + + updated: Final = _member_update(gateway, {"team_id": team, "user_id": user, "tpm_limit": 30}) + assert updated.tpm_limit == 30, repr(updated) + + admitted: Final = _chat(gateway, model, key, f"cold member tpm admitted request {uuid.uuid4().hex}") + assert admitted.status_code == 200, admitted.text + limited: Final = _chat(gateway, model, key, f"cold member tpm limited request {uuid.uuid4().hex}") + assert limited.status_code == 429, limited.text + assert _ErrorResponse.model_validate_json(limited.text).error.type == "throttling_error", limited.text + observed: Final = _observed_models(upstream) + assert observed == (provider_model,), repr(observed) + + +def test_member_update_by_email_sets_models_role_and_forks_the_team_default_budget( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model_one, provider_one = _free_model(scenario) + model_two, provider_two = _free_model(scenario) + team: Final = scenario.team(models=[model_one, model_two], team_member_budget=25) + email_one: Final = f"member-{uuid.uuid4().hex}@example.com" + user_one: Final = scenario.user(user_email=email_one, user_role="internal_user") + user_two: Final = scenario.user(user_role="internal_user") + added: Final = gateway.request( + "POST", + "/team/member_add", + { + "team_id": team, + "member": [ + {"user_email": email_one, "role": "user"}, + {"user_id": user_two, "role": "user"}, + ], + }, + ) + assert added.status_code == 200, added.text + default_budget_id: Final = _team_default_budget_id(team) + default_budget_before: Final = _budget_record(default_budget_id) + assert len(default_budget_before) == 1, repr(default_budget_before) + initial_one: Final = _budget_state(team, user_one) + initial_two: Final = _budget_state(team, user_two) + assert len(initial_one) == 1 and initial_one[0].budget_id == default_budget_id, initial_one + assert len(initial_two) == 1 and initial_two[0].budget_id == default_budget_id, initial_two + + user_one_key: Final = scenario.key(team_id=team, user_id=user_one) + model_scope: Final = _member_update( + gateway, + {"team_id": team, "user_email": email_one, "allowed_models": [model_one]}, + ) + assert model_scope.user_id == user_one, repr(model_scope) + assert model_scope.allowed_models == [model_one], repr(model_scope) + scoped_state: Final = _budget_state(team, user_one) + assert len(scoped_state) == 1, scoped_state + assert scoped_state[0].allowed_models == [model_one], scoped_state + assert scoped_state[0].budget_id != default_budget_id, scoped_state + other_member_state: Final = _budget_state(team, user_two) + assert other_member_state[0].budget_id == default_budget_id, other_member_state + assert _budget_record(default_budget_id) == default_budget_before, repr(default_budget_before) + + allowed: Final = _chat(gateway, model_one, user_one_key, f"email allowed model {uuid.uuid4().hex}") + assert allowed.status_code == 200, allowed.text + denied: Final = _chat(gateway, model_two, user_one_key, f"email restricted model {uuid.uuid4().hex}") + assert denied.status_code == 403, denied.text + assert _ErrorResponse.model_validate_json(denied.text).error.type == "team_model_access_denied", denied.text + observed_before_clear: Final = _observed_models(upstream) + assert observed_before_clear == (provider_one,), repr(observed_before_clear) + + cleared_models: Final = _member_update( + gateway, + {"team_id": team, "user_email": email_one, "allowed_models": []}, + ) + assert cleared_models.user_id == user_one, repr(cleared_models) + assert cleared_models.allowed_models == [], repr(cleared_models) + cleared_state: Final = _budget_state(team, user_one) + assert len(cleared_state) == 1, cleared_state + assert cleared_state[0].allowed_models == [], cleared_state + unrestricted: Final = _chat(gateway, model_two, user_one_key, f"email cleared model scope {uuid.uuid4().hex}") + assert unrestricted.status_code == 200, unrestricted.text + observed_after_clear: Final = _observed_models(upstream) + assert observed_after_clear == (provider_two,), repr(observed_after_clear) + + budget_update: Final = _member_update( + gateway, + {"team_id": team, "user_email": email_one, "max_budget_in_team": 12}, + ) + assert budget_update.user_id == user_one, repr(budget_update) + assert budget_update.max_budget_in_team == 12, repr(budget_update) + user_one_state: Final = _budget_state(team, user_one) + user_two_state: Final = _budget_state(team, user_two) + assert len(user_one_state) == 1 and user_one_state[0].budget_id != default_budget_id, user_one_state + assert user_one_state[0].max_budget == 12, user_one_state + assert user_two_state == initial_two, user_two_state + assert _budget_record(default_budget_id) == default_budget_before, repr(default_budget_before) + team_info: Final = _team_info(gateway, team) + listed_one: Final = next(member for member in team_info.team_memberships if member.user_id == user_one) + assert listed_one.litellm_budget_table is not None, repr(listed_one) + assert listed_one.litellm_budget_table.max_budget == 12, repr(listed_one) + assert listed_one.litellm_budget_table.allowed_models == [], repr(listed_one) + member_info: Final = _member_me(gateway, team, user_one_key) + assert member_info.litellm_budget_table is not None, repr(member_info) + assert member_info.litellm_budget_table.max_budget == 12, repr(member_info) + assert member_info.litellm_budget_table.allowed_models == [], repr(member_info) + roster_member: Final = next( + member for member in team_info.team_info.members_with_roles if member.user_id == user_one + ) + assert roster_member.role == "user", repr(roster_member) + + new_member: Final = scenario.user() + user_key: Final = scenario.key(user_id=user_one) + denied_admin_action: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"user_id": new_member, "role": "user"}}, + key=user_key, + ) + assert denied_admin_action.status_code == 403, denied_admin_action.text + role_update: Final = _member_update( + gateway, + {"team_id": team, "user_email": email_one, "role": "admin"}, + ) + assert role_update.user_id == user_one, repr(role_update) + role_info: Final = _team_info(gateway, team) + admin_roster: Final = next( + member for member in role_info.team_info.members_with_roles if member.user_id == user_one + ) + assert admin_roster.role == "admin", repr(admin_roster) + roster_rows: Final = read_rows( + 'SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', + (team,), + ) + assert len(roster_rows) == 1, roster_rows + database_roster: Final = _TeamState.model_validate( + {"team_id": team, "members_with_roles": roster_rows[0]["members_with_roles"]} + ).members_with_roles + assert next(member for member in database_roster if member.user_id == user_one).role == "admin", repr( + database_roster + ) + allowed_admin_action: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"user_id": new_member, "role": "user"}}, + key=user_key, + ) + assert allowed_admin_action.status_code == 200, allowed_admin_action.text + assert observed_before_clear + observed_after_clear == (provider_one, provider_two), repr( + (observed_before_clear, observed_after_clear) + ) + + +def test_bulk_update_is_a_merge_patch_per_member(gateway: Gateway, upstream: httpx.Client) -> None: + with gateway.scenario() as scenario: + model_one, provider_one = _free_model(scenario) + model_two, _ = _free_model(scenario) + spend_model, provider_spend = _priced_model(scenario) + team: Final = scenario.team(models=[model_one, model_two, spend_model], team_member_budget=25) + user_one: Final = scenario.user() + email_two: Final = f"bulk-member-{uuid.uuid4().hex}@example.com" + user_two: Final = scenario.user(user_email=email_two, user_role="internal_user") + added: Final = gateway.request( + "POST", + "/team/member_add", + { + "team_id": team, + "member": [ + {"user_id": user_one, "role": "user"}, + {"user_email": email_two, "role": "user"}, + ], + }, + ) + assert added.status_code == 200, added.text + default_budget_id: Final = _team_default_budget_id(team) + default_budget_before: Final = _budget_record(default_budget_id) + key_one: Final = scenario.key(team_id=team, user_id=user_one) + key_two: Final = scenario.key(team_id=team, user_id=user_two) + personal_key_one: Final = scenario.key(user_id=user_one) + + first_chat: Final = _chat(gateway, spend_model, key_one, f"bulk member spend {uuid.uuid4().hex}") + assert first_chat.status_code == 200, first_chat.text + observed_before_update: Final = _observed_models(upstream) + assert observed_before_update == (provider_spend,), repr(observed_before_update) + accrued: Final = eventually( + lambda: _budget_state(team, user_one), + lambda values: len(values) == 1 and values[0].spend is not None and values[0].spend >= 0.06, + seconds=70, + ) + assert accrued[0].spend == pytest.approx(0.06), accrued + + update: Final = gateway.request( + "POST", + f"/management/v1/teams/{team}/members/bulk_update", + { + "members": [ + {"user_id": user_one, "max_budget_in_team": 0.0001, "rpm_limit": 7}, + {"user_email": email_two, "rpm_limit": 5, "allowed_models": [model_one]}, + ] + }, + ) + assert update.status_code == 200, update.text + update_response: Final = _BulkBudgetUpdateResponse.model_validate_json(update.text) + state_one: Final = _budget_state(team, user_one) + state_two: Final = _budget_state(team, user_two) + assert len(state_one) == 1 and len(state_two) == 1, (state_one, state_two) + assert state_one[0].budget_id is not None, state_one + assert state_two[0].budget_id is not None, state_two + assert state_one[0].budget_id != default_budget_id, state_one + assert state_two[0].budget_id != default_budget_id, state_two + assert update_response.data == ( + _BulkBudgetResult( + user_id=user_one, + success=True, + budget_id=state_one[0].budget_id, + max_budget=0.0001, + max_budget_source="member", + rpm_limit=7, + allowed_models=tuple(state_one[0].allowed_models or ()), + ), + _BulkBudgetResult( + user_id=user_two, + user_email=email_two, + success=True, + budget_id=state_two[0].budget_id, + max_budget=25, + max_budget_source="member", + rpm_limit=5, + allowed_models=(model_one,), + ), + ), update.text + assert state_one[0].max_budget == 0.0001 and state_one[0].rpm_limit == 7, state_one + assert state_two[0].max_budget == 25 and state_two[0].rpm_limit == 5, state_two + assert state_two[0].allowed_models == [model_one], state_two + assert _budget_record(default_budget_id) == default_budget_before, repr(default_budget_before) + info: Final = _team_info(gateway, team) + info_one: Final = next(member for member in info.team_memberships if member.user_id == user_one) + assert info_one.litellm_budget_table is not None, repr(info_one) + assert info_one.litellm_budget_table.max_budget == 0.0001, repr(info_one) + assert info_one.litellm_budget_table.rpm_limit == 7, repr(info_one) + me_one: Final = _member_me(gateway, team, personal_key_one) + assert me_one.litellm_budget_table is not None, repr(me_one) + assert me_one.litellm_budget_table.max_budget == 0.0001, repr(me_one) + assert me_one.litellm_budget_table.rpm_limit == 7, repr(me_one) + + over_budget: Final = _chat(gateway, spend_model, key_one, f"bulk member blocked {uuid.uuid4().hex}") + _budget_refused(over_budget) + observed_after_over_budget: Final = _observed_models(upstream) + assert observed_after_over_budget == (), repr(observed_after_over_budget) + allowed_two: Final = _chat(gateway, model_one, key_two, f"bulk member scoped allowed {uuid.uuid4().hex}") + assert allowed_two.status_code == 200, allowed_two.text + denied_two: Final = _chat(gateway, model_two, key_two, f"bulk member scoped denied {uuid.uuid4().hex}") + assert denied_two.status_code == 403, denied_two.text + assert _ErrorResponse.model_validate_json(denied_two.text).error.type == "team_model_access_denied", ( + denied_two.text + ) + observed_after_first_patch: Final = _observed_models(upstream) + assert observed_after_first_patch == (provider_one,), repr(observed_after_first_patch) + + cleared_max: Final = gateway.request( + "POST", + f"/management/v1/teams/{team}/members/bulk_update", + {"members": [{"user_id": user_one, "max_budget_in_team": None}]}, + ) + assert cleared_max.status_code == 200, cleared_max.text + cleared_max_response: Final = _BulkBudgetUpdateResponse.model_validate_json(cleared_max.text) + cleared_max_state: Final = _budget_state(team, user_one) + assert len(cleared_max_state) == 1, cleared_max_state + assert cleared_max_response.data == ( + _BulkBudgetResult( + user_id=user_one, + success=True, + budget_id=cleared_max_state[0].budget_id, + max_budget=25, + max_budget_source="team_default", + rpm_limit=7, + allowed_models=tuple(cleared_max_state[0].allowed_models or ()), + ), + ), cleared_max.text + assert cleared_max_state[0].budget_id == state_one[0].budget_id, cleared_max_state + assert cleared_max_state[0].max_budget is None, cleared_max_state + assert cleared_max_state[0].rpm_limit == 7, cleared_max_state + assert _budget_record(default_budget_id) == default_budget_before, repr(default_budget_before) + me_after_max_clear: Final = _member_me(gateway, team, personal_key_one) + assert me_after_max_clear.litellm_budget_table is not None, repr(me_after_max_clear) + assert me_after_max_clear.litellm_budget_table.max_budget is None, repr(me_after_max_clear) + assert me_after_max_clear.litellm_budget_table.rpm_limit == 7, repr(me_after_max_clear) + admitted_again: Final = _chat(gateway, spend_model, key_one, f"bulk member readmitted {uuid.uuid4().hex}") + assert admitted_again.status_code == 200, admitted_again.text + observed_after_max_clear: Final = _observed_models(upstream) + assert observed_after_max_clear == (provider_spend,), repr(observed_after_max_clear) + + assert observed_before_update + observed_after_first_patch + observed_after_max_clear == ( + provider_spend, + provider_one, + provider_spend, + ), repr((observed_before_update, observed_after_first_patch, observed_after_max_clear)) + + +def test_bulk_update_restores_team_default_when_last_member_limit_is_cleared( + gateway: Gateway, upstream: httpx.Client +) -> None: + with gateway.scenario() as scenario: + model, provider_model = _priced_model(scenario) + team: Final = scenario.team(models=[model], team_member_budget=0.0001) + user: Final = scenario.user() + added: Final = gateway.request( + "POST", + "/team/member_add", + {"team_id": team, "member": {"user_id": user, "role": "user"}}, + ) + assert added.status_code == 200, added.text + default_budget_id: Final = _team_default_budget_id(team) + default_budget_before: Final = _budget_record(default_budget_id) + initial_state: Final = _budget_state(team, user) + assert len(initial_state) == 1 and initial_state[0].budget_id == default_budget_id, initial_state + key: Final = scenario.key(team_id=team, user_id=user, models=[model]) + increased: Final = gateway.request( + "POST", + f"/management/v1/teams/{team}/members/bulk_update", + {"members": [{"user_id": user, "max_budget_in_team": 5}]}, + ) + assert increased.status_code == 200, increased.text + increased_response: Final = _BulkBudgetUpdateResponse.model_validate_json(increased.text) + increased_state: Final = _budget_state(team, user) + assert len(increased_state) == 1, increased_state + assert increased_state[0].budget_id is not None and increased_state[0].budget_id != default_budget_id, ( + increased_state + ) + assert increased_response.data == ( + _BulkBudgetResult( + user_id=user, + success=True, + budget_id=increased_state[0].budget_id, + max_budget=5, + max_budget_source="member", + allowed_models=tuple(increased_state[0].allowed_models or ()), + ), + ), increased.text + first_chat: Final = _chat(gateway, model, key, f"member raised limit {uuid.uuid4().hex}") + assert first_chat.status_code == 200, first_chat.text + first_observations: Final = _observed_models(upstream) + assert first_observations == (provider_model,), repr(first_observations) + accrued: Final = eventually( + lambda: _budget_state(team, user), + lambda values: len(values) == 1 and values[0].spend is not None and values[0].spend >= 0.06, + seconds=70, + ) + assert accrued[0].spend == pytest.approx(0.06), accrued + second_chat: Final = _chat(gateway, model, key, f"member still under limit {uuid.uuid4().hex}") + assert second_chat.status_code == 200, second_chat.text + second_observations: Final = _observed_models(upstream) + assert second_observations == (provider_model,), repr(second_observations) + cleared: Final = gateway.request( + "POST", + f"/management/v1/teams/{team}/members/bulk_update", + {"members": [{"user_id": user, "max_budget_in_team": None}]}, + ) + assert cleared.status_code == 200, cleared.text + cleared_response: Final = _BulkBudgetUpdateResponse.model_validate_json(cleared.text) + final_state: Final = _budget_state(team, user) + assert len(final_state) == 1, final_state + assert final_state[0].budget_id is None, final_state + assert cleared_response.data == ( + _BulkBudgetResult( + user_id=user, + success=True, + budget_id=None, + max_budget=0.0001, + max_budget_source="team_default", + tpm_limit=None, + rpm_limit=None, + budget_duration=None, + allowed_models=None, + ), + ), cleared.text + assert read_rows( + 'SELECT budget_id FROM "LiteLLM_TeamMembership" WHERE team_id=%s AND user_id=%s', + (team, user), + ) == [{"budget_id": None}], cleared.text + assert _budget_record(default_budget_id) == default_budget_before, repr(default_budget_before) + refused: Final = _chat(gateway, model, key, f"member default cap reached {uuid.uuid4().hex}") + _budget_refused(refused) + observed_after_refusal: Final = _observed_models(upstream) + assert observed_after_refusal == (), repr(observed_after_refusal) + assert first_observations + second_observations == (provider_model, provider_model), repr( + (first_observations, second_observations) + ) + + +@pytest.fixture(autouse=True) +def _drain_unobserved_requests(gateway: Gateway) -> Iterator[None]: + yield + with httpx.Client(base_url=gateway.upstream_url, timeout=15, trust_env=False) as client: + response: Final = client.get("/__observations") + assert response.status_code == 200, response.text