mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-13 03:59:12 +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>
78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
import pytest
|
|
|
|
from tests.integration._support.actors import Actor
|
|
from tests.integration._support.actors import create_scratch_team
|
|
|
|
pytestmark = pytest.mark.asyncio(loop_scope="session")
|
|
|
|
_MARKER_MODEL = "behavior-pin-team-model-marker"
|
|
_ROUTE_URL = {"add": "/team/model/add", "delete": "/team/model/delete"}
|
|
|
|
|
|
# POST /team/model/add + /team/model/delete. The handler gate is PROXY_ADMIN
|
|
# or team admin or org admin, but the management-route gate fronts it — these
|
|
# are neither internal_user nor org-admin nor info routes, so every
|
|
# non-proxy-admin is 401 before the handler runs. Only PROXY_ADMIN reaches the
|
|
# handler, making the team-admin / org-admin handler branches unreachable here.
|
|
_MATRIX = [
|
|
("proxy_admin", Actor.PROXY_ADMIN, 200),
|
|
("org_admin", Actor.ORG_ADMIN, 401),
|
|
("team_admin", Actor.TEAM_ADMIN, 401),
|
|
("internal_user", Actor.INTERNAL_USER, 401),
|
|
("owner", Actor.OWNER, 401),
|
|
("unrelated_same_org", Actor.UNRELATED_SAME_ORG, 401),
|
|
("cross_org_user", Actor.CROSS_ORG_USER, 401),
|
|
("service_account", Actor.SERVICE_ACCOUNT, 401),
|
|
("org_b_admin", Actor.ORG_B_ADMIN, 401),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("route", ["add", "delete"])
|
|
@pytest.mark.parametrize(
|
|
"actor,expected_status",
|
|
[(a, s) for (_id, a, s) in _MATRIX],
|
|
ids=[s[0] for s in _MATRIX],
|
|
)
|
|
async def test_team_model_authz_matrix(
|
|
route: str,
|
|
actor: Actor,
|
|
expected_status: int,
|
|
gateway_client,
|
|
db,
|
|
resources,
|
|
actors,
|
|
):
|
|
initial = [] if route == "add" else [_MARKER_MODEL]
|
|
await create_scratch_team(
|
|
db, resources.prefix, organization_id=actors.org_a_id, models=initial
|
|
)
|
|
caller = actors.keys[actor]
|
|
|
|
resp = await gateway_client.post(
|
|
_ROUTE_URL[route],
|
|
headers={"Authorization": f"Bearer {caller.cleartext}"},
|
|
json={"team_id": resources.prefix, "models": [_MARKER_MODEL]},
|
|
)
|
|
assert (
|
|
resp.status_code == expected_status
|
|
), f"{route} {actor.value}: {resp.status_code} {resp.text}"
|
|
|
|
row = await db.db.litellm_teamtable.find_unique(
|
|
where={"team_id": resources.prefix}
|
|
)
|
|
assert row is not None
|
|
if expected_status == 200:
|
|
assert (_MARKER_MODEL in row.models) is (route == "add")
|
|
else:
|
|
assert list(row.models) == initial, "denied but models mutated"
|
|
|
|
|
|
@pytest.mark.parametrize("route", ["add", "delete"])
|
|
async def test_team_model_missing_team_is_404(route: str, gateway_client, actors):
|
|
"""A team_id absent from the DB is 404 — the existence check precedes authz."""
|
|
resp = await gateway_client.post(
|
|
_ROUTE_URL[route],
|
|
headers={"Authorization": f"Bearer {actors.keys[Actor.PROXY_ADMIN].cleartext}"},
|
|
json={"team_id": "behavior-pin-no-such-team", "models": [_MARKER_MODEL]},
|
|
)
|
|
assert resp.status_code == 404, resp.text
|