merge: take main's e2e team allow-list settle helper

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

.circleci/config.yml

@@ -3009,7 +3009,7 @@ workflows:
           name: integration-<< matrix.suite >>
           matrix:
             parameters:
-              suite: [management, accounting, database, providers, extensions, browser]
+              suite: [management, accounting, database, providers, extensions, sdk, browser]
           filters:
             branches:
               only:

tests/e2e/access_control/access_control_client.py

@@ -122,16 +122,19 @@ class AccessControlClient:
         )
         return unwrap(result) if is_ok(result) else None
 
+    def team_models(self, team_id: str) -> list[str] | None:
+        result = self.proxy.transport.get(
+            "/team/info",
+            headers=self.proxy.transport.master,
+            params=TeamInfoParams(team_id=team_id),
+            response_type=TeamInfoResponse,
+        )
+        return unwrap(result).team_info.models if is_ok(result) else None
+
     def _await_team(self, team_id: str) -> None:
         deadline = time.monotonic() + self.proxy.poll_timeout
         while time.monotonic() < deadline:
-            result = self.proxy.transport.get(
-                "/team/info",
-                headers=self.proxy.transport.master,
-                params=TeamInfoParams(team_id=team_id),
-                response_type=TeamInfoResponse,
-            )
-            if is_ok(result):
+            if self.team_models(team_id) is not None:
                 return
             time.sleep(self.proxy.poll_interval)
         raise AssertionError(f"/team/info never resolved team {team_id!r} created by /team/new")

tests/e2e/access_control/test_model_access_group_e2e.py

@@ -23,7 +23,7 @@ from access_control_client import (
     MODEL_ACCESS_DENIED_MARKER,
     TEAM_MODEL_ACCESS_DENIED_MARKER,
 )
-from e2e_config import settle_propagation, unique_marker
+from e2e_config import unique_marker
 from lifecycle import ResourceManager
 from models import (
     ChatResponse,
@@ -31,7 +31,6 @@ from models import (
     LiteLLMParamsBody,
     ModelInfoBody,
     ModelNewBody,
-    TeamInfoResponse,
 )
 
 pytestmark = pytest.mark.e2e
@@ -112,6 +111,20 @@ def _await_group_members(client: AccessControlClient, access_group: str, expecte
     )
 
 
+def _await_team_allowlist(client: AccessControlClient, team_id: str, access_group: str) -> None:
+    deadline = time.monotonic() + client.proxy.poll_timeout
+    listed: list[str] | None = None
+    while time.monotonic() < deadline:
+        listed = client.team_models(team_id)
+        if listed == [access_group]:
+            return
+        time.sleep(client.proxy.poll_interval)
+    pytest.fail(
+        f"/team/info never settled the team's allow-list to [{access_group!r}] after the team-scoped "
+        f"deployment was registered; last read {listed}"
+    )
+
+
 @pytest.fixture(scope="module")
 def grouped(client: AccessControlClient) -> Iterator[GroupedDeployments]:
     marker: Final = unique_marker()
@@ -155,15 +168,9 @@ def team_grant(client: AccessControlClient) -> Iterator[TeamGrant]:
         ),
         listed_for=key,
     )
+    client.set_team_models(team_id, team_alias, [access_group])
     try:
-        client.set_team_models(team_id, team_alias, [access_group])
-        written_at: Final = time.monotonic()
-        _ = client.proxy.read_body_back_everywhere(
-            f"/team/info?team_id={team_id}",
-            TeamInfoResponse,
-            settled=lambda response: response.team_id == team_id and response.team_info.models == [access_group],
-        )
-        settle_propagation(written_at)
+        _await_team_allowlist(client, team_id, access_group)
         yield TeamGrant(access_group=access_group, team_id=team_id, key=key)
     finally:
         client.proxy.delete_model(model_id)

tests/e2e/batches/batch_cleanup.py

@@ -5,7 +5,7 @@ from time import monotonic, sleep
 from typing import Final, Protocol
 
 from batch_client import BatchObject, FileDeleteResponse
-from capabilities import is_managed_id
+from capabilities import is_cloud_storage_id, is_managed_id
 from e2e_http import NetworkError, RateLimitedError, Result, Success, UnknownApiError
 from pydantic import BaseModel
 
@@ -19,6 +19,8 @@ BATCH_CANCEL_POLL_SECONDS: Final = 10.0
 class BatchCleanupClient(Protocol):
     def delete_file(self, file_id: str, *, key: str, provider: str | None = None) -> Result[FileDeleteResponse]: ...
 
+    def delete_file_as_admin(self, file_id: str, *, provider: str | None = None) -> Result[FileDeleteResponse]: ...
+
     def retrieve_batch(self, batch_id: str, *, key: str, provider: str | None = None) -> Result[BatchObject]: ...
 
     def cancel_batch(self, batch_id: str, *, key: str, provider: str | None = None) -> Result[BatchObject]: ...
@@ -49,7 +51,12 @@ def _require_cleanup_success[R: BaseModel](result: Result[R], operation: str) ->
 
 
 def cleanup_file(client: BatchCleanupClient, file_id: str, *, key: str, provider: str | None = None) -> None:
-    result: Final = cleanup_result(lambda: client.delete_file(file_id, key=key, provider=provider))
+    delete: Final[Callable[[], Result[FileDeleteResponse]]] = (
+        (lambda: client.delete_file_as_admin(file_id, provider=provider))
+        if is_cloud_storage_id(file_id)
+        else (lambda: client.delete_file(file_id, key=key, provider=provider))
+    )
+    result: Final = cleanup_result(delete)
     if isinstance(result, UnknownApiError) and result.status_code == 404:
         return
     deleted: Final = _require_cleanup_success(result, f"Delete file {file_id}")

tests/e2e/batches/batch_client.py

@@ -233,6 +233,14 @@ class BatchClient:
             response_type=FileDeleteResponse,
         )
 
+    def delete_file_as_admin(self, file_id: str, *, provider: str | None = None) -> Result[FileDeleteResponse]:
+        return self.proxy.transport.delete(
+            f"{_files_path(provider)}/{file_id}",
+            headers=self.proxy.transport.master,
+            json=NoBody(),
+            response_type=FileDeleteResponse,
+        )
+
 
 def _files_path(provider: str | None) -> str:
     return f"/{provider}/v1/files" if provider else "/v1/files"

tests/e2e/batches/capabilities.py

@@ -222,6 +222,13 @@ def is_managed_id(id_str: str) -> bool:
     return _b64_decode(id_str).startswith("litellm_proxy")
 
 
+CLOUD_STORAGE_SCHEMES: Final = ("s3://", "gs://")
+
+
+def is_cloud_storage_id(id_str: str) -> bool:
+    return id_str.startswith(CLOUD_STORAGE_SCHEMES)
+
+
 def is_model_encoded_id(id_str: str) -> bool:
     for prefix in ("file-", "batch_"):
         if id_str.startswith(prefix):

tests/e2e/batches/test_batch_cleanup.py

@@ -45,6 +45,10 @@ class CleanupClient:
         self.calls(f"delete {provider} {file_id}")
         return self.file_response()
 
+    def delete_file_as_admin(self, file_id: str, *, provider: str | None = None) -> Result[FileDeleteResponse]:
+        self.calls(f"admin delete {provider} {file_id}")
+        return self.file_response()
+
     def retrieve_batch(self, batch_id: str, *, key: str, provider: str | None = None) -> Result[BatchObject]:
         self.calls(f"retrieve {provider} {batch_id}")
         return self.batch_response()

tests/integration/README.md

@@ -2,7 +2,7 @@
 
 These tests exercise a running gateway, PostgreSQL and Redis with an owned local upstream. CircleCI owns this suite. Tests are grouped by behavior, with no automatic test retries or fallback to paid provider calls
 
-Use `tests/integration/run.py management`, `accounting`, `database`, `providers` or `extensions` to run a selected group. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate
+Use `tests/integration/run.py management`, `accounting`, `database`, `providers`, `extensions` or `sdk` to run a selected group. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate
 
 Management also requires `INTEGRATION_PEER_URL`, `REDIS_HOST` and `REDIS_PORT`. CircleCI starts two directly addressed proxy processes sharing only that job's stores. The test-only CLI wrapper supplies enterprise route entitlement, following the existing behavior suite's convention. It does not qualify license validation; run it with one worker and no reload
 
@@ -26,6 +26,8 @@ Provider contracts exercise actual TCP requests with synthetic credentials and l
 
 Streaming checks send real HTTP transfer chunks, including one-byte partitions, fragmented tools, incomplete transfers and a cancellation barrier. They assert meaningful text, tool arguments, final usage and persisted cost. The Redis recovery case owns a separate database and Redis process, uses the supported one-second circuit-breaker recovery setting, waits for the real subscriber and verifies response data in Redis after restart. CircleCI reuses its existing Redis image for that extra process; it never pulls an image during tests
 
+The sdk shard exercises the SDK's own HTTP clients against local protocol peers with no gateway in the path, so a case here fails only when the client library or its wire behavior changes. The HTTP/2 case runs a hypercorn TLS peer offering h2 and http/1.1 over ALPN, drives the sync and async httpx handlers at it with `LITELLM_HTTP2` off and on, and asserts the version both the client and the peer observed on the wire. Put a test here only when it needs no proxy, database or Redis; a case that reaches the gateway belongs in one of the other shards
+
 The extensions shard reuses the existing MCP arithmetic functions with a real SDK server, and uses the built-in generic callback and guardrail transports. It checks actual tool calls after saved edits, discovery preservation, malformed/error responses, callback correlation and credential exclusion, guardrail rewriting and denial, retained OpenAI consumers, persisted toolsets and A2A wire versions
 
 Browser contracts live in `tests/e2e/ui/tests/integrationCritical` and run only through `tests/e2e/ui/integration.config.ts`. The CircleCI browser shard builds the checked-out dashboard, starts the owned proxy with that build, and verifies one exact browser result without retries or skips. The default Playwright selection excludes this directory. The focused project flow asserts the submitted create and clear values, fresh SQL state and actual blocked/restored serving while preserving model restrictions

tests/integration/_support/manifest.py

@@ -19,6 +19,7 @@ OWNED_DIRECTORIES: Final = frozenset(
         "mcp",
         "observability",
         "compatibility",
+        "sdk",
     }
 )
 

tests/integration/contracts.json

@@ -21,6 +21,9 @@
       "mcp",
       "observability",
       "compatibility"
+    ],
+    "sdk": [
+      "sdk"
     ]
   },
   "tests": {
@@ -204,6 +207,12 @@
     ],
     "tests/integration/management/test_project_lifecycle.py::test_project_delete_with_attached_key_refuses_and_preserves_state": [
       "mgmt.project.delete.attached_key_refusal_preserves_state"
+    ],
+    "tests/integration/sdk/test_http2_wire.py::test_async_handler_negotiates_http2_only_when_enabled": [
+      "other.sdk_wire.http2.async_handler_negotiates_h2_only_when_enabled"
+    ],
+    "tests/integration/sdk/test_http2_wire.py::test_sync_handler_negotiates_http2_only_when_enabled": [
+      "other.sdk_wire.http2.sync_handler_negotiates_h2_only_when_enabled"
     ]
   },
   "browser": {

tests/integration/sdk/test_http2_wire.py

@@ -1,21 +1,14 @@
-"""Outbound HTTP/2 negotiation for LiteLLM-built httpx clients.
-
-Spins up a local hypercorn TLS server that offers h2 and http/1.1 over ALPN and
-drives the real AsyncHTTPHandler / HTTPHandler at it, so the negotiated protocol
-on the wire is the assertion. No running proxy or provider credentials needed,
-which is why these tests carry no `e2e` marker (same shape as the markerless
-harness checks under tests/e2e/load/).
-"""
-
 from __future__ import annotations
 
 import asyncio
 import datetime
 import ipaddress
+import json
 import socket
 import threading
 import time
 from collections.abc import Iterator
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Final, cast
 
@@ -28,16 +21,17 @@ from hypercorn.asyncio import (
     serve,  # pyright: ignore[reportUnknownVariableType]  # hypercorn's serve signature passes through untyped worker hooks
 )
 from hypercorn.config import Config
-from hypercorn.typing import (
-    ASGIReceiveCallable,
-    ASGISendCallable,
-    HTTPResponseBodyEvent,
-    HTTPResponseStartEvent,
-    Scope,
-)
+from hypercorn.typing import ASGIReceiveCallable, ASGISendCallable, HTTPResponseBodyEvent, HTTPResponseStartEvent, Scope
 
-import litellm
-from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
+STREAM_CHUNKS: Final = 3
+
+
+@dataclass(frozen=True, slots=True)
+class Observed:
+    post_version: str
+    post_peer_version: str
+    stream_version: str
+    stream_body: bytes
 
 
 def _write_self_signed_cert(cert_dir: Path) -> tuple[Path, Path]:
@@ -71,7 +65,7 @@ def _write_self_signed_cert(cert_dir: Path) -> tuple[Path, Path]:
     return cert_file, key_file
 
 
-async def _asgi_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
+async def _peer(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
     if scope["type"] != "http":
         return
     while True:
@@ -80,16 +74,17 @@ async def _asgi_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCa
             return
         if message["type"] == "http.request" and not message["more_body"]:
             break
+    version: Final = scope["http_version"]
     if scope["path"] == "/stream":
         await send(
             HTTPResponseStartEvent(
                 type="http.response.start", status=200, headers=[(b"content-type", b"text/event-stream")]
             )
         )
-        for index in range(3):
+        for index in range(STREAM_CHUNKS):
             await send(
                 HTTPResponseBodyEvent(
-                    type="http.response.body", body=f"data: chunk-{index}\n\n".encode(), more_body=True
+                    type="http.response.body", body=f"data: {version}-{index}\n\n".encode(), more_body=True
                 )
             )
         await send(HTTPResponseBodyEvent(type="http.response.body", body=b"", more_body=False))
@@ -97,18 +92,19 @@ async def _asgi_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCa
     await send(
         HTTPResponseStartEvent(type="http.response.start", status=200, headers=[(b"content-type", b"application/json")])
     )
-    await send(HTTPResponseBodyEvent(type="http.response.body", body=b'{"ok": true}', more_body=False))
+    await send(
+        HTTPResponseBodyEvent(
+            type="http.response.body", body=json.dumps({"http_version": version}).encode(), more_body=False
+        )
+    )
 
 
 @pytest.fixture(scope="module")
-def http2_tls_server(tmp_path_factory: pytest.TempPathFactory) -> Iterator[str]:
-    cert_dir: Final = tmp_path_factory.mktemp("h2certs")
-    cert_file, key_file = _write_self_signed_cert(cert_dir)
-
+def http2_tls_peer(tmp_path_factory: pytest.TempPathFactory) -> Iterator[str]:
+    cert_file, key_file = _write_self_signed_cert(tmp_path_factory.mktemp("h2certs"))
     with socket.socket() as sock:
         sock.bind(("127.0.0.1", 0))
         port: Final = cast(int, sock.getsockname()[1])
-
     shutdown: Final = threading.Event()
 
     def _serve() -> None:
@@ -118,12 +114,11 @@ def http2_tls_server(tmp_path_factory: pytest.TempPathFactory) -> Iterator[str]:
         config.certfile = str(cert_file)
         config.keyfile = str(key_file)
         config.alpn_protocols = ["h2", "http/1.1"]
-        loop.run_until_complete(serve(_asgi_app, config, shutdown_trigger=lambda: asyncio.to_thread(shutdown.wait)))
+        loop.run_until_complete(serve(_peer, config, shutdown_trigger=lambda: asyncio.to_thread(shutdown.wait)))
         loop.close()
 
     thread: Final = threading.Thread(target=_serve, daemon=True)
     thread.start()
-
     for _ in range(100):
         try:
             with socket.create_connection(("127.0.0.1", port), timeout=0.2):
@@ -131,78 +126,75 @@ def http2_tls_server(tmp_path_factory: pytest.TempPathFactory) -> Iterator[str]:
         except OSError:
             time.sleep(0.05)
     else:
-        pytest.fail("hypercorn test server did not start")
-
+        pytest.fail("hypercorn peer did not start")
     yield f"https://127.0.0.1:{port}"
-
     shutdown.set()
     thread.join(timeout=10)
 
 
-def _async_exchange(base_url: str) -> tuple[str, str, bytes]:
-    async def _run() -> tuple[str, str, bytes]:
+def _async_exchange(base_url: str) -> Observed:
+    from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+
+    async def _run() -> Observed:
         handler: Final = AsyncHTTPHandler(ssl_verify=False)
         try:
             response: Final = await handler.client.post(f"{base_url}/echo", json={"ping": "pong"})
-            post_version: Final = response.http_version
             async with handler.client.stream("POST", f"{base_url}/stream", json={}) as stream_response:
-                stream_version: Final = stream_response.http_version
-                body: Final = b"".join([chunk async for chunk in stream_response.aiter_bytes()])
-            return post_version, stream_version, body
+                return Observed(
+                    post_version=response.http_version,
+                    post_peer_version=response.json()["http_version"],
+                    stream_version=stream_response.http_version,
+                    stream_body=b"".join([chunk async for chunk in stream_response.aiter_bytes()]),
+                )
         finally:
             await handler.close()
 
     return asyncio.run(_run())
 
 
-def _sync_exchange(base_url: str) -> tuple[str, str, bytes]:
+def _sync_exchange(base_url: str) -> Observed:
+    from litellm.llms.custom_httpx.http_handler import HTTPHandler
+
     handler: Final = HTTPHandler(ssl_verify=False)
     try:
         response: Final = handler.client.post(f"{base_url}/echo", json={"ping": "pong"})
-        post_version: Final = response.http_version
         with handler.client.stream("POST", f"{base_url}/stream", json={}) as stream_response:
-            stream_version: Final = stream_response.http_version
-            body: Final = b"".join(stream_response.iter_bytes())
-        return post_version, stream_version, body
+            return Observed(
+                post_version=response.http_version,
+                post_peer_version=response.json()["http_version"],
+                stream_version=stream_response.http_version,
+                stream_body=b"".join(stream_response.iter_bytes()),
+            )
     finally:
         handler.close()
 
 
-class TestOutboundHttp2:
-    @pytest.mark.parametrize("use_http2, expected_version", [(True, "HTTP/2"), (False, "HTTP/1.1")])
-    def test_async_handler_negotiates_http2_only_when_enabled(
-        self,
-        monkeypatch: pytest.MonkeyPatch,
-        http2_tls_server: str,
-        use_http2: bool,
-        expected_version: str,
-    ) -> None:
-        monkeypatch.setattr(litellm, "http2", use_http2)
+def _set_http2(monkeypatch: pytest.MonkeyPatch, enabled: bool) -> None:
+    if enabled:
+        monkeypatch.setenv("LITELLM_HTTP2", "True")
+    else:
         monkeypatch.delenv("LITELLM_HTTP2", raising=False)
-        monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
-        monkeypatch.setattr(litellm, "force_ipv4", False)
 
-        post_version, stream_version, body = _async_exchange(http2_tls_server)
 
-        assert post_version == expected_version
-        assert stream_version == expected_version
-        assert b"data: chunk-0" in body
+def _assert_negotiated(observed: Observed, enabled: bool) -> None:
+    client_version, peer_version = ("HTTP/2", "2") if enabled else ("HTTP/1.1", "1.1")
+    assert observed.post_version == client_version
+    assert observed.post_peer_version == peer_version
+    assert observed.stream_version == client_version
+    expected_stream: Final = b"".join(f"data: {peer_version}-{index}\n\n".encode() for index in range(STREAM_CHUNKS))
+    assert observed.stream_body == expected_stream
 
-    @pytest.mark.parametrize("use_http2, expected_version", [(True, "HTTP/2"), (False, "HTTP/1.1")])
-    def test_sync_handler_negotiates_http2_only_when_enabled(
-        self,
-        monkeypatch: pytest.MonkeyPatch,
-        http2_tls_server: str,
-        use_http2: bool,
-        expected_version: str,
-    ) -> None:
-        monkeypatch.setattr(litellm, "http2", use_http2)
-        monkeypatch.delenv("LITELLM_HTTP2", raising=False)
-        monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
-        monkeypatch.setattr(litellm, "force_ipv4", False)
 
-        post_version, stream_version, body = _sync_exchange(http2_tls_server)
+@pytest.mark.covers("other.sdk_wire.http2.async_handler_negotiates_h2_only_when_enabled")
+def test_async_handler_negotiates_http2_only_when_enabled(monkeypatch: pytest.MonkeyPatch, http2_tls_peer: str) -> None:
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    for enabled in (False, True):
+        _set_http2(monkeypatch, enabled)
+        _assert_negotiated(_async_exchange(http2_tls_peer), enabled)
 
-        assert post_version == expected_version
-        assert stream_version == expected_version
-        assert b"data: chunk-0" in body
+
+@pytest.mark.covers("other.sdk_wire.http2.sync_handler_negotiates_h2_only_when_enabled")
+def test_sync_handler_negotiates_http2_only_when_enabled(monkeypatch: pytest.MonkeyPatch, http2_tls_peer: str) -> None:
+    for enabled in (False, True):
+        _set_http2(monkeypatch, enabled)
+        _assert_negotiated(_sync_exchange(http2_tls_peer), enabled)