ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(proxy): behavior-pinning matrix for team management endpoints (#28441)

Browse Source 
* test(proxy): behavior-pinning matrix for team management endpoints

PR2 (Team Tier-1) of the management-endpoint behavior-pinning effort.
Extends the tests/proxy_behavior/management/ harness PR1 built and adds
the actor x target-resource authz matrix for the 7 team endpoints:
/team/new, /team/info, /team/list, /team/update, /team/member_add,
/team/member_delete, /team/member_update.

Tests-only, no production code changes.

Harness extensions:
- actors.py: ORG_B_ADMIN actor (org admin of ORG_B) and TEAM_GAMMA (an
  ORG_A team with no actor members), so team-targeting endpoints get a
  clean own / same-org-other / cross-org target axis.
- conftest.py: create_scratch_team() raw-seeds target teams without
  /team/new side effects; the scratch teardown now also strips dangling
  scratch-team refs from LiteLLM_UserTable.teams.

156 new scenarios; status codes pinned to observed handler behavior.

* test(proxy): record mutmut run blockers in PR2 triage doc

Attempted a scoped local mutmut run for G5; it did not complete. Record
the three concrete blockers in mutmut_triage/pr2-team-tier1.md so the next
attempt has a head start:

1. mutmut's mutants/ sandbox is import-shadowed by the worktree source.
2. the legacy mock suite and the real-DB behavior suite cannot share a
   pytest session (mock suite globally patches prisma_client).
3. the CI mutation-test.yml workflow starts no Postgres, so its stats
   phase now aborts on the behavior-suite tests PR1 added to tests_dir.

mutmut stays a deferred follow-up (as in PR1); the binding pre-merge
signal remains the behavior matrix (G1) and the G4 regression-replay.

* test(proxy): drop suite README + triage doc, trim test comments

Remove the two prose docs from the behavior suite (README.md and
mutmut_triage/pr2-team-tier1.md) and tighten the comment blocks on the
team test files + harness down to the load-bearing parts (the gate each
matrix pins, plus genuinely surprising results). No behavior change —
all 286 scenarios still pass.

* test(proxy): remove mutmut tests_dir comment
This commit is contained in:
yuneng-jiang 2026-05-21 16:57:25 -07:00 committed by GitHub
parent 10bd7406e0
commit 67e6e5e1df
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 901 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
pyproject.toml
View File

@ -288,11 +288,6 @@ paths_to_mutate = [
]
tests_dir = [
"tests/test_litellm/proxy/management_endpoints/",
# PR1 (key Tier-1) behavior-pinning suite. Manual mutmut runs
# (.github/workflows/mutation-test.yml) include this directory so the
# behavior matrix contributes to mutation-score signal alongside the
# legacy mock suite. See tests/proxy_behavior/management/README.md
# for the G5 triage protocol.
"tests/proxy_behavior/management/",
]
also_copy = [

24
tests/proxy_behavior/management/actors.py
View File

@ -1,4 +1,4 @@
"""8-actor read-world seed for the authz matrix tests."""
"""Read-world seed for the authz matrix tests: 2 orgs, 3 teams, 9 actors."""
import enum
import uuid
@ -20,6 +20,7 @@ class Actor(str, enum.Enum):
UNRELATED_SAME_ORG = "unrelated_same_org"
CROSS_ORG_USER = "cross_org_user"
SERVICE_ACCOUNT = "service_account"
ORG_B_ADMIN = "org_b_admin"
PREFIX = "behavior-pin-"
@ -27,6 +28,7 @@ ORG_A = PREFIX + "org-a"
ORG_B = PREFIX + "org-b"
TEAM_ALPHA = PREFIX + "team-alpha"
TEAM_BETA = PREFIX + "team-beta"
TEAM_GAMMA = PREFIX + "team-gamma"
BUDGET_ID = PREFIX + "budget"
@ -43,6 +45,7 @@ class World:
org_b_id: str
team_alpha_id: str
team_beta_id: str
team_gamma_id: str
keys: Dict[Actor, SeededKey]
@ -92,6 +95,11 @@ def _actor_profile() -> Dict[Actor, Dict[str, Any]]:
"team_id": TEAM_ALPHA,
"organization_id": ORG_A,
},
Actor.ORG_B_ADMIN: {
"user_role": LitellmUserRoles.ORG_ADMIN.value,
"team_id": None,
"organization_id": ORG_B,
},
}
@ -195,6 +203,18 @@ async def seed_world(prisma: PrismaClient) -> World:
),
}
)
# TEAM_GAMMA: ORG_A team with no actor members — the "same-org,
# not-my-team" read target.
await prisma.db.litellm_teamtable.create(
data={
"team_id": TEAM_GAMMA,
"team_alias": "gamma-1",
"organization_id": ORG_A,
"admins": [],
"members": [],
"members_with_roles": Json([]),
}
)
for actor, org_id, role in [
(Actor.ORG_ADMIN, ORG_A, "org_admin"),
@ -204,6 +224,7 @@ async def seed_world(prisma: PrismaClient) -> World:
(Actor.UNRELATED_SAME_ORG, ORG_A, "internal_user"),
(Actor.SERVICE_ACCOUNT, ORG_A, "internal_user"),
(Actor.CROSS_ORG_USER, ORG_B, "internal_user"),
(Actor.ORG_B_ADMIN, ORG_B, "org_admin"),
]:
await prisma.db.litellm_organizationmembership.create(
data={
@ -253,5 +274,6 @@ async def seed_world(prisma: PrismaClient) -> World:
org_b_id=ORG_B,
team_alpha_id=TEAM_ALPHA,
team_beta_id=TEAM_BETA,
team_gamma_id=TEAM_GAMMA,
keys=keys,
)

50
tests/proxy_behavior/management/conftest.py
View File

@ -9,6 +9,7 @@ from typing import Any, AsyncIterator, Dict, Optional
import httpx
import pytest_asyncio
import yaml
from prisma import Json
MASTER_KEY = "sk-1234"
@ -124,6 +125,42 @@ async def create_scratch_key(
return resp.json()["key"]
async def create_scratch_team(
prisma,
team_id: str,
*,
organization_id: Optional[str] = None,
admin_user_ids: Optional[list] = None,
member_user_ids: Optional[list] = None,
) -> str:
"""Raw-seed a scratch-tagged team row; returns its team_id.
The target team for the team write matrices (update / member_*). Raw
prisma (not POST /team/new) avoids creation side effects — no creator
auto-add, no membership rows written onto the world's users — so seeding
never mutates the immutable read-world. The authz gates read the team's
members_with_roles JSON, so a raw-seeded team exercises them exactly as
a /team/new-created team would. team_id must start with the scratch
prefix so the `scratch` fixture reclaims the row.
"""
admin_user_ids = list(admin_user_ids or [])
member_user_ids = list(member_user_ids or [])
members_with_roles = [
{"user_id": uid, "role": "admin"} for uid in admin_user_ids
] + [{"user_id": uid, "role": "user"} for uid in member_user_ids]
data: Dict[str, Any] = {
"team_id": team_id,
"team_alias": team_id,
"admins": admin_user_ids,
"members": admin_user_ids + member_user_ids,
"members_with_roles": Json(members_with_roles),
}
if organization_id is not None:
data["organization_id"] = organization_id
await prisma.db.litellm_teamtable.create(data=data)
return team_id
@pytest_asyncio.fixture
async def scratch(prisma):
handle = Scratch(prefix=f"{SCRATCH_PREFIX}{uuid.uuid4().hex[:12]}")
@ -154,3 +191,16 @@ async def scratch(prisma):
await prisma.db.litellm_budgettable.delete_many(
where={"budget_id": {"startswith": handle.prefix}}
)
# /team/member_add writes LiteLLM_UserTable.teams; the available-team
# self-join writes it on a world actor whose row must survive. Strip
# dangling scratch-team refs so the read-world stays immutable.
polluted = await prisma.db.litellm_usertable.find_many(
where={"teams": {"isEmpty": False}}
)
for user in polluted:
cleaned = [t for t in user.teams if not t.startswith(handle.prefix)]
if cleaned != list(user.teams):
await prisma.db.litellm_usertable.update(
where={"user_id": user.user_id},
data={"teams": {"set": cleaned}},
)

70
tests/proxy_behavior/management/test_team_info.py Normal file
View File

@ -0,0 +1,70 @@
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# GET /team/info — actor x team-target authz matrix, pinned against
# validate_membership(): a team is readable by a proxy admin, a key whose
# own team_id matches, a listed member, or an org admin of the team's org;
# everything else is 403. TEAM_GAMMA has no members, so only PROXY_ADMIN
# and ORG_A's org admin can read it.
_SCENARIOS = [
("alpha/proxy_admin", Actor.PROXY_ADMIN, "alpha", 200),
("alpha/org_admin", Actor.ORG_ADMIN, "alpha", 200),
("alpha/team_admin", Actor.TEAM_ADMIN, "alpha", 200),
("alpha/internal_user", Actor.INTERNAL_USER, "alpha", 200),
("alpha/owner", Actor.OWNER, "alpha", 200),
("alpha/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "alpha", 200),
("alpha/cross_org_user", Actor.CROSS_ORG_USER, "alpha", 403),
("alpha/service_account", Actor.SERVICE_ACCOUNT, "alpha", 200),
("alpha/org_b_admin", Actor.ORG_B_ADMIN, "alpha", 403),
("gamma/proxy_admin", Actor.PROXY_ADMIN, "gamma", 200),
("gamma/org_admin", Actor.ORG_ADMIN, "gamma", 200),
("gamma/team_admin", Actor.TEAM_ADMIN, "gamma", 403),
("gamma/internal_user", Actor.INTERNAL_USER, "gamma", 403),
("gamma/owner", Actor.OWNER, "gamma", 403),
("gamma/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "gamma", 403),
("gamma/cross_org_user", Actor.CROSS_ORG_USER, "gamma", 403),
("gamma/service_account", Actor.SERVICE_ACCOUNT, "gamma", 403),
("gamma/org_b_admin", Actor.ORG_B_ADMIN, "gamma", 403),
("beta/proxy_admin", Actor.PROXY_ADMIN, "beta", 200),
("beta/org_admin", Actor.ORG_ADMIN, "beta", 403),
("beta/team_admin", Actor.TEAM_ADMIN, "beta", 403),
("beta/internal_user", Actor.INTERNAL_USER, "beta", 403),
("beta/owner", Actor.OWNER, "beta", 403),
("beta/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "beta", 403),
("beta/cross_org_user", Actor.CROSS_ORG_USER, "beta", 200),
("beta/service_account", Actor.SERVICE_ACCOUNT, "beta", 403),
("beta/org_b_admin", Actor.ORG_B_ADMIN, "beta", 200),
]
@pytest.mark.parametrize(
"actor,target,expected_status",
[(a, t, s) for (_id, a, t, s) in _SCENARIOS],
ids=[s[0] for s in _SCENARIOS],
)
async def test_team_info_authz_matrix(
actor: Actor, target: str, expected_status: int, proxy_client, world
):
caller = world.keys[actor]
target_team_id = {
"alpha": world.team_alpha_id,
"gamma": world.team_gamma_id,
"beta": world.team_beta_id,
}[target]
resp = await proxy_client.get(
f"/team/info?team_id={target_team_id}",
headers={"Authorization": f"Bearer {caller.cleartext}"},
)
assert (
resp.status_code == expected_status
), f"{actor.value} -> {target}: {resp.status_code} {resp.text}"
if expected_status == 200:
body = resp.json()
assert body["team_id"] == target_team_id
assert body["team_info"]["team_id"] == target_team_id

105
tests/proxy_behavior/management/test_team_list.py Normal file
View File

@ -0,0 +1,105 @@
from typing import FrozenSet, Optional
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# The behavior DB may hold teams beyond the three seeded ones, so every
# assertion intersects the returned team_ids with the known seeded set.
def _seeded_visible(resp_json, world) -> set:
known = {
world.team_alpha_id: "alpha",
world.team_beta_id: "beta",
world.team_gamma_id: "gamma",
}
return {
known[entry["team_id"]]
for entry in resp_json
if isinstance(entry, dict) and entry.get("team_id") in known
}
# Family 1 — bare GET /team/list (no query params). _authorize_and_filter_teams
# authorizes only an admin view (proxy admin) or an org admin; everyone else
# is 401. An org admin sees every team in its org(s).
_BARE = [
("proxy_admin", Actor.PROXY_ADMIN, 200, {"alpha", "beta", "gamma"}),
("org_admin", Actor.ORG_ADMIN, 200, {"alpha", "gamma"}),
("team_admin", Actor.TEAM_ADMIN, 401, None),
("internal_user", Actor.INTERNAL_USER, 401, None),
("owner", Actor.OWNER, 401, None),
("unrelated_same_org", Actor.UNRELATED_SAME_ORG, 401, None),
("cross_org_user", Actor.CROSS_ORG_USER, 401, None),
("service_account", Actor.SERVICE_ACCOUNT, 401, None),
("org_b_admin", Actor.ORG_B_ADMIN, 200, {"beta"}),
]
@pytest.mark.parametrize(
"actor,expected_status,expected_visible",
[(a, s, v) for (_id, a, s, v) in _BARE],
ids=[s[0] for s in _BARE],
)
async def test_team_list_bare_authz(
actor: Actor,
expected_status: int,
expected_visible: Optional[set],
proxy_client,
world,
):
caller = world.keys[actor]
resp = await proxy_client.get(
"/team/list",
headers={"Authorization": f"Bearer {caller.cleartext}"},
)
assert (
resp.status_code == expected_status
), f"{actor.value}: {resp.status_code} {resp.text}"
if expected_status == 200:
visible = _seeded_visible(resp.json(), world)
assert visible == expected_visible, (
f"{actor.value}: expected {sorted(expected_visible)}, "
f"got {sorted(visible)}"
)
# Family 2 — GET /team/list?user_id=<caller's own id> ("own query"). Every
# actor may query its own teams (200); the result is exactly the teams it
# belongs to. A user_id filter scopes proxy/org admins to their own
# membership too — the broad admin view from family 1 does not carry over.
_OWN = {
Actor.PROXY_ADMIN: frozenset(),
Actor.ORG_ADMIN: frozenset(),
Actor.TEAM_ADMIN: frozenset({"alpha"}),
Actor.INTERNAL_USER: frozenset({"alpha"}),
Actor.OWNER: frozenset({"alpha"}),
Actor.UNRELATED_SAME_ORG: frozenset({"alpha"}),
Actor.CROSS_ORG_USER: frozenset({"beta"}),
Actor.SERVICE_ACCOUNT: frozenset({"alpha"}),
Actor.ORG_B_ADMIN: frozenset(),
}
@pytest.mark.parametrize(
"actor,expected_visible",
list(_OWN.items()),
ids=[a.value for a in _OWN],
)
async def test_team_list_own_query(
actor: Actor, expected_visible: FrozenSet[str], proxy_client, world
):
caller = world.keys[actor]
resp = await proxy_client.get(
f"/team/list?user_id={caller.user_id}",
headers={"Authorization": f"Bearer {caller.cleartext}"},
)
assert resp.status_code == 200, f"{actor.value}: {resp.status_code} {resp.text}"
visible = _seeded_visible(resp.json(), world)
assert visible == set(expected_visible), (
f"{actor.value}: expected {sorted(expected_visible)}, " f"got {sorted(visible)}"
)

149
tests/proxy_behavior/management/test_team_member_add.py Normal file
View File

@ -0,0 +1,149 @@
import litellm
import pytest
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/member_add — actor x team-shape matrix, pinned against
# _validate_team_member_add_permissions: PROXY_ADMIN, the team's team admin,
# or an org admin of the team's org may add members; everyone else is 403.
# Unlike /team/update there is no route gate in front, so the team-admin
# branch is reachable (TEAM_ADMIN, an internal_user, is allowed on its team).
_MATRIX = [
("alpha/proxy_admin", Actor.PROXY_ADMIN, "alpha", 200),
("alpha/org_admin", Actor.ORG_ADMIN, "alpha", 200),
("alpha/team_admin", Actor.TEAM_ADMIN, "alpha", 200),
("alpha/internal_user", Actor.INTERNAL_USER, "alpha", 403),
("alpha/owner", Actor.OWNER, "alpha", 403),
("alpha/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "alpha", 403),
("alpha/cross_org_user", Actor.CROSS_ORG_USER, "alpha", 403),
("alpha/service_account", Actor.SERVICE_ACCOUNT, "alpha", 403),
("alpha/org_b_admin", Actor.ORG_B_ADMIN, "alpha", 403),
("beta/proxy_admin", Actor.PROXY_ADMIN, "beta", 200),
("beta/org_admin", Actor.ORG_ADMIN, "beta", 403),
("beta/team_admin", Actor.TEAM_ADMIN, "beta", 403),
("beta/internal_user", Actor.INTERNAL_USER, "beta", 403),
("beta/owner", Actor.OWNER, "beta", 403),
("beta/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "beta", 403),
("beta/cross_org_user", Actor.CROSS_ORG_USER, "beta", 403),
("beta/service_account", Actor.SERVICE_ACCOUNT, "beta", 403),
("beta/org_b_admin", Actor.ORG_B_ADMIN, "beta", 200),
]
async def _seed_target(prisma, world, shape: str, team_id: str) -> None:
if shape == "alpha":
await create_scratch_team(
prisma,
team_id,
organization_id=world.org_a_id,
admin_user_ids=[world.keys[Actor.TEAM_ADMIN].user_id],
)
elif shape == "beta":
await create_scratch_team(prisma, team_id, organization_id=world.org_b_id)
else: # pragma: no cover - guard
pytest.fail(f"unknown shape={shape}")
def _member_ids(row) -> list:
return [m["user_id"] for m in (row.members_with_roles or [])]
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_member_add_authz_matrix(
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
await _seed_target(prisma, world, shape, scratch.prefix)
caller = world.keys[actor]
new_member_id = scratch.tag("newmember")
resp = await proxy_client.post(
"/team/member_add",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={
"team_id": scratch.prefix,
"member": {"user_id": new_member_id, "role": "user"},
},
)
assert (
resp.status_code == expected_status
), f"{actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert new_member_id in _member_ids(row)
else:
assert new_member_id not in _member_ids(row), "denied but member added"
# Available-team self-join: a non-admin caller may add ITSELF to a team listed
# in litellm.default_internal_user_params["available_teams"], but the bypass
# must not escalate to role=admin or inject another user.
_SELF_JOIN = [
("self_as_user", "self", "user", 200),
("self_as_admin", "self", "admin", 403),
("other_as_user", "other", "user", 403),
]
@pytest.mark.parametrize(
"who,role,expected_status",
[(w, r, s) for (_id, w, r, s) in _SELF_JOIN],
ids=[s[0] for s in _SELF_JOIN],
)
async def test_team_member_add_available_team_self_join(
who: str,
role: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
monkeypatch,
):
# Org-less team with no admins: the INTERNAL_USER caller is neither team
# nor org admin, so it lands on the available-team branch.
await create_scratch_team(prisma, scratch.prefix)
monkeypatch.setattr(
litellm, "default_internal_user_params", {"available_teams": [scratch.prefix]}
)
caller = world.keys[Actor.INTERNAL_USER]
member_id = caller.user_id if who == "self" else world.keys[Actor.OWNER].user_id
resp = await proxy_client.post(
"/team/member_add",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={
"team_id": scratch.prefix,
"member": {"user_id": member_id, "role": role},
},
)
assert (
resp.status_code == expected_status
), f"{who}/{role}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert member_id in _member_ids(row)
else:
assert member_id not in _member_ids(row), "denied but member added"

92
tests/proxy_behavior/management/test_team_member_delete.py Normal file
View File

@ -0,0 +1,92 @@
import pytest
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/member_delete — actor x team-shape matrix. The scratch team is
# raw-seeded with a victim member already in it; PROXY_ADMIN, the team's team
# admin, or an org admin of the team's org may remove members; else 403.
_MATRIX = [
("alpha/proxy_admin", Actor.PROXY_ADMIN, "alpha", 200),
("alpha/org_admin", Actor.ORG_ADMIN, "alpha", 200),
("alpha/team_admin", Actor.TEAM_ADMIN, "alpha", 200),
("alpha/internal_user", Actor.INTERNAL_USER, "alpha", 403),
("alpha/owner", Actor.OWNER, "alpha", 403),
("alpha/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "alpha", 403),
("alpha/cross_org_user", Actor.CROSS_ORG_USER, "alpha", 403),
("alpha/service_account", Actor.SERVICE_ACCOUNT, "alpha", 403),
("alpha/org_b_admin", Actor.ORG_B_ADMIN, "alpha", 403),
("beta/proxy_admin", Actor.PROXY_ADMIN, "beta", 200),
("beta/org_admin", Actor.ORG_ADMIN, "beta", 403),
("beta/team_admin", Actor.TEAM_ADMIN, "beta", 403),
("beta/internal_user", Actor.INTERNAL_USER, "beta", 403),
("beta/owner", Actor.OWNER, "beta", 403),
("beta/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "beta", 403),
("beta/cross_org_user", Actor.CROSS_ORG_USER, "beta", 403),
("beta/service_account", Actor.SERVICE_ACCOUNT, "beta", 403),
("beta/org_b_admin", Actor.ORG_B_ADMIN, "beta", 200),
]
async def _seed_target(prisma, world, shape: str, team_id: str, victim_id: str) -> None:
if shape == "alpha":
await create_scratch_team(
prisma,
team_id,
organization_id=world.org_a_id,
admin_user_ids=[world.keys[Actor.TEAM_ADMIN].user_id],
member_user_ids=[victim_id],
)
elif shape == "beta":
await create_scratch_team(
prisma,
team_id,
organization_id=world.org_b_id,
member_user_ids=[victim_id],
)
else: # pragma: no cover - guard
pytest.fail(f"unknown shape={shape}")
def _member_ids(row) -> list:
return [m["user_id"] for m in (row.members_with_roles or [])]
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_member_delete_authz_matrix(
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
victim_id = scratch.tag("victim")
await _seed_target(prisma, world, shape, scratch.prefix, victim_id)
caller = world.keys[actor]
resp = await proxy_client.post(
"/team/member_delete",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"team_id": scratch.prefix, "user_id": victim_id},
)
assert (
resp.status_code == expected_status
), f"{actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert victim_id not in _member_ids(row)
else:
assert victim_id in _member_ids(row), "denied but member removed"

97
tests/proxy_behavior/management/test_team_member_update.py Normal file
View File

@ -0,0 +1,97 @@
import pytest
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/member_update — actor x team-shape matrix. The scratch team is
# raw-seeded with a "user"-role member; each scenario tries to promote it to
# "admin". PROXY_ADMIN, the team's team admin, or an org admin of the team's
# org may update members; else 403. (The harness forces premium_user, so the
# promotion does not hit the admin-role premium gate.)
_MATRIX = [
("alpha/proxy_admin", Actor.PROXY_ADMIN, "alpha", 200),
("alpha/org_admin", Actor.ORG_ADMIN, "alpha", 200),
("alpha/team_admin", Actor.TEAM_ADMIN, "alpha", 200),
("alpha/internal_user", Actor.INTERNAL_USER, "alpha", 403),
("alpha/owner", Actor.OWNER, "alpha", 403),
("alpha/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "alpha", 403),
("alpha/cross_org_user", Actor.CROSS_ORG_USER, "alpha", 403),
("alpha/service_account", Actor.SERVICE_ACCOUNT, "alpha", 403),
("alpha/org_b_admin", Actor.ORG_B_ADMIN, "alpha", 403),
("beta/proxy_admin", Actor.PROXY_ADMIN, "beta", 200),
("beta/org_admin", Actor.ORG_ADMIN, "beta", 403),
("beta/team_admin", Actor.TEAM_ADMIN, "beta", 403),
("beta/internal_user", Actor.INTERNAL_USER, "beta", 403),
("beta/owner", Actor.OWNER, "beta", 403),
("beta/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "beta", 403),
("beta/cross_org_user", Actor.CROSS_ORG_USER, "beta", 403),
("beta/service_account", Actor.SERVICE_ACCOUNT, "beta", 403),
("beta/org_b_admin", Actor.ORG_B_ADMIN, "beta", 200),
]
async def _seed_target(prisma, world, shape: str, team_id: str, member_id: str) -> None:
if shape == "alpha":
await create_scratch_team(
prisma,
team_id,
organization_id=world.org_a_id,
admin_user_ids=[world.keys[Actor.TEAM_ADMIN].user_id],
member_user_ids=[member_id],
)
elif shape == "beta":
await create_scratch_team(
prisma,
team_id,
organization_id=world.org_b_id,
member_user_ids=[member_id],
)
else: # pragma: no cover - guard
pytest.fail(f"unknown shape={shape}")
def _role_of(row, user_id: str):
for m in row.members_with_roles or []:
if m["user_id"] == user_id:
return m["role"]
return None
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_member_update_authz_matrix(
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
member_id = scratch.tag("member")
await _seed_target(prisma, world, shape, scratch.prefix, member_id)
caller = world.keys[actor]
resp = await proxy_client.post(
"/team/member_update",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"team_id": scratch.prefix, "user_id": member_id, "role": "admin"},
)
assert (
resp.status_code == expected_status
), f"{actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert _role_of(row, member_id) == "admin"
else:
assert _role_of(row, member_id) == "user", "denied but role changed"

139
tests/proxy_behavior/management/test_team_new.py Normal file
View File

@ -0,0 +1,139 @@
from typing import Any, Dict
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/new — actor x org-target matrix (org_target picks the request's
# organization_id: none / ORG_A / ORG_B). Pinned against the role gate, which
# 401s every denial: PROXY_ADMIN always passes; any other caller must name an
# organization_id AND be ORG_ADMIN of that org.
_SCENARIOS = [
("none/proxy_admin", Actor.PROXY_ADMIN, "none", 200),
("none/org_admin", Actor.ORG_ADMIN, "none", 401),
("none/team_admin", Actor.TEAM_ADMIN, "none", 401),
("none/internal_user", Actor.INTERNAL_USER, "none", 401),
("none/owner", Actor.OWNER, "none", 401),
("none/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "none", 401),
("none/cross_org_user", Actor.CROSS_ORG_USER, "none", 401),
("none/service_account", Actor.SERVICE_ACCOUNT, "none", 401),
("none/org_b_admin", Actor.ORG_B_ADMIN, "none", 401),
("org_a/proxy_admin", Actor.PROXY_ADMIN, "org_a", 200),
("org_a/org_admin", Actor.ORG_ADMIN, "org_a", 200),
("org_a/team_admin", Actor.TEAM_ADMIN, "org_a", 401),
("org_a/internal_user", Actor.INTERNAL_USER, "org_a", 401),
("org_a/owner", Actor.OWNER, "org_a", 401),
("org_a/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "org_a", 401),
("org_a/cross_org_user", Actor.CROSS_ORG_USER, "org_a", 401),
("org_a/service_account", Actor.SERVICE_ACCOUNT, "org_a", 401),
("org_a/org_b_admin", Actor.ORG_B_ADMIN, "org_a", 401),
("org_b/proxy_admin", Actor.PROXY_ADMIN, "org_b", 200),
("org_b/org_admin", Actor.ORG_ADMIN, "org_b", 401),
("org_b/team_admin", Actor.TEAM_ADMIN, "org_b", 401),
("org_b/internal_user", Actor.INTERNAL_USER, "org_b", 401),
("org_b/owner", Actor.OWNER, "org_b", 401),
("org_b/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "org_b", 401),
("org_b/cross_org_user", Actor.CROSS_ORG_USER, "org_b", 401),
("org_b/service_account", Actor.SERVICE_ACCOUNT, "org_b", 401),
("org_b/org_b_admin", Actor.ORG_B_ADMIN, "org_b", 200),
]
@pytest.mark.parametrize(
"actor,org_target,expected_status",
[(a, o, s) for (_id, a, o, s) in _SCENARIOS],
ids=[s[0] for s in _SCENARIOS],
)
async def test_team_new_authz_matrix(
actor: Actor,
org_target: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
caller = world.keys[actor]
org_id = {
"none": None,
"org_a": world.org_a_id,
"org_b": world.org_b_id,
}[org_target]
body: Dict[str, Any] = {"team_id": scratch.prefix, "team_alias": scratch.prefix}
if org_id is not None:
body["organization_id"] = org_id
resp = await proxy_client.post(
"/team/new",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json=body,
)
assert (
resp.status_code == expected_status
), f"{actor.value} org={org_target}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
if expected_status == 200:
assert row is not None
assert row.organization_id == org_id
else:
assert row is None, f"{actor.value}: denied but team row leaked"
async def test_team_new_rejects_negative_budget(proxy_client, prisma, scratch, world):
"""Input-validation pin: max_budget < 0 is a 400, no row created."""
resp = await proxy_client.post(
"/team/new",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"team_id": scratch.prefix, "max_budget": -1},
)
assert resp.status_code == 400, resp.text
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is None
async def test_team_new_rejects_duplicate_team_id(proxy_client, prisma, scratch, world):
"""Input-validation pin: a colliding team_id is a 400 on the second call."""
seeder = world.keys[Actor.PROXY_ADMIN].cleartext
first = await proxy_client.post(
"/team/new",
headers={"Authorization": f"Bearer {seeder}"},
json={"team_id": scratch.prefix, "team_alias": scratch.prefix},
)
assert first.status_code == 200, first.text
second = await proxy_client.post(
"/team/new",
headers={"Authorization": f"Bearer {seeder}"},
json={"team_id": scratch.prefix, "team_alias": scratch.prefix},
)
assert second.status_code == 400, second.text
async def test_team_new_unknown_organization_is_500(
proxy_client, prisma, scratch, world
):
"""SURFACED, NOT ENDORSED: a /team/new with an organization_id that does
not exist currently fails 500 (the role-resolution layer raises before
the handler's own 400 'Organization not found' check is reached)."""
resp = await proxy_client.post(
"/team/new",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={
"team_id": scratch.prefix,
"organization_id": scratch.tag("no-such-org"),
},
)
assert resp.status_code == 500, resp.text
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is None

176
tests/proxy_behavior/management/test_team_update.py Normal file
View File

@ -0,0 +1,176 @@
import pytest
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/update — actor x team-shape matrix (shapes built by _seed_target).
# Each request carries the team's own organization_id so a non-proxy-admin can
# reach the org-scoped branch of the route-permission gate (401 on denial),
# which fronts the handler's _verify_team_access. Only PROXY_ADMIN and an
# ORG_ADMIN of the team's org pass: an internal_user team admin is filtered by
# the route gate before _verify_team_access's team-admin branch is reached.
MARKER_ALIAS = "behavior-pin-update-marker-alias"
_MATRIX = [
("alpha/proxy_admin", Actor.PROXY_ADMIN, "alpha", 200),
("alpha/org_admin", Actor.ORG_ADMIN, "alpha", 200),
("alpha/team_admin", Actor.TEAM_ADMIN, "alpha", 401),
("alpha/internal_user", Actor.INTERNAL_USER, "alpha", 401),
("alpha/owner", Actor.OWNER, "alpha", 401),
("alpha/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "alpha", 401),
("alpha/cross_org_user", Actor.CROSS_ORG_USER, "alpha", 401),
("alpha/service_account", Actor.SERVICE_ACCOUNT, "alpha", 401),
("alpha/org_b_admin", Actor.ORG_B_ADMIN, "alpha", 401),
("beta/proxy_admin", Actor.PROXY_ADMIN, "beta", 200),
("beta/org_admin", Actor.ORG_ADMIN, "beta", 401),
("beta/team_admin", Actor.TEAM_ADMIN, "beta", 401),
("beta/internal_user", Actor.INTERNAL_USER, "beta", 401),
("beta/owner", Actor.OWNER, "beta", 401),
("beta/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "beta", 401),
("beta/cross_org_user", Actor.CROSS_ORG_USER, "beta", 401),
("beta/service_account", Actor.SERVICE_ACCOUNT, "beta", 401),
("beta/org_b_admin", Actor.ORG_B_ADMIN, "beta", 200),
]
async def _seed_target(prisma, world, shape: str, team_id: str) -> str:
"""Raw-seed the scratch target team; returns its organization_id."""
if shape == "alpha":
await create_scratch_team(
prisma,
team_id,
organization_id=world.org_a_id,
admin_user_ids=[world.keys[Actor.TEAM_ADMIN].user_id],
member_user_ids=[
world.keys[Actor.INTERNAL_USER].user_id,
world.keys[Actor.OWNER].user_id,
world.keys[Actor.UNRELATED_SAME_ORG].user_id,
world.keys[Actor.SERVICE_ACCOUNT].user_id,
],
)
return world.org_a_id
if shape == "beta":
await create_scratch_team(
prisma,
team_id,
organization_id=world.org_b_id,
member_user_ids=[world.keys[Actor.CROSS_ORG_USER].user_id],
)
return world.org_b_id
pytest.fail(f"unknown shape={shape}") # pragma: no cover
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_update_authz_matrix(
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
org_id = await _seed_target(prisma, world, shape, scratch.prefix)
caller = world.keys[actor]
resp = await proxy_client.post(
"/team/update",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={
"team_id": scratch.prefix,
"team_alias": MARKER_ALIAS,
"organization_id": org_id,
},
)
assert (
resp.status_code == expected_status
), f"{actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert row.team_alias == MARKER_ALIAS
else:
assert row.team_alias != MARKER_ALIAS, "denied but team mutated"
async def test_team_update_requires_proxy_admin_without_org_context(
proxy_client, prisma, scratch, world
):
"""With no organization_id in the body the route gate has no org context
and falls back to proxy-admin-only: an org admin of the team's own org
is 401, PROXY_ADMIN is 200."""
await _seed_target(prisma, world, "alpha", scratch.prefix)
denied = await proxy_client.post(
"/team/update",
headers={"Authorization": f"Bearer {world.keys[Actor.ORG_ADMIN].cleartext}"},
json={"team_id": scratch.prefix, "team_alias": MARKER_ALIAS},
)
assert denied.status_code == 401, denied.text
allowed = await proxy_client.post(
"/team/update",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"team_id": scratch.prefix, "team_alias": MARKER_ALIAS},
)
assert allowed.status_code == 200, allowed.text
# Relocation gate — moving a team to a different org. The scratch team starts
# in ORG_A; each scenario relocates it to ORG_B. PROXY_ADMIN bypasses;
# ORG_B_ADMIN clears the route gate (dest-org admin) but fails
# _verify_team_access on the source team (403); the rest fail the route gate
# (401). The relocation-allowed branch needs a caller who is org admin of both
# orgs — no seeded actor is, so it is left to a later slice.
_RELOCATION = [
("proxy_admin", Actor.PROXY_ADMIN, 200),
("org_b_admin", Actor.ORG_B_ADMIN, 403),
("org_admin", Actor.ORG_ADMIN, 401),
("team_admin", Actor.TEAM_ADMIN, 401),
("internal_user", Actor.INTERNAL_USER, 401),
]
@pytest.mark.parametrize(
"actor,expected_status",
[(a, s) for (_id, a, s) in _RELOCATION],
ids=[s[0] for s in _RELOCATION],
)
async def test_team_update_org_relocation_gate(
actor: Actor,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
await _seed_target(prisma, world, "alpha", scratch.prefix)
caller = world.keys[actor]
resp = await proxy_client.post(
"/team/update",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"team_id": scratch.prefix, "organization_id": world.org_b_id},
)
assert (
resp.status_code == expected_status
), f"{actor.value}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert row.organization_id == world.org_b_id
else:
assert row.organization_id == world.org_a_id, "denied but team relocated"