mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-12 03:48:56 +00:00
780c504e00
* test: migrate tests/proxy_behavior into integration and unit tiers (partial) * test(integration): run settings-dependent management tests on an owned proxy * test(integration): assert stored values in the rewritten management rows * test(integration): pin sibling team allocations against the org guaranteed throughput cap * test(integration): keep Prisma clients out of the shared integration conftest import path --------- Co-authored-by: yuneng <yuneng@berri.ai>
86 lines
3.8 KiB
Python
86 lines
3.8 KiB
Python
import pytest
|
|
|
|
from tests.integration._support.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, gateway_client, actors
|
|
):
|
|
caller = actors.keys[actor]
|
|
target_team_id = {
|
|
"alpha": actors.team_alpha_id,
|
|
"gamma": actors.team_gamma_id,
|
|
"beta": actors.team_beta_id,
|
|
}[target]
|
|
|
|
resp = await gateway_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
|
|
|
|
|
|
# Phase 4 F6 — explicit pin on the `team_access_denied` 403 message string.
|
|
# alpha/org_b_admin already covers the branch in the matrix; this guard
|
|
# turns a silent rename of the exception detail into a CI red, which is the
|
|
# behavior tripwire that the matrix's status-only assertion cannot catch.
|
|
async def test_team_info_org_admin_cross_org_rejection_detail(gateway_client, actors):
|
|
resp = await gateway_client.get(
|
|
f"/team/info?team_id={actors.team_alpha_id}",
|
|
headers={"Authorization": f"Bearer {actors.keys[Actor.ORG_B_ADMIN].cleartext}"},
|
|
)
|
|
assert resp.status_code == 403, resp.text
|
|
# validate_membership() at GET /team/info raises with this exact phrase.
|
|
# Pinning it lock-step locks down the visible auth message — a rename
|
|
# would flip CI red even when the status stays 403.
|
|
assert "not authorized to access this team" in resp.text, resp.text
|