diff --git a/tests/integration/README.md b/tests/integration/README.md index c5d52922de6..a1d224749b3 100644 --- a/tests/integration/README.md +++ b/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 diff --git a/tests/integration/_support/asgi.py b/tests/integration/_support/asgi.py index 30eb9f21760..4dcf3614836 100644 --- a/tests/integration/_support/asgi.py +++ b/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( diff --git a/tests/integration/_support/async_server.py b/tests/integration/_support/async_server.py index 53dcbc8d92d..73851b1b4a6 100644 --- a/tests/integration/_support/async_server.py +++ b/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 diff --git a/tests/integration/_support/aws_control_plane.py b/tests/integration/_support/aws_control_plane.py index 8cef78f8bee..1c984cdf03e 100644 --- a/tests/integration/_support/aws_control_plane.py +++ b/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[^/]+)/(?P[^,]+), " r"SignedHeaders=(?P[^,]+), Signature=(?P[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: diff --git a/tests/integration/_support/device_login.py b/tests/integration/_support/device_login.py index 13a7169a01a..349021a99e9 100644 --- a/tests/integration/_support/device_login.py +++ b/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() diff --git a/tests/integration/_support/forward_proxy.py b/tests/integration/_support/forward_proxy.py index 0d92c3d0719..98c440c3f6c 100644 --- a/tests/integration/_support/forward_proxy.py +++ b/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: diff --git a/tests/integration/_support/idle_gateway.py b/tests/integration/_support/idle_gateway.py index 5a86d7e0a35..42cbf385174 100644 --- a/tests/integration/_support/idle_gateway.py +++ b/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: diff --git a/tests/integration/_support/mail.py b/tests/integration/_support/mail.py index 3894baeccc3..08562d9a8c3 100644 --- a/tests/integration/_support/mail.py +++ b/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() diff --git a/tests/integration/_support/responses_ws_peer.py b/tests/integration/_support/responses_ws_peer.py index 1bf3e1fcbae..fb02c10f39d 100644 --- a/tests/integration/_support/responses_ws_peer.py +++ b/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() diff --git a/tests/integration/_support/riva.py b/tests/integration/_support/riva.py index 438d877386b..f50c28b705d 100644 --- a/tests/integration/_support/riva.py +++ b/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) diff --git a/tests/integration/_support/wire.py b/tests/integration/_support/wire.py index d4cbccfc43e..343430db4a6 100644 --- a/tests/integration/_support/wire.py +++ b/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: diff --git a/tests/unit/integration_support/test_wire.py b/tests/unit/integration_support/test_wire.py index c786895720f..7dd168eb212 100644 --- a/tests/unit/integration_support/test_wire.py +++ b/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")])