mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-14 04:09:09 +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>
105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
from typing import FrozenSet, Optional
|
|
|
|
import pytest
|
|
|
|
from tests.integration._support.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, actors) -> set:
|
|
known = {
|
|
actors.team_alpha_id: "alpha",
|
|
actors.team_beta_id: "beta",
|
|
actors.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],
|
|
gateway_client,
|
|
actors,
|
|
):
|
|
caller = actors.keys[actor]
|
|
resp = await gateway_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(), actors)
|
|
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], gateway_client, actors
|
|
):
|
|
caller = actors.keys[actor]
|
|
resp = await gateway_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(), actors)
|
|
assert visible == set(expected_visible), (
|
|
f"{actor.value}: expected {sorted(expected_visible)}, " f"got {sorted(visible)}"
|
|
)
|