From 910c30381b03ce304ead4452ebcd4a6fa7b487b5 Mon Sep 17 00:00:00 2001 From: yuneng-jiang Date: Fri, 16 Oct 2026 15:45:29 -0700 Subject: [PATCH] test(integration): fix seven CircleCI integration tests that race or assume a worker spread (#47001) * test(integration): expect the tool search call replayed beside the function call for Anthropic targets * test(integration): wait for the team spend flush before killing a worker * test(integration): poll for the S3 attempt the batch sinks write concurrently * test(integration): queue the terminal object and its sibling into the same s3_v2 flush * test(integration): time the long Bedrock version against an equally long unversioned name * test(integration): pin each half of a burst to one worker instead of hoping the kernel spreads it * test(integration): assert parsed SSE events for live pings instead of transport chunk boundaries --- tests/integration/_support/wire.py | 28 +++++-- .../test_batch_line_items_cold_storage.py | 2 +- .../observability/test_s3_v2_upload_fanout.py | 20 ++++- ...drock_runtime_chat_completions_sad_wire.py | 6 +- ...ect_prompt_cache_breakpoint_owned_proxy.py | 80 +++++++++++-------- ...test_responses_bridge_tool_search_chaos.py | 7 +- ...stream_retry_budget_sources_owned_proxy.py | 78 +++++++++++++----- .../test_disable_end_user_cost_tracking.py | 1 + .../test_faked_responses_stream_chaos.py | 33 +++++--- 9 files changed, 175 insertions(+), 80 deletions(-) diff --git a/tests/integration/_support/wire.py b/tests/integration/_support/wire.py index b261973973c..e5ce909ab18 100644 --- a/tests/integration/_support/wire.py +++ b/tests/integration/_support/wire.py @@ -6,6 +6,7 @@ import time from collections.abc import Callable, Generator, Iterator, Mapping from contextlib import contextmanager from dataclasses import dataclass +from functools import partial from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from queue import SimpleQueue from types import MappingProxyType @@ -60,6 +61,7 @@ class Reply: abort_after: int | None = None gate_after_first: threading.Event | None = None pause_between_chunks: float = 0 + heartbeat: bytes | None = None headers: Mapping[str, str] = MappingProxyType({}) drop_connection: bool = False @@ -82,10 +84,14 @@ class Wire: return self.connected.qsize() - self.closed.qsize() -def _await_release(gate: threading.Event, closing: threading.Event) -> bool: +def _await_release( + gate: threading.Event, closing: threading.Event, beat: Callable[[], None] | None = None, every: float = 0.05 +) -> bool: while not closing.is_set(): - if gate.wait(timeout=0.05): + if gate.wait(timeout=every): return True + if beat is not None: + beat() return gate.is_set() @@ -117,6 +123,10 @@ def wire_server( super().finish() closed.put(f"{self.client_address[0]}:{self.client_address[1]}") + def write_chunk(self, chunk: bytes) -> None: + self.wfile.write(b"%x\r\n%s\r\n" % (len(chunk), chunk)) + self.wfile.flush() + def respond(self) -> None: headers: Final = {name.lower(): value for name, value in self.headers.items()} try: @@ -156,12 +166,20 @@ def wire_server( for index, chunk in enumerate(reply.chunks): if reply.abort_after == index: break - self.wfile.write(b"%x\r\n%s\r\n" % (len(chunk), chunk)) - self.wfile.flush() + self.write_chunk(chunk) if ( index == 0 and reply.gate_after_first is not None - and not _await_release(reply.gate_after_first, closing) + and not ( + _await_release(reply.gate_after_first, closing) + if reply.heartbeat is None + else _await_release( + reply.gate_after_first, + closing, + partial(self.write_chunk, reply.heartbeat), + reply.pause_between_chunks, + ) + ) ): break if reply.pause_between_chunks and index + 1 < len(reply.chunks): diff --git a/tests/integration/observability/test_batch_line_items_cold_storage.py b/tests/integration/observability/test_batch_line_items_cold_storage.py index 9c2a1ecd2c4..587a21e4b62 100644 --- a/tests/integration/observability/test_batch_line_items_cold_storage.py +++ b/tests/integration/observability/test_batch_line_items_cold_storage.py @@ -817,7 +817,7 @@ def test_a_failing_bucket_still_bills_the_batch_and_the_other_sink_keeps_its_lin assert [record["custom_id"] for record in _records(gcs_body)] == [custom_id] assert spend > 0 s3_name: Final = _object_name(f"{S3_PREFIX}/", batch.batch_id) - assert rig.s3.tried(s3_name) >= 1 + eventually(lambda: rig.s3.tried(s3_name), lambda tries: tries >= 1, POLLER_SECONDS) assert s3_name not in rig.s3.objects() diff --git a/tests/integration/observability/test_s3_v2_upload_fanout.py b/tests/integration/observability/test_s3_v2_upload_fanout.py index 01c1f563be4..d845d332237 100644 --- a/tests/integration/observability/test_s3_v2_upload_fanout.py +++ b/tests/integration/observability/test_s3_v2_upload_fanout.py @@ -420,7 +420,16 @@ def test_s3_v2_object_rejected_with_a_bucket_wide_code_is_delivered_once_the_fau def test_s3_v2_terminal_object_is_put_once_and_dropped_by_default(gateway: Gateway, tmp_path: Path) -> None: marker: Final = "s3toolarge" + uuid.uuid4().hex[:8] sink: Final = RejectingS3Sink(reject_marker=f"{marker}-huge", reject_status=400, reject_code="EntityTooLarge") - with wire_server(_chat_reply) as provider, wire_server(sink.respond) as bucket: + flushing: Final = threading.Event() + release: Final = threading.Event() + + def respond(request: Request) -> Reply: + if f"{marker}-held".encode() in request.body: + flushing.set() + assert release.wait(timeout=60), "the held flush was never released" + return sink.respond(request) + + with wire_server(_chat_reply) as provider, wire_server(respond) as bucket: config: Final = _s3_config(tmp_path, bucket.url, {"s3_batch_file_upload": False}) with ( owned_proxy(gateway, tmp_path, {"DEFAULT_S3_FLUSH_INTERVAL_SECONDS": "2"}, config=config) as candidate, @@ -428,11 +437,16 @@ def test_s3_v2_terminal_object_is_put_once_and_dropped_by_default(gateway: Gatew ): model: Final = scenario.model(api_base=provider.url + "/v1", api_key="synthetic-provider-key") key: Final = scenario.key(models=[model]) + _send(candidate, model, key, f"{marker}-held") + eventually(flushing.is_set, bool, seconds=30) _send(candidate, model, key, f"{marker}-huge") - _send_and_wait_until_landed(candidate, model, key, sink, f"{marker}-sibling") + _send(candidate, model, key, f"{marker}-sibling") + release.set() + eventually(sink.landed_ids, lambda landed: f"{marker}-sibling" in landed, seconds=60) _send_and_wait_until_landed(candidate, model, key, sink, f"{marker}-second-flush") _send_and_wait_until_landed(candidate, model, key, sink, f"{marker}-third-flush") - assert sum(1 for r in provider.drain() if r.method == "POST") == 4 + assert sum(1 for r in provider.drain() if r.method == "POST") == 5 + assert sink.landed_ids() == {f"{marker}-{name}" for name in ("held", "sibling", "second-flush", "third-flush")} assert sink.rejected_attempts == 1, ( f"an EntityTooLarge object was PUT {sink.rejected_attempts} times next to delivered siblings; " "with the default s3_drop_on_terminal_error it must be attempted once and dropped" diff --git a/tests/integration/providers/test_bedrock_runtime_chat_completions_sad_wire.py b/tests/integration/providers/test_bedrock_runtime_chat_completions_sad_wire.py index 0860d79459e..54923fe0c97 100644 --- a/tests/integration/providers/test_bedrock_runtime_chat_completions_sad_wire.py +++ b/tests/integration/providers/test_bedrock_runtime_chat_completions_sad_wire.py @@ -27,6 +27,7 @@ BAD_KEY: Final = "sk-synthetic-bad-key" NATIVE_TARGET: Final = "/openai/v1/chat/completions" CONVERSE_TARGET: Final = f"/model/{GPT}/converse" LONG_VERSION_GPT: Final = "openai.gpt-" + "1" * 30000 +LONG_UNVERSIONED_GPT: Final = "openai.gpt-x" + "1" * 30000 PNG_DATA_URL: Final = ( "data:image/png;base64," "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==" @@ -258,11 +259,12 @@ def test_thirty_thousand_digit_version_is_classified_quickly_and_served_by_conve marker: Final = uuid.uuid4().hex control_marker: Final = uuid.uuid4().hex with wire_server(respond) as wire, gateway.scenario() as scenario, ThreadPoolExecutor(max_workers=1) as pool: - control_model: Final = _deployment(scenario, wire) + control_model: Final = _deployment(scenario, wire, model=f"bedrock/{LONG_UNVERSIONED_GPT}") model: Final = _deployment(scenario, wire, model=f"bedrock/{LONG_VERSION_GPT}") control: Final = timed(lambda: _chat(gateway, control_model, control_marker)) assert _content(control.value) == answer(control_marker), control.value.text - _only_request(wire, control_marker) + control_request: Final = _only_request(wire, control_marker) + assert target_of(control_request) == f"/model/{LONG_UNVERSIONED_GPT}/converse", control_request.target liveliness: Final = pool.submit(_timed_liveliness, gateway) served: Final = timed(lambda: _chat(gateway, model, marker)) response: Final = served.value diff --git a/tests/integration/providers/test_openai_dialect_prompt_cache_breakpoint_owned_proxy.py b/tests/integration/providers/test_openai_dialect_prompt_cache_breakpoint_owned_proxy.py index 920002316bc..00f2c430ea5 100644 --- a/tests/integration/providers/test_openai_dialect_prompt_cache_breakpoint_owned_proxy.py +++ b/tests/integration/providers/test_openai_dialect_prompt_cache_breakpoint_owned_proxy.py @@ -2,8 +2,8 @@ import asyncio import re import threading from collections.abc import Callable, Iterator -from concurrent.futures import ThreadPoolExecutor -from dataclasses import dataclass +from concurrent.futures import Future, ThreadPoolExecutor +from dataclasses import dataclass, replace from pathlib import Path from queue import SimpleQueue from types import MappingProxyType @@ -15,7 +15,7 @@ import psutil import pytest from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment from integration._support.database import read_rows -from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process +from integration._support.process import OwnedProxy, WorkerHealthcheck, graceful_stop_seconds, owned_proxy_process from integration._support.responses_vendor import newest_marker from integration._support.wire import Reply, Request, Wire, wire_server from integration.providers._cache_control_marks_support import owned_config @@ -27,6 +27,7 @@ from integration.providers._mantle_gpt_prompt_cache_support import ( SYSTEM_POINT, TOKEN, Outcome, + Step, assert_answered, assert_burst_landed, assert_wire, @@ -60,7 +61,6 @@ _CELL_TIMEOUT: Final = int(2 * graceful_stop_seconds() + 120) _RESTART_TIMEOUT: Final = int(4 * graceful_stop_seconds() + 240) _HELD_BURST: Final = 20 _SPREAD_BURST: Final = 12 -_SPREAD_ATTEMPTS: Final = 20 _MANTLE_NAME: Final = "mantle-gpt-cache-owned" @@ -122,7 +122,9 @@ def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]: release.set() held: Final[SimpleQueue[str]] = SimpleQueue() with gateway_from_environment() as environment, wire_server(_peer(release, held)) as wire: - with owned_proxy_process(environment, directory, _overrides(wire), workers=2) as owned: + with owned_proxy_process( + environment, directory, _overrides(wire), workers=2, healthcheck=WorkerHealthcheck.LENIENT + ) as owned: eventually(lambda: len(_started_workers(owned.log)), lambda count: count == 2, seconds=60) wire.drain() yield _Rig(owned.gateway, wire, owned, release, held) @@ -204,38 +206,48 @@ def test_b11_foundry_gpt6_chat_stays_on_the_plain_chat_wire(rig: _Rig) -> None: success_row(name, marker) -def _spread_burst(rig: _Rig, name: str, workers: tuple[int, ...]) -> dict[int, int]: - plan: Final = plan_burst(_SPREAD_BURST) - _drain(rig.held) - rig.wire.drain() - rig.release.clear() - with ThreadPoolExecutor(max_workers=_SPREAD_BURST) as pool: - futures: Final = tuple( - pool.submit(send, rig.gateway, endpoint, request_body(endpoint, name, prompt_text(marker), stream=stream)) - for endpoint, stream, marker in plan - ) - eventually(rig.held.qsize, lambda size: size == _SPREAD_BURST, seconds=60) - held_by: Final = {pid: _open_peer_connections(pid, rig.wire.url) for pid in workers} - rig.release.set() - outcomes: Final = tuple(future.result() for future in futures) - assert sum(held_by.values()) == _SPREAD_BURST, held_by - assert_burst_landed( - rig.wire, name, tuple((marker, outcome) for (_, _, marker), outcome in zip(plan, outcomes)), marked=True - ) - return held_by - - @pytest.mark.timeout(_CELL_TIMEOUT) def test_e6_every_worker_of_a_two_worker_proxy_marks_the_mantle_request(rig: _Rig) -> None: - with rig.gateway.scenario() as scenario: + plan: Final = plan_burst(_SPREAD_BURST) + half: Final = len(plan) // 2 + with ( + rig.gateway.scenario() as scenario, + httpx.Client( + base_url=str(rig.gateway.client.base_url), + timeout=60, + trust_env=False, + limits=httpx.Limits(max_keepalive_connections=0), + ) as unpooled, + ThreadPoolExecutor(max_workers=len(plan)) as pool, + ): name: Final = mantle_deployment(rig.gateway, scenario, rig.wire) - workers: Final = _live_workers(rig.owned.log) - assert len(workers) == 2, workers - for _ in range(_SPREAD_ATTEMPTS): - if all(count > 0 for count in _spread_burst(rig, name, workers).values()): - break - else: - raise AssertionError("One worker never took a marked request") + first, second = sorted(_live_workers(rig.owned.log)) + gateway: Final = replace(rig.gateway, client=unpooled) + _drain(rig.held) + rig.wire.drain() + rig.release.clear() + + def submit(part: tuple[Step, ...]) -> tuple[Future[Outcome], ...]: + return tuple( + pool.submit(send, gateway, endpoint, request_body(endpoint, name, prompt_text(marker), stream=stream)) + for endpoint, stream, marker in part + ) + + try: + with rig.owned.suspended(second): + on_first: Final = submit(plan[:half]) + eventually(rig.held.qsize, lambda size: size == half, seconds=60) + with rig.owned.suspended(first): + on_second: Final = submit(plan[half:]) + eventually(rig.held.qsize, lambda size: size == len(plan), seconds=60) + held_by: Final = {pid: _open_peer_connections(pid, rig.wire.url) for pid in (first, second)} + finally: + rig.release.set() + outcomes: Final = tuple(future.result() for future in (*on_first, *on_second)) + assert held_by == {first: half, second: len(plan) - half}, held_by + assert_burst_landed( + rig.wire, name, tuple((marker, outcome) for (_, _, marker), outcome in zip(plan, outcomes)), marked=True + ) def _mantle_config(wire: Wire, directory: Path, **litellm_params: JsonValue) -> Path: diff --git a/tests/integration/providers/test_responses_bridge_tool_search_chaos.py b/tests/integration/providers/test_responses_bridge_tool_search_chaos.py index 0101824054a..6beccefe58f 100644 --- a/tests/integration/providers/test_responses_bridge_tool_search_chaos.py +++ b/tests/integration/providers/test_responses_bridge_tool_search_chaos.py @@ -529,7 +529,7 @@ def _marked_posts(received: tuple[Request, ...], marker: str) -> tuple[Request, @pytest.mark.timeout(_OWNED_PROXY_CELL_SECONDS) -def test_a_previous_response_id_turn_replays_the_function_call_without_the_search_call(owned_rig: _OwnedRig) -> None: +def test_a_previous_response_id_turn_replays_the_search_call_beside_the_function_call(owned_rig: _OwnedRig) -> None: marker: Final = uuid.uuid4().hex owned_rig.wire.drain() proxy: Final = owned_rig.proxy.gateway @@ -566,11 +566,8 @@ def test_a_previous_response_id_turn_replays_the_function_call_without_the_searc _assert_forwarded_with_tool_search(opening) _assert_forwarded_with_tool_search(replay) messages: Final = rv.ITEMS.validate_python(rv.JSON_OBJECT.validate_json(replay.body)["messages"]) - assert "srvtoolu_" not in json.dumps(messages) and "tool_search" not in json.dumps(messages), messages (assistant,) = [turn for turn in messages if turn["role"] == "assistant"] - blocks: Final = rv.ITEMS.validate_python(assistant["content"]) - tool_uses: Final = [(block["type"], block.get("id")) for block in blocks if block["type"] == "tool_use"] - assert tool_uses == [("tool_use", f"toolu_{marker}")], blocks + assert assistant["content"] == _message(marker)["content"], assistant assert [block["tool_use_id"] for block in _tool_results(messages)] == [f"toolu_{marker}"], messages diff --git a/tests/integration/routing/test_messages_stream_retry_budget_sources_owned_proxy.py b/tests/integration/routing/test_messages_stream_retry_budget_sources_owned_proxy.py index 93e4f97ae1e..0257b8372b0 100644 --- a/tests/integration/routing/test_messages_stream_retry_budget_sources_owned_proxy.py +++ b/tests/integration/routing/test_messages_stream_retry_budget_sources_owned_proxy.py @@ -1,7 +1,9 @@ import json +import threading import uuid from collections.abc import Callable, Iterator from dataclasses import dataclass +from itertools import groupby, islice from pathlib import Path from typing import Final, Literal @@ -12,6 +14,7 @@ from integration._support.anthropic_sse import ( LIFECYCLE, PING, Attempts, + SseEvent, delta_text, dropping_reply, error_body, @@ -55,7 +58,9 @@ def _served_id(backend: str, marker: str, attempt: int) -> str: return f"msg_{backend}_{marker}_a{attempt}" -def _primary_reply(behavior: str, attempt: int, full: tuple[bytes, bytes, bytes, bytes]) -> Reply: +def _primary_reply( + behavior: str, attempt: int, full: tuple[bytes, bytes, bytes, bytes], content_gate: threading.Event +) -> Reply: dropped: Final = dropping_reply(full, abort_after=1) match behavior: case "drop-once": @@ -63,7 +68,13 @@ def _primary_reply(behavior: str, attempt: int, full: tuple[bytes, bytes, bytes, case "drop-always": return dropped case "hold-ping": - return stream_reply((full[0], PING, full[1] + full[2] + full[3]), pause=0.25) + return Reply( + content_type="text/event-stream", + chunks=(full[0], full[1] + full[2] + full[3]), + gate_after_first=content_gate, + heartbeat=PING, + pause_between_chunks=0.25, + ) case "drop-then-too-long": if attempt == 1: return dropped @@ -81,7 +92,7 @@ def _fallback_reply(behavior: str, full: tuple[bytes, bytes, bytes, bytes]) -> R return stream_reply(full) -def _respond(attempts: Attempts) -> Callable[[Request], Reply]: +def _respond(attempts: Attempts, content_gate: threading.Event) -> Callable[[Request], Reply]: def respond(request: Request) -> Reply: assert (request.method, request.target) == ("POST", "/v1/messages"), request assert request.headers["x-api-key"] == _API_KEY, request.headers @@ -93,11 +104,21 @@ def _respond(attempts: Attempts) -> Callable[[Request], Reply]: full: Final = message_stream(_served_id(backend, marker, attempt), backend, _TEXT) if backend != _PRIMARY: return _fallback_reply(behavior, full) - return _primary_reply(behavior, attempt, full) + return _primary_reply(behavior, attempt, full, content_gate) return respond +def _stream_body(model: str, behavior: Behavior, marker: str, **extra: JsonValue) -> dict[str, JsonValue]: + return { + "model": model, + "max_tokens": 16, + "stream": True, + "messages": [{"role": "user", "content": f"{behavior}:{marker}"}], + **extra, + } + + def _deployment(name: str, backend: str, **extra: JsonValue) -> dict[str, JsonValue]: return { "model_name": name, @@ -137,19 +158,10 @@ def _config(wire: Wire, directory: Path) -> Path: class _Rig: proxy: Gateway attempts: Attempts + content_gate: threading.Event def stream(self, model: str, behavior: Behavior, marker: str, **extra: JsonValue) -> httpx.Response: - return self.proxy.request( - "POST", - "/v1/messages", - { - "model": model, - "max_tokens": 16, - "stream": True, - "messages": [{"role": "user", "content": f"{behavior}:{marker}"}], - **extra, - }, - ) + return self.proxy.request("POST", "/v1/messages", _stream_body(model, behavior, marker, **extra)) def attempts_on(self, backend: str, marker: str) -> int: return self.attempts.count(f"{backend}:{marker}") @@ -158,10 +170,14 @@ class _Rig: @pytest.fixture(scope="module") def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]: attempts: Final = Attempts() + content_gate: Final = threading.Event() directory: Final = tmp_path_factory.mktemp("messages-retry-budget-sources") - with gateway_from_environment() as gateway, wire_server(_respond(attempts)) as wire: + with gateway_from_environment() as gateway, wire_server(_respond(attempts, content_gate)) as wire: with owned_proxy(gateway, directory, {}, config=_config(wire, directory), workers=2) as proxy: - yield _Rig(proxy, attempts) + try: + yield _Rig(proxy, attempts, content_gate) + finally: + content_gate.set() def _assert_completed(response: httpx.Response, served_id: str) -> None: @@ -202,12 +218,32 @@ def test_request_num_retries_zero_turns_the_router_budget_off(rig: _Rig) -> None assert rig.attempts_on(_PRIMARY, marker) == 1 +def _live_events(lines: Iterator[str]) -> Iterator[SseEvent]: + for has_text, block in groupby(lines, key=bool): + if has_text: + yield from parse_sse("\n".join(block)) + + def test_lifecycle_frames_are_held_until_content_while_pings_go_out_live(rig: _Rig) -> None: marker: Final = uuid.uuid4().hex - response: Final = rig.stream(_ROUTER_BUDGET, "hold-ping", marker) - assert response.status_code == 200, response.text - events: Final = parse_sse(response.text) - assert event_types(events) == ("ping", *LIFECYCLE), events + rig.content_gate.clear() + try: + with rig.proxy.client.stream( + "POST", + "/v1/messages", + json=_stream_body(_ROUTER_BUDGET, "hold-ping", marker), + headers={"Authorization": f"Bearer {rig.proxy.key}"}, + ) as response: + assert response.status_code == 200, response.read() + arriving: Final = _live_events(response.iter_lines()) + before_content: Final = tuple(islice(arriving, 2)) + rig.content_gate.set() + after_content: Final = tuple(arriving) + finally: + rig.content_gate.set() + assert event_types(before_content) == ("ping", "ping"), before_content + events: Final = before_content + after_content + assert tuple(kind for kind in event_types(events) if kind != "ping") == LIFECYCLE, events assert message_id(events) == _served_id(_PRIMARY, marker, 1), events assert delta_text(events) == _TEXT, events assert rig.attempts_on(_PRIMARY, marker) == 1 diff --git a/tests/integration/spend/test_disable_end_user_cost_tracking.py b/tests/integration/spend/test_disable_end_user_cost_tracking.py index f3ec572b69a..d0233378d1e 100644 --- a/tests/integration/spend/test_disable_end_user_cost_tracking.py +++ b/tests/integration/spend/test_disable_end_user_cost_tracking.py @@ -695,6 +695,7 @@ def test_flag_on_burst_lands_each_row_once_through_sink_outage_and_worker_kill(r before_kill: Final = _burst(rig, models, key, calls[: len(_SURFACES) * 2]) _assert_burst_landed_once(rig, models, before_kill) assert not rig.lens.outages.acquire(blocking=False), "the Lens sink outage was never hit" + _team_requests_flushed(team, len(before_kill)) victim: Final = min(rig.proxy.worker_pids()) replacement: Final = rig.proxy.kill_worker(victim) assert replacement != victim, (victim, replacement) diff --git a/tests/integration/streaming/test_faked_responses_stream_chaos.py b/tests/integration/streaming/test_faked_responses_stream_chaos.py index 28efe912157..fca99c9d525 100644 --- a/tests/integration/streaming/test_faked_responses_stream_chaos.py +++ b/tests/integration/streaming/test_faked_responses_stream_chaos.py @@ -24,7 +24,7 @@ from integration._support.client import ( string_value, ) from integration._support.database import read_rows -from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process +from integration._support.process import OwnedProxy, WorkerHealthcheck, graceful_stop_seconds, owned_proxy_process from integration._support.wire import Reply, Request, wire_server from integration._support.ports import allocate_port from integration.streaming.faked_responses_case import ( @@ -111,7 +111,14 @@ def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]: port: Final = allocate_port() with ( gateway_from_environment() as shared, - owned_proxy_process(shared, directory, {}, config=_config(port, directory), workers=2) as owned, + owned_proxy_process( + shared, + directory, + {}, + config=_config(port, directory), + workers=2, + healthcheck=WorkerHealthcheck.LENIENT, + ) as owned, ): yield _Rig(port, owned) @@ -306,9 +313,11 @@ class _Held: return _respond(request) -async def _held_burst(gateway: Gateway, calls: tuple[_Call, ...], held: _Held) -> asyncio.Task[tuple[_Served, ...]]: +async def _held_burst( + gateway: Gateway, calls: tuple[_Call, ...], held: _Held, *, already_held: int = 0 +) -> asyncio.Task[tuple[_Served, ...]]: burst: Final = asyncio.create_task(_burst(gateway, calls, tolerate_transport_errors=True)) - await asyncio.to_thread(eventually, held.markers.qsize, lambda size: size == len(calls), 60) + await asyncio.to_thread(eventually, held.markers.qsize, lambda size: size == already_held + len(calls), 60) return burst @@ -319,12 +328,18 @@ def _held_by(rig: _Rig, upstream: str) -> Mapping[int, int]: async def test_held_mixed_burst_spreads_over_both_workers_and_each_value_arrives_once(rig: _Rig) -> None: calls: Final = _calls(30) held: Final = _Held(threading.Event(), SimpleQueue()) + first, second = sorted(rig.proxy.worker_pids()) with wire_server(held.respond, port=rig.port) as wire: - burst: Final = await _held_burst(rig.gateway, calls, held) - held_by: Final = _held_by(rig, wire.url) - held.release.set() - served: Final = await burst - assert len(held_by) == 2 and all(held_by.values()) and sum(held_by.values()) == 30, held_by + try: + with rig.proxy.suspended(second): + on_first: Final = await _held_burst(rig.gateway, calls[:15], held) + with rig.proxy.suspended(first): + on_second: Final = await _held_burst(rig.gateway, calls[15:], held, already_held=15) + held_by: Final = _held_by(rig, wire.url) + finally: + held.release.set() + served: Final = await on_first + await on_second + assert held_by == {first: 15, second: 15}, held_by assert len(served) == 30 for item in served: _assert_each_value_once(item)