test(integration): keep owned peers off ephemeral ports (#51019)

Co-authored-by: mateo <277851410+mateo-berri@users.noreply.github.com>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

tests/integration/README.md

@@ -28,7 +28,7 @@ Owned proxies disable `.env` loading by default. A contract testing dotenv behav
 
 Owned proxy workers configure a 32-thread executor before application startup, preserving additional worker startup hooks. Held synchronous provider calls must not consume a CPU-dependent executor limit before the rest of a fault burst can start. The proxy entrypoint imports LiteLLM inside `main`, so spawned workers begin supervision before importing the application. Worker healthchecks have a five-second deadline by default, so uvicorn kills a worker that stays paused longer than that. Tests never signal proxy processes themselves. `OwnedProxy.kill_worker` SIGKILLs a worker and returns once its replacement answers readiness with its own pid, `replacing_worker` does the same for a kill that has to land mid-burst, and `suspended` pauses a worker or the supervisor, which only works on a proxy launched with `healthcheck=WorkerHealthcheck.LENIENT`. The `kill()` that `replacing_worker` yields returns as soon as the worker is dead, so a test can release requests it holds on the surviving worker right away. Uvicorn may take a full worker healthcheck to notice the death, so the block waits for the supervisor to reap the worker and for the replacement only when it exits. Helper processes that are not the proxy stop through `stop_child`, `stop_processes` and `stop_process_group` and pause through `paused`. `scripts/check_test_quality.py` rule TQ010 rejects a raw `os.kill`, SIGKILL / SIGSTOP / SIGCONT, `.kill()`, `.terminate()`, `.suspend()` or `.send_signal()` anywhere under `tests/integration` outside `_support`, and TQ011 rejects an assertion that caps an elapsed time with a fixed number. Bound timings with `_support/timing.py` instead, relative to a measured control or to the injected delay. `run.py` measures how long this box takes to import the proxy application before it starts pytest, records it in `boot-calibration.json` beside the results and exports it as `INTEGRATION_IMPORT_SECONDS`. Readiness deadlines, graceful stops and the default per-test timeout are multiples of that measurement (`_support/budget.py`), so a slower box gets longer budgets and a two-worker proxy gets twice the readiness deadline of a one-worker one. Readiness requires responsive worker identities that are still alive and a positively identified owned listener. Unreadable socket records cannot establish ownership. Crash tests own their client tasks with `TaskGroup`, release held peers in `finally` and verify that the intended worker dies. A test requiring traffic on both workers prepares that traffic explicitly instead of assuming OS load balancing, for example by sending each half while the other worker is suspended
 
-Any port the harness picks before the server that listens on it has started comes from `allocate_port()` in `_support/ports.py`. That covers owned proxies, scripted upstreams, fake providers, OTLP sinks, database relays and Redis. A server that binds port 0 itself and keeps that socket, such as `wire_server` without a port, needs no allocation. Each pytest process claims a block of 200 ports between 10000 and 32767 by holding an exclusive lock file per block in the system temp directory, so two runs on one machine, such as parallel worktrees, never share a block, and the OS drops the lock when the process exits. The range sits above the fixed service ports and below the Linux and macOS ephemeral ranges, so a port-0 bind or an outbound connection never lands in a block. Ports inside a block go out in order and are never handed out twice in a run, a port that something outside the harness already holds is skipped, and a process that uses up its block claims the next free one. A test that restarts a server on the same port allocates that port up front instead of reading it back from the first server. TQ012 in `scripts/check_test_quality.py` rejects reading a port back from a port-0 bind anywhere under `tests/integration`
+Every owned listener's default port comes from `allocate_port()` in `_support/ports.py`. That covers owned proxies, scripted upstreams, fake providers, OTLP sinks, database relays and Redis, including peers such as `wire_server` that bind and keep their own sockets. Keeping those listeners out of the ephemeral range prevents them from taking a port Prisma selected with a bind-and-close probe before its query engine starts. Each pytest process claims a block of 200 ports between 10000 and 32767 by holding an exclusive lock file per block in the system temp directory, so two runs on one machine, such as parallel worktrees, never share a block, and the OS drops the lock when the process exits. The range sits above the fixed service ports and below the Linux and macOS ephemeral ranges, so a port-0 bind or an outbound connection never lands in a block. Ports inside a block go out in order and are never handed out twice in a run, a port that something outside the harness already holds is skipped, and a process that uses up its block claims the next free one. A test that restarts a server on the same port allocates that port up front instead of reading it back from the first server. TQ012 in `scripts/check_test_quality.py` rejects reading a port back from a port-0 bind anywhere under `tests/integration`
 
 The CircleCI workflow starts its own database and Redis, restricts test-phase egress to its owned services and writes JUnit plus an executed-node manifest. Missing setup, failed cleanup or a selected test with neither a passed call nor a skip fail qualification. Skipped nodes are listed under `skipped` in `execution.json`, so the skip reasons double as the open bug list. `GITHUB_FILES` in `run.py` lists the files excluded from the `run_integration.sh` selection
 

tests/integration/_support/asgi.py

@@ -12,13 +12,15 @@ from typing import Final
 import uvicorn
 from starlette.types import ASGIApp
 
+from tests.integration._support.ports import allocate_port
+
 
 @contextmanager
 def asgi_server(
     app: ASGIApp, *, before_stop: Callable[[], None] | None = None, port: int = 0, keepalive_seconds: int = 30
 ) -> Iterator[str]:
     with socket.socket() as listener:
-        listener.bind(("127.0.0.1", port))
+        listener.bind(("127.0.0.1", port or allocate_port()))
         bound_port: Final = listener.getsockname()[1]
         server: Final = uvicorn.Server(
             uvicorn.Config(

tests/integration/_support/async_server.py

@@ -7,6 +7,8 @@ from contextlib import suppress
 from threading import Thread
 from typing import Final
 
+from tests.integration._support.ports import allocate_port
+
 Handler = Callable[[asyncio.StreamReader, asyncio.StreamWriter], Awaitable[None]]
 
 
@@ -19,7 +21,7 @@ async def close_writer(writer: asyncio.StreamWriter) -> None:
 class LoopbackServer:
     def __init__(self, port: int, handler: Handler, ssl_context: ssl.SSLContext | None = None) -> None:
         self.loop: Final = asyncio.new_event_loop()
-        self._port: Final = port
+        self._port: Final = port or allocate_port()
         self._handler: Final = handler
         self._ssl_context: Final = ssl_context
         self._server: asyncio.Server | None = None

tests/integration/_support/aws_control_plane.py

@@ -18,6 +18,8 @@ from integration._support.sigv4 import canonical_query, signature
 from integration._support.wire import Reply
 from pydantic import JsonValue
 
+from tests.integration._support.ports import allocate_port
+
 _AUTHORIZATION: Final = re.compile(
     r"^AWS4-HMAC-SHA256 Credential=(?P<key>[^/]+)/(?P<scope>[^,]+), "
     r"SignedHeaders=(?P<signed>[^,]+), Signature=(?P<signature>[0-9a-f]{64})$"
@@ -267,7 +269,7 @@ def control_plane(directory: Path, hosts: tuple[str, ...]) -> Generator[ControlP
         block_on_close = False
         request_queue_size = 128
 
-    with Server(("127.0.0.1", 0), Proxy) as server:
+    with Server(("127.0.0.1", allocate_port()), Proxy) as server:
         thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
         thread.start()
         try:

tests/integration/_support/device_login.py

@@ -22,6 +22,8 @@ from integration._support.tls import server_context, write_self_signed_cert
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue
 
+from tests.integration._support.ports import allocate_port
+
 CHATGPT_AUTH_HOST: Final = "auth.openai.com"
 GITHUB_HOST: Final = "github.com"
 GITHUB_API_HOST: Final = "api.github.com"
@@ -366,7 +368,7 @@ def _tunnel(
                 _pipe(upstream, self.request)
                 outbound.join(timeout=12)
 
-    with _TunnelServer(("127.0.0.1", 0), Tunnel) as server:
+    with _TunnelServer(("127.0.0.1", allocate_port()), Tunnel) as server:
         thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
         thread.start()
         try:
@@ -378,9 +380,7 @@ def _tunnel(
 
 
 @contextmanager
-def device_login_peers(
-    directory: Path, provider: Callable[[Request], Reply] | None = None
-) -> Generator[Peers]:
+def device_login_peers(directory: Path, provider: Callable[[Request], Reply] | None = None) -> Generator[Peers]:
     cert, key = write_self_signed_cert(directory, (*AUTH_HOSTS, COPILOT_API_HOST))
     switches: Final = Switches(threading.Event(), threading.Event(), threading.Event(), threading.Event())
     authorities: Final[SimpleQueue[str]] = SimpleQueue()

tests/integration/_support/forward_proxy.py

@@ -10,6 +10,8 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from queue import SimpleQueue
 from typing import Final
 
+from tests.integration._support.ports import allocate_port
+
 
 @dataclass(frozen=True, slots=True)
 class Tunnel:
@@ -61,7 +63,7 @@ def refusing_forward_proxy(port: int = 0) -> Generator[ForwardProxy, None, None]
         def log_message(self, format: str, *args: object) -> None:
             pass
 
-    with ThreadingHTTPServer(("127.0.0.1", port), Handler) as server:
+    with ThreadingHTTPServer(("127.0.0.1", port or allocate_port()), Handler) as server:
         thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
         thread.start()
         try:
@@ -105,7 +107,7 @@ def tunnelling_forward_proxy(destination_port: int, port: int = 0) -> Generator[
         def log_message(self, format: str, *args: object) -> None:
             pass
 
-    with ThreadingHTTPServer(("127.0.0.1", port), Handler) as server:
+    with ThreadingHTTPServer(("127.0.0.1", port or allocate_port()), Handler) as server:
         thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
         thread.start()
         try:

tests/integration/_support/idle_gateway.py

@@ -18,6 +18,8 @@ from typing import Final
 
 from integration._support.wire import Reply, Request
 
+from tests.integration._support.ports import allocate_port
+
 MARKER: Final = re.compile(rb"idle-[0-9a-f]{16}")
 _LOOPBACK: Final = "127.0.0.1"
 
@@ -193,7 +195,7 @@ def idle_gateway(
     port: int = 0,
 ) -> Generator[IdleGateway]:
     arrivals: Final[SimpleQueue[Arrival]] = SimpleQueue()
-    with _Server(port, respond, drop_after, drop, arrivals) as server:
+    with _Server(port or allocate_port(), respond, drop_after, drop, arrivals) as server:
         thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
         thread.start()
         try:

tests/integration/_support/mail.py

@@ -10,6 +10,8 @@ from email.message import Message
 from queue import SimpleQueue
 from typing import Final
 
+from tests.integration._support.ports import allocate_port
+
 
 @dataclass(frozen=True, slots=True)
 class Delivery:
@@ -112,7 +114,7 @@ def smtp_sink() -> Generator[Mailbox, None, None]:
         allow_reuse_address = True
         daemon_threads = False
 
-    with OwnedServer(("127.0.0.1", 0), Handler) as server:
+    with OwnedServer(("127.0.0.1", allocate_port()), Handler) as server:
         mailbox: Final = Mailbox("127.0.0.1", server.server_address[1])
         thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
         thread.start()

tests/integration/_support/responses_ws_peer.py

@@ -21,6 +21,8 @@ from integration._support.tls import server_context
 from pydantic import JsonValue
 from websockets.asyncio.server import ServerConnection, serve
 
+from tests.integration._support.ports import allocate_port
+
 PROVIDER_MODEL: Final = "ws-peer-model"
 USAGE: Final = {"input_tokens": 5, "output_tokens": 2, "total_tokens": 7}
 TERMINAL: Final = frozenset({"response.completed", "response.failed", "error"})
@@ -105,7 +107,10 @@ async def _serve(
     script: Script,
 ) -> None:
     async with serve(
-        lambda connection: _answer(connection, paths, credentials, frames, script), "127.0.0.1", 0, ssl=tls
+        lambda connection: _answer(connection, paths, credentials, frames, script),
+        "127.0.0.1",
+        allocate_port(),
+        ssl=tls,
     ) as server:
         ports.put(next(iter(server.sockets)).getsockname()[1])
         await stop.wait()

tests/integration/_support/riva.py

@@ -15,6 +15,8 @@ from typing import Final
 import grpc
 from riva.client.proto import riva_asr_pb2, riva_asr_pb2_grpc
 
+from tests.integration._support.ports import allocate_port
+
 SAMPLE_RATE_HZ: Final = 16000
 OFFLINE_ONLY_REFUSAL: Final = (
     "Error: Unavailable model requested given these parameters: language_code=en-US; sample_rate=16000; type=online;"
@@ -139,7 +141,7 @@ def riva_server(
     options: Final = () if max_receive_bytes is None else (("grpc.max_receive_message_length", max_receive_bytes),)
     server: Final = grpc.server(futures.ThreadPoolExecutor(max_workers=8), options=options)
     riva_asr_pb2_grpc.add_RivaSpeechRecognitionServicer_to_server(_Asr(words, offline_only, stall, calls), server)
-    port: Final = server.add_insecure_port("127.0.0.1:0")
+    port: Final = server.add_insecure_port(f"127.0.0.1:{allocate_port()}")
     server.start()
     try:
         yield RivaServer(f"127.0.0.1:{port}", calls)

tests/integration/_support/wire.py

@@ -15,6 +15,8 @@ from queue import SimpleQueue
 from types import MappingProxyType
 from typing import TYPE_CHECKING, BinaryIO, Final
 
+from tests.integration._support.ports import allocate_port
+
 if TYPE_CHECKING:
     from collections.abc import Buffer
 
@@ -334,7 +336,7 @@ def wire_server(
             if tls is not None:
                 self.socket = tls.wrap_socket(self.socket, server_side=True)
 
-    with OwnedHTTPServer(("127.0.0.1", port), Handler) as server:
+    with OwnedHTTPServer(("127.0.0.1", port or allocate_port()), Handler) as server:
         thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": _POLL_SECONDS})
         thread.start()
         try:

tests/unit/integration_support/test_wire.py

@@ -7,7 +7,7 @@ import ssl
 import time
 from collections.abc import Callable, Iterator
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import ExitStack
+from contextlib import ExitStack, closing
 from http.client import HTTPConnection, HTTPResponse, HTTPSConnection
 from pathlib import Path
 from typing import Final
@@ -19,6 +19,7 @@ from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.x509.oid import NameOID
 
+from tests.integration._support.ports import FIRST_PORT, LAST_PORT, allocate_port
 from tests.integration._support.wire import Reply, Request, wire_server
 
 PAUSE_SECONDS: Final = 6
@@ -28,6 +29,24 @@ LATE: Final = b'{"call": "late"}'
 UNREAD: Final = b"x" * (32 * 1024 * 1024)
 
 
+@pytest.mark.parametrize("port", (None, 0), ids=("default", "zero"))
+def test_default_listener_uses_the_harness_port_range(port: int | None) -> None:
+    with wire_server(_echo, **({} if port is None else {"port": port})) as wire:
+        bound_port: Final = urlsplit(wire.url).port
+        assert bound_port is not None and FIRST_PORT <= bound_port <= LAST_PORT
+        with closing(_client(wire.url)) as connection:
+            connection.request("POST", "/echo", FIRST)
+            response: Final = connection.getresponse()
+            assert (response.status, response.read()) == (200, FIRST)
+
+
+def test_explicit_listener_port_is_preserved() -> None:
+    port: Final = allocate_port()
+    with wire_server(_echo, port=port) as wire:
+        assert urlsplit(wire.url).port == port
+        assert wire.drain() == ()
+
+
 def _self_signed_server_context(directory: Path) -> ssl.SSLContext:
     key: Final = ec.generate_private_key(ec.SECP256R1())
     name: Final = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "wire peer")])