litellm/tests/integration/management/test_team_blocking_multi_instance.py
devin-ai-integration[bot] 1dbde0845e
test: migrate SDK and proxy-booting legacy CircleCI tests to their tiers (#45758)
* test: move guardrails, image, audio, batches, search legacy tests to unit

* test: restore the custom guardrail and remaining chunk 1 unit rows

* test: move the ocr matrix to e2e

* test: drop comments from moved tests and fix vertex ocr credentials

* test: move litellm utils and unified google legacy tests to their tiers

* test: move logging callback and pass-through unit legacy tests to their tiers

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: move store_model_in_db legacy tests and start the admin ui migration

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: finish the proxy job rows and drop their CircleCI jobs

Moves the remaining multi-instance, openai endpoint, otel, pass-through and
top-level proxy tests into tests/integration with scripted upstreams, rewrites
the weak replacements to assert the outbound wire and exact responses, and
removes the build_and_test job group this PR owns from CircleCI

* test: tighten the key generation and error log assertions, rename migrated unit files to their mapped names

* test: shard the pagerduty unit tests and record the rust harness file on the coverage allowlist

* test: drive the SSO sign-in tests through a scripted IdP and isolate the cost center stand-in

* test: type the e2e beta header, prompt cache and secret manager values

* test: assert the missing model error message on the OpenAI routes

* test: drop the deleted otel base class import and restore mcp redaction and exact langfuse usage checks

* test: make the langfuse fixture directory a package for the layout check

* test: restore collected no-match cases, real langfuse and gcs payload checks, and gate the aws secret lane

* test: keep the ocr e2e pdf fixture next to the test

* test: put meta below covers on the streaming tool search e2e

* test: run the OCR matrix against a scripted upstream and fix two runner-only e2e failures

* test: count only OCR calls in the OCR matrix recorder

* test: pin OCR parse bodies and billed cost, group new e2e cases into classes

* ci: restore the python image anchor dropped in the main sync

---------

Co-authored-by: yuneng <yuneng@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-11 00:17:21 -07:00

40 lines
1.6 KiB
Python

from typing import Final
import httpx
from tests.integration._support.client import Gateway, eventually, object_value
_PROMPT: Final = {"messages": [{"role": "user", "content": "team blocking"}]}
def _chat(proxy: Gateway, model: str, key: str) -> httpx.Response:
return proxy.request("POST", "/v1/chat/completions", {"model": model, **_PROMPT}, key=key)
def _rejected(proxy: Gateway, model: str, key: str) -> httpx.Response:
return eventually(lambda: _chat(proxy, model, key), lambda response: response.status_code == 401, seconds=70)
def _blocked_flag(proxy: Gateway, team_id: str) -> object:
return object_value(proxy.get("/team/info", {"team_id": team_id})["team_info"])["blocked"]
def test_a_team_blocked_on_one_instance_is_rejected_on_both(gateway: Gateway, peer: Gateway) -> None:
with gateway.scenario() as scenario:
model: Final = scenario.model()
team: Final = scenario.team()
key: Final = scenario.key(team_id=team)
assert _blocked_flag(peer, team) is False
allowed: Final = _chat(gateway, model, key)
assert allowed.status_code == 200, allowed.text
peer.post("/team/update", {"team_id": team, "blocked": True})
assert _blocked_flag(gateway, team) is True
assert _blocked_flag(peer, team) is True
rejections: Final = tuple(_rejected(proxy, model, key) for proxy in (gateway, peer))
for rejection in rejections:
error: Final = object_value(rejection.json()["error"])
assert error["type"] == "auth_error", rejection.text
assert f"Team={team} is blocked" in str(error["message"]), rejection.text