From 0915691d320a58072bf3104a8e05086bb8e85a22 Mon Sep 17 00:00:00 2001 From: yuneng-jiang Date: Wed, 14 Oct 2026 12:00:35 -0700 Subject: [PATCH] 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 | 2 +- tests/integration/README.md | 2 +- tests/integration/conftest.py | 1 + ..._user_updates_wedged_coordination_redis.py | 19 +++++------ ...vertex_anthropic_inline_tools_beta_wire.py | 16 +++++---- ...df_document_pre_call_checks_owned_proxy.py | 33 +++++++++---------- .../test_priority_scheduler_queue_cleanup.py | 18 +++++++--- tests/integration/run.py | 2 +- .../security/test_fine_tuning_managed_ids.py | 26 ++++++--------- 9 files changed, 62 insertions(+), 57 deletions(-) diff --git a/.circleci/scripts/run_integration.sh b/.circleci/scripts/run_integration.sh index be7e70d491c..00b3c5f8fdd 100644 --- a/.circleci/scripts/run_integration.sh +++ b/.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" diff --git a/tests/integration/README.md b/tests/integration/README.md index 572b80ad5c6..0f1da7b3d97 100644 --- a/tests/integration/README.md +++ b/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 diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index cb23713e6b2..a4bd6092a0c 100644 --- a/tests/integration/conftest.py +++ b/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 diff --git a/tests/integration/management/test_user_updates_wedged_coordination_redis.py b/tests/integration/management/test_user_updates_wedged_coordination_redis.py index 41fefdcd3b2..09b16e7d5ef 100644 --- a/tests/integration/management/test_user_updates_wedged_coordination_redis.py +++ b/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, diff --git a/tests/integration/providers/test_vertex_anthropic_inline_tools_beta_wire.py b/tests/integration/providers/test_vertex_anthropic_inline_tools_beta_wire.py index a6b039dc8f7..3175f2ec075 100644 --- a/tests/integration/providers/test_vertex_anthropic_inline_tools_beta_wire.py +++ b/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) diff --git a/tests/integration/routing/test_pdf_document_pre_call_checks_owned_proxy.py b/tests/integration/routing/test_pdf_document_pre_call_checks_owned_proxy.py index 9fad73f8080..c9a4e8e8881 100644 --- a/tests/integration/routing/test_pdf_document_pre_call_checks_owned_proxy.py +++ b/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], [ diff --git a/tests/integration/routing/test_priority_scheduler_queue_cleanup.py b/tests/integration/routing/test_priority_scheduler_queue_cleanup.py index bcfb28f790a..993da52fc67 100644 --- a/tests/integration/routing/test_priority_scheduler_queue_cleanup.py +++ b/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) diff --git a/tests/integration/run.py b/tests/integration/run.py index 43e606f46a7..99518e69d1f 100644 --- a/tests/integration/run.py +++ b/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"}) diff --git a/tests/integration/security/test_fine_tuning_managed_ids.py b/tests/integration/security/test_fine_tuning_managed_ids.py index 0c1a0f71b3f..2c98bca6eea 100644 --- a/tests/integration/security/test_fine_tuning_managed_ids.py +++ b/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), (