test(integration): remove six setup causes of CircleCI integration reds (#46490)

Drain the worker's fake-upstream request log when a test's gateway fixture starts, so traffic left by earlier tests on the same xdist worker no longer shows up in a test's first observation read

Generate the fine-tuning rig's provider file and job ids per fixture build, so a rebuilt module fixture in the same worker no longer collides on the managed object unique constraint

Wait for every wedged team_id and team_alias invalidation publish to fail before taking the recovery marker, now that team updates publish both keys

Pin the cooled-down priority scheduler request to the connection that observed the cooldown, since a sibling worker reads cooldowns from Redis on an interval

Simulate upstream outages and the held budget peer on ports the test keeps owning, instead of releasing a port another worker can bind

Start a peer proxy for the accounting group, which has a test that needs one

.circleci/scripts/run_integration.sh

@@ -201,7 +201,7 @@ if [ "$isolated_workers" = false ]; then
   curl --noproxy '*' -sSf -X POST "$INTEGRATION_PROXY_URL/config/update" \
     -H "Authorization: Bearer $LITELLM_MASTER_KEY" -H 'Content-Type: application/json' \
     -d '{"router_settings": {"num_retries": 0}}' > "$results/seed-router-settings.json"
-  if [ "$suite" = management ] || [ "$suite" = mcp ]; then
+  if [ "$suite" = management ] || [ "$suite" = mcp ] || [ "$suite" = accounting ]; then
     export INTEGRATION_PEER_URL=http://127.0.0.1:4001
     start_proxy 4001 peer.log
     peer_pid="$launched_pid"

tests/integration/README.md

@@ -10,7 +10,7 @@ The `cost` group is driven by the `CostTrackingTestCase` literals under `cost_ca
 
 Use `tests/integration/run.py management`, `accounting`, `database`, `providers`, `extensions`, `mcp`, `sdk` or `cost` to run a selected group. The group to directory mapping is the `GROUPS` literal at the top of `run.py`; a new directory needs a `GROUPS` entry and an `OWNED_DIRECTORIES` entry in `_support/manifest.py`. Set `INTEGRATION_WORKERS` above 1 to run a group under pytest-xdist. Every group run by more than one worker gives each worker a private database, Redis, scripted upstream process, fake provider and proxy, with a peer and read-only database role where required. There is no opt-in list and no shared proxy, upstream, database or Redis in that mode: `run_integration.sh` does not start them and does not hand their addresses to the run, so a test can only reach its own worker's services. The pytest controller owns these services and closes them after worker failures. Their cases use pytest-xdist load groups inside the files selected by CircleCI timing splits. Module and class fixtures are grouped automatically, so authors do not need scheduling markers. Function fixtures remain independently schedulable. Tests that own an independent mutable sink for every case should use function fixtures, so a shared module fixture does not serialize them or share pending observations. Nested harness probes discard the parent run's shard, worker and result settings, then supply their own selections. MCP keeps its explicit load groups. Automatic group suffixes are removed from JUnit and execution records to preserve case identities. Owned proxy helpers wait for a healthy response from every configured, live worker before yielding. Serial runs keep the supplied deployment, including routing parity runs. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. When that deployment runs more than one proxy worker, set `INTEGRATION_PROXY_WORKERS` to the count so a test that writes a model and then calls it waits out the config reload interval, the only cross-worker convergence bound the wire exposes. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate
 
-Management and MCP also require `INTEGRATION_PEER_URL`, `REDIS_HOST` and `REDIS_PORT`. Under several workers each worker gets a second directly addressed proxy sharing only that worker's stores, and a single-worker run uses the two proxies `run_integration.sh` starts. 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
+Management, MCP and accounting also require `INTEGRATION_PEER_URL`, `REDIS_HOST` and `REDIS_PORT`. Under several workers each worker gets a second directly addressed proxy sharing only that worker's stores, and a single-worker run uses the two proxies `run_integration.sh` starts. 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
 
 The generated lifecycle models use 20 examples, eight steps, generation and shrinking, with isolated resources per example. HTTP operation caps include generation and shrinking and exempt cleanup. Local qualification defaults to seed 4106601 and canonical order; CircleCI derives exploration and ordering seeds from the checked-out revision and workflow ID. The ordering seed shuffles files, scoped fixture configurations and cases within each configuration. Ordering keeps shared module and class configurations together where their dependencies permit it. Dependent fixtures with their own module parameters can still rebuild an outer configuration, with every case retained
 

tests/integration/conftest.py

@@ -285,6 +285,7 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
 @pytest.fixture
 def gateway() -> Iterator[Gateway]:
     with gateway_from_environment() as value:
+        httpx.get(f"{value.upstream_url}/__observations", timeout=5, trust_env=False).raise_for_status()
         yield value
 
 

tests/integration/management/test_user_updates_wedged_coordination_redis.py

@@ -249,27 +249,26 @@ def test_user_budget_updates_return_promptly_while_coordination_redis_is_wedged(
                     assert timings["c1_survivor"] < _HANDLER_BUDGET_SECONDS, (
                         f"control update on the surviving worker took {timings['c1_survivor']:.3f}s"
                     )
-                    wedged_keys: Final = {users[i] for i in range(_BURST) if i % 5 == 0 and i != 0} | {
-                        f"team_id:{team_id}"
-                    }
+                    wedged_users: Final = {users[i] for i in range(_BURST) if i % 5 == 0 and i != 0}
+                    wedged_team_keys: Final = {f"team_id:{team_id}", f"team_alias:{identity}"}
+                    wedged_keys: Final = wedged_users | wedged_team_keys
+                    wedged_team_updates: Final = 1 + sum(1 for i in range(_BURST) if i % 5 == 3)
 
                     def proxy_log() -> str:
                         return "".join(
                             path.read_text() for path in results_dir.glob("owned-proxy-*.log") if path not in prior_logs
                         )
 
-                    team_wedged_key: Final = f"team_id:{team_id}"
                     eventually(
                         proxy_log,
                         lambda text: (
                             all(
-                                text.count(f"publish for {wedged_key} failed") >= _WEDGED_PUBLISHES_PER_BURST_USER
-                                for wedged_key in wedged_keys
-                                if wedged_key != team_wedged_key
+                                text.count(f"publish for {user} failed") >= _WEDGED_PUBLISHES_PER_BURST_USER
+                                for user in wedged_users
                             )
-                            and (
-                                f"publish for {team_wedged_key} failed" in text
-                                or f"internal usage cache entry {team_wedged_key}" in text
+                            and all(
+                                text.count(f"publish for {team_key} failed") >= wedged_team_updates
+                                for team_key in wedged_team_keys
                             )
                         ),
                         seconds=45,

tests/integration/providers/test_vertex_anthropic_inline_tools_beta_wire.py

@@ -512,15 +512,19 @@ def test_upstream_outage_mid_run_recovers_with_inline_tools_still_forwarded(gate
             seen: Final = tuple(sum(marker in sent.body for sent in delivered) for _, marker in results)
             assert seen == (1,) * len(clients), seen
 
-        with wire_server(_peer) as wire:
-            port: Final = int(wire.url.rsplit(":", 1)[1])
-            model: Final = _vertex_deployment(gateway, scenario, wire.url)
-            forwarded(wire, wave("before outage"))
+        down: Final = threading.Event()
+        wire: Final = stack.enter_context(
+            wire_server(lambda request: Reply(drop_connection=True) if down.is_set() else _peer(request))
+        )
+        model: Final = _vertex_deployment(gateway, scenario, wire.url)
+        forwarded(wire, wave("before outage"))
+        down.set()
         outage: Final = wave("during outage")
         assert all(status is not None and status >= 500 for status, _ in outage), outage
         assert gateway.request("GET", "/health/liveliness").status_code == 200
-        with wire_server(_peer, port=port) as revived:
-            forwarded(revived, wave("after outage"))
+        wire.drain()
+        down.clear()
+        forwarded(wire, wave("after outage"))
 
 
 @pytest.mark.timeout(_OWNED_PROXY_CELL_SECONDS)

tests/integration/routing/test_pdf_document_pre_call_checks_owned_proxy.py

@@ -1,5 +1,4 @@
 import os
-import socket
 import threading
 import uuid
 from collections.abc import Callable, Iterator, Mapping
@@ -50,19 +49,14 @@ _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
 class _Rig:
     owned: OwnedProxy
     wire: Wire
-    held_port: int
+    held: Wire
+    release: threading.Event
 
 
 def _marker() -> str:
     return uuid.uuid4().hex
 
 
-def _free_port() -> int:
-    with socket.socket() as reserve:
-        reserve.bind(("127.0.0.1", 0))
-        return int(reserve.getsockname()[1])
-
-
 def _deployment(
     name: str,
     backend: str,
@@ -86,7 +80,8 @@ def _peer(request: Request) -> Reply:
 
 def _held(release: threading.Event) -> Callable[[Request], Reply]:
     def respond(request: Request) -> Reply:
-        assert release.wait(timeout=120), "Held peer was never released"
+        if request.method == "POST":
+            assert release.wait(timeout=120), "Held peer was never released"
         return _peer(request)
 
     return respond
@@ -107,14 +102,14 @@ def _request(surface: str, model: str, pages: int, marker: str) -> tuple[str, di
 @pytest.fixture(scope="module")
 def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
     directory: Final = tmp_path_factory.mktemp("pdf-pre-call")
-    held_port: Final = _free_port()
-    with gateway_from_environment() as gateway, wire_server(_peer) as wire:
+    release: Final = threading.Event()
+    with gateway_from_environment() as gateway, wire_server(_peer) as wire, wire_server(_held(release)) as held:
         config: Final = owned_config(
             directory,
             [
                 _deployment(_WINDOW, "claude-haiku-5-5", wire.url, model_info={"max_input_tokens": _LIMIT}),
                 _deployment(_ITPM, "claude-sonnet-4-6", wire.url, itpm=_LIMIT),
-                _deployment(_BUDGET, "claude-opus-4-8", f"http://127.0.0.1:{held_port}"),
+                _deployment(_BUDGET, "claude-opus-4-8", held.url),
             ],
             litellm_settings={"cache": False},
             router_settings={
@@ -122,8 +117,11 @@ def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
                 "optional_pre_call_checks": ["enforce_model_rate_limits"],
             },
         )
-        with owned_proxy_process(gateway, directory, {}, config=config, workers=2) as owned:
-            yield _Rig(owned, wire, held_port)
+        try:
+            with owned_proxy_process(gateway, directory, {}, config=config, workers=2) as owned:
+                yield _Rig(owned, wire, held, release)
+        finally:
+            release.set()
 
 
 def _spend_row(call_id: str) -> dict[str, JsonValue]:
@@ -168,9 +166,10 @@ def test_model_itpm_check_rejects_a_pdf_whose_pages_exceed_the_limit(rig: _Rig)
 
 
 def test_key_budget_reservation_rejects_the_second_concurrent_pdf_request(rig: _Rig) -> None:
-    release: Final = threading.Event()
     requests: Final = tuple(_request("messages", _BUDGET, _PAGE_COUNT, _marker()) for _ in range(2))
-    with wire_server(_held(release), port=rig.held_port) as wire, rig.owned.gateway.scenario() as scenario:
+    wire: Final = rig.held
+    wire.drain()
+    with rig.owned.gateway.scenario() as scenario:
         key: Final = scenario.key(max_budget=0.01)
         headers: Final = {"Authorization": f"Bearer {key}"}
         with (
@@ -187,7 +186,7 @@ def test_key_budget_reservation_rejects_the_second_concurrent_pdf_request(rig: _
                     seconds=60,
                 )
             finally:
-                release.set()
+                rig.release.set()
             responses: Final = tuple(future.result(timeout=90) for future in futures)
         received: Final = wire.drain()
     assert sorted(response.status_code for response in responses) == [200, 422], [

tests/integration/routing/test_priority_scheduler_queue_cleanup.py

@@ -617,11 +617,19 @@ def test_concurrent_prioritized_burst_across_instances_and_endpoints_is_served(p
 @pytest.mark.timeout(PAIR_CELL_TIMEOUT)
 def test_fresh_queue_during_a_cooldown_is_refused_at_once(pair: Pair) -> None:
     group: Final = PAIR_GROUPS[2]
-    cooled(pair.second, pair.upstream, group)
-    client: Final = sdk(pair.second)
-    started: Final = time.monotonic()
-    with pytest.raises(openai.APIStatusError) as refused:
-        client.completions.create(model=group, prompt=new_marker(), extra_body={"priority": 1})
+    with ExitStack() as stack:
+        worker: Final = pinned(pair.second, stack)
+        cooled(worker, pair.upstream, group)
+        client: Final = openai.OpenAI(
+            base_url=str(worker.client.base_url),
+            api_key=worker.key,
+            max_retries=0,
+            timeout=30,
+            http_client=worker.client,
+        )
+        started: Final = time.monotonic()
+        with pytest.raises(openai.APIStatusError) as refused:
+            client.completions.create(model=group, prompt=new_marker(), extra_body={"priority": 1})
     assert refused.value.status_code == 429, refused.value.message
     assert NO_DEPLOYMENTS in refused.value.message
     assert_cut_short(time.monotonic() - started, injected_delay=ROUTER_TIMEOUT_SECONDS)

tests/integration/run.py

@@ -39,7 +39,7 @@ GITHUB_FILES: Final = frozenset(
 )
 
 
-PEER_GROUPS: Final = frozenset({"management", "mcp"})
+PEER_GROUPS: Final = frozenset({"management", "mcp", "accounting"})
 COST_MAP_GROUPS: Final = frozenset({"cost"})
 
 

tests/integration/security/test_fine_tuning_managed_ids.py

@@ -21,11 +21,6 @@ from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
 
 MANAGED_MODEL: Final = "gpt-4.1-managed-ids"
 DEPLOYMENT_KEY: Final = "sk-fine-tuning-managed-deployment"
-RUN: Final = uuid.uuid4().hex[:12]
-PROVIDER_FILE: Final = f"file-managed-{RUN}"
-PROVIDER_JOB: Final = f"ftjob-managed-{RUN}"
-DEFAULT_PROVIDER_FILE: Final = f"file-managed-default-{RUN}"
-DEFAULT_PROVIDER_JOB: Final = f"ftjob-managed-default-{RUN}"
 TRAINING_LINE: Final = b'{"messages": [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}]}\n'
 JOB_PATH: Final = re.compile(r"^/v1/fine_tuning/jobs/([^/]+?)(/cancel)?$")
 JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
@@ -190,7 +185,10 @@ class Rig:
 
 
 @contextmanager
-def _rig(directory: Path, litellm_settings: dict[str, JsonValue], file_id: str, job_id: str) -> Iterator[Rig]:
+def _rig(directory: Path, litellm_settings: dict[str, JsonValue]) -> Iterator[Rig]:
+    build: Final = uuid.uuid4().hex
+    file_id: Final = f"file-managed-{build}"
+    job_id: Final = f"ftjob-managed-{build}"
     with gateway_from_environment() as shared, wire_server(_responder(file_id, job_id)) as wire:
         config: Final = directory / "config.yaml"
         config.write_text(yaml.safe_dump(_config(wire, litellm_settings)))
@@ -239,17 +237,13 @@ def _rig(directory: Path, litellm_settings: dict[str, JsonValue], file_id: str,
 
 @pytest.fixture(scope="module")
 def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
-    with _rig(
-        tmp_path_factory.mktemp("fine-tuning-managed-ids"), {"require_managed_files": True}, PROVIDER_FILE, PROVIDER_JOB
-    ) as built:
+    with _rig(tmp_path_factory.mktemp("fine-tuning-managed-ids"), {"require_managed_files": True}) as built:
         yield built
 
 
 @pytest.fixture(scope="module")
 def default_rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
-    with _rig(
-        tmp_path_factory.mktemp("fine-tuning-managed-ids-default"), {}, DEFAULT_PROVIDER_FILE, DEFAULT_PROVIDER_JOB
-    ) as built:
+    with _rig(tmp_path_factory.mktemp("fine-tuning-managed-ids-default"), {}) as built:
         yield built
 
 
@@ -433,7 +427,7 @@ def test_keys_granted_the_retrieve_route_get_the_owner_check_and_the_decoded_job
         with rig.sdk(owner) as client:
             retrieved: Final = client.fine_tuning.jobs.with_raw_response.retrieve(rig.managed_job)
         assert [(r.method, r.target, r.headers["authorization"], r.body) for r in rig.wire.drain()] == [
-            ("GET", f"/v1/fine_tuning/jobs/{PROVIDER_JOB}", f"Bearer {DEPLOYMENT_KEY}", b"")
+            ("GET", f"/v1/fine_tuning/jobs/{rig.provider_job}", f"Bearer {DEPLOYMENT_KEY}", b"")
         ]
         assert retrieved.status_code == 200, retrieved.text
         assert Job.model_validate_json(retrieved.text) == _client_job(rig.managed_job, "running", rig.provider_file), (
@@ -455,7 +449,7 @@ def test_keys_granted_the_retrieve_route_get_default_job_ownership_and_decoded_i
         assert [
             (request.method, request.target, request.headers["authorization"], request.body)
             for request in default_rig.wire.drain()
-        ] == [("GET", f"/v1/fine_tuning/jobs/{DEFAULT_PROVIDER_JOB}", f"Bearer {DEPLOYMENT_KEY}", b"")]
+        ] == [("GET", f"/v1/fine_tuning/jobs/{default_rig.provider_job}", f"Bearer {DEPLOYMENT_KEY}", b"")]
         assert retrieved.status_code == 200, retrieved.text
         assert Job.model_validate_json(retrieved.text) == _client_job(
             default_rig.managed_job, "running", default_rig.provider_file
@@ -472,7 +466,7 @@ def test_owner_retrieve_reaches_the_provider_with_the_decoded_job_id(rig: Rig) -
     with rig.sdk(rig.owner) as client:
         retrieved: Final = client.fine_tuning.jobs.with_raw_response.retrieve(rig.managed_job)
     assert [(r.method, r.target, r.headers["authorization"], r.body) for r in rig.wire.drain()] == [
-        ("GET", f"/v1/fine_tuning/jobs/{PROVIDER_JOB}", f"Bearer {DEPLOYMENT_KEY}", b"")
+        ("GET", f"/v1/fine_tuning/jobs/{rig.provider_job}", f"Bearer {DEPLOYMENT_KEY}", b"")
     ]
     assert retrieved.status_code == 200, retrieved.text
     assert Job.model_validate_json(retrieved.text) == _client_job(rig.managed_job, "running", rig.provider_file), (
@@ -484,7 +478,7 @@ def test_owner_cancel_reaches_the_provider_with_the_decoded_job_id(rig: Rig) ->
     with rig.sdk(rig.owner) as client:
         cancelled: Final = client.fine_tuning.jobs.with_raw_response.cancel(rig.managed_job)
     assert [(r.method, r.target, r.headers["authorization"], r.body) for r in rig.wire.drain()] == [
-        ("POST", f"/v1/fine_tuning/jobs/{PROVIDER_JOB}/cancel", f"Bearer {DEPLOYMENT_KEY}", b"")
+        ("POST", f"/v1/fine_tuning/jobs/{rig.provider_job}/cancel", f"Bearer {DEPLOYMENT_KEY}", b"")
     ]
     assert cancelled.status_code == 200, cancelled.text
     assert Job.model_validate_json(cancelled.text) == _client_job(rig.managed_job, "cancelled", rig.provider_file), (