scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-16 04:29:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 138

test(integration): hand every harness server a port from one per-worker block (#46761)

Browse source 
* test(integration): hand every harness server a port from one per-worker block

About 45 places in the integration harness picked a port by binding port 0,
reading the number back, closing the socket and starting a server on that
number later. Another process could take the port in between, which surfaced
as "Address already in use" flakes. Earlier fixes retried around the race in
single call sites without removing it.

tests/integration/_support/ports.py now owns every port the harness picks
ahead of a server. Each pytest process claims a 200-port block between 10000
and 32767 by taking an exclusive flock on a per-block lock file in the system
temp directory, so parallel runs on one machine never share a block and the
lock goes away with the process. The range is above the fixed service ports
and below the Linux and macOS ephemeral ranges. Ports go out in order, are
never reused in a run, skip anything an outside process already holds, and
a process that exhausts its block claims the next free one.

The three _free_port copies, worker_endpoints' port-0 binds,
free_loopback_port, the owned proxy's lost-port-race retry and the owned
Redis retry loop are gone. Tests that restart a server on the same port now
allocate that port up front instead of reading it back from the first
server. TQ012 in scripts/check_test_quality.py rejects a port-0 bind read
back with getsockname anywhere under tests/integration.

* ci(integration): record listening sockets and the ephemeral port range

* Revert "ci(integration): record listening sockets and the ephemeral port range"

This reverts commit d3fafc3da2.
This commit is contained in:
yuneng-jiang 2026-10-15 16:56:40 -07:00 • committed by GitHub
parent 1e5d83f6ce
commit b602209ca6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
67 changed files with 360 additions and 575 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
scripts/check_test_quality.py
View file

@ -88,6 +88,13 @@ TQ011 An integration-test assertion that compares an elapsed time against a nu
is a literal when it is built only from numbers, or names a module-level number.
Only `assert` statements are examined, so a polling deadline in a `while` loop is
left alone. Same path scope as TQ010.
TQ012 A port read back from a port-0 bind: `sock.bind((host, 0))` in a function
that also calls `getsockname`. The number is free only until the socket
closes, so a server started on it later can lose it to another process and
fail with "Address already in use". Take the port from `allocate_port()` in
tests/integration/_support/ports.py, which hands each pytest worker its own
block of ports. Every file under tests/integration is examined, including
tests/integration/_support.
Every rule is suppressible with `# test-quality-ok: <reason>` on the reported
line, following the repo's `*-ok: <reason>` convention. A suppression without a
@ -172,7 +179,7 @@ SUBPROCESS_SPAWNS: Final = frozenset(("run", "Popen", "check_output", "check_cal
INTERPRETER_ISOLATION_FLAGS: Final = frozenset(("-I", "-P"))
RULE_CODES: Final = frozenset((
"TQ000", "TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006", "TQ007", "TQ009", "TQ010", "TQ011",
"TQ000", "TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006", "TQ007", "TQ009", "TQ010", "TQ011", "TQ012",
))
INTEGRATION_ROOT: Final = ("tests", "integration")
@ -181,6 +188,8 @@ SIGNAL_SENDERS: Final = frozenset(("os.kill", "os.killpg"))
PROCESS_CONTROL_SIGNALS: Final = frozenset(("SIGKILL", "SIGSTOP", "SIGCONT"))
ZERO_ARGUMENT_PROCESS_CONTROLS: Final = frozenset(("kill", "terminate", "suspend", "resume"))
SIGNAL_METHOD: Final = "send_signal"
BIND_METHOD: Final = "bind"
PORT_READBACK: Final = "getsockname"
CLOCK_READERS: Final = frozenset((
"time.monotonic", "time.perf_counter", "time.time",
"time.monotonic_ns", "time.perf_counter_ns", "time.time_ns",
@ -877,6 +886,50 @@ def iter_wall_clock_bound_violations(path: Path, tree: ast.Module) -> Iterator[V
)
def is_integration_tree_path(path: Path) -> bool:
parts: Final = path.parts
return any(parts[index:index + 2] == INTEGRATION_ROOT for index in range(len(parts) - 1))
def _calls_in_scope(node: ast.AST, scope: ast.AST) -> Iterator[tuple[ast.Call, ast.AST]]:
for child in ast.iter_child_nodes(node):
if isinstance(child, ast.Call):
yield child, scope
inner: Final = child if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)) else scope
yield from _calls_in_scope(child, inner)
def _binds_port_zero(call: ast.Call) -> bool:
if not isinstance(call.func, ast.Attribute) or call.func.attr != BIND_METHOD or not call.args:
return False
address: Final = call.args[0]
return (
isinstance(address, ast.Tuple)
and len(address.elts) == 2
and isinstance(address.elts[1], ast.Constant)
and address.elts[1].value == 0
)
def _reads_back_port(scope: ast.AST) -> bool:
return any(isinstance(node, ast.Attribute) and node.attr == PORT_READBACK for node in ast.walk(scope))
def iter_port_readback_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
if not is_integration_tree_path(path):
return
for call, scope in _calls_in_scope(tree, tree):
if _binds_port_zero(call) and _reads_back_port(scope):
yield Violation(
path,
call.lineno,
"TQ012",
"a port read back from a port-0 bind is free only until the socket closes, so a server started on "
"it later can lose it to another process; take the port from allocate_port() in "
f"tests/integration/_support/ports.py (suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
)
def iter_conftest_inventory_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
if path.name != CONFTEST_NAME:
return
@ -916,6 +969,7 @@ def check_file(path: Path) -> tuple[Violation, ...]:
*iter_child_interpreter_violations(path, tree),
*iter_process_signal_violations(path, tree),
*iter_wall_clock_bound_violations(path, tree),
*iter_port_readback_violations(path, tree),
)
if violation.line not in skip
)

2
tests/integration/README.md
View file

@ -28,6 +28,8 @@ 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`. 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`
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
There is no hand-maintained list of test cases. A positional argument is a file of the group or a pytest node id inside one (`path::test[param]`), so one cell of a parametrized file can run alone. The runner fails only when pytest fails, when collection errors, or when a selected file collects zero tests. Older tests still carry `@pytest.mark.covers(...)` decorators; the marker stays registered so they collect, but the IDs are not checked against anything and new tests should not use it. The GitHub Actions coverage census reads the `GROUPS` literal in `run.py` and treats every `tests/integration/<directory>/test_*.py` file in a scheduled group as owned by CircleCI

18
tests/integration/_support/database_relay.py
View file

@ -1,5 +1,4 @@
import asyncio
import socket
import threading
import time
from collections.abc import Generator
@ -7,23 +6,16 @@ from contextlib import contextmanager
from typing import Final
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from pydantic import TypeAdapter
from tests.integration._support.async_server import LoopbackServer, close_writer
from tests.integration._support.ports import allocate_port
PORT: Final = TypeAdapter(int)
OUTAGE_SECONDS: Final = 10.0
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return PORT.validate_python(reserve.getsockname()[1])
class DatabaseRelay:
def __init__(self, upstream_host: str, upstream_port: int, trigger: bytes) -> None:
self.port: Final = _free_port()
self.port: Final = allocate_port()
self._upstream_host: Final = upstream_host
self._upstream_port: Final = upstream_port
self._trigger: Final = trigger
@ -83,7 +75,7 @@ class DatabaseRelay:
class HeldStatementRelay:
def __init__(self, upstream_host: str, upstream_port: int, trigger: bytes) -> None:
self.port: Final = _free_port()
self.port: Final = allocate_port()
self._upstream_host: Final = upstream_host
self._upstream_port: Final = upstream_port
self._trigger: Final = trigger
@ -144,7 +136,7 @@ class TriggerScanner:
class DroppedConnectionRelay:
def __init__(self, upstream_host: str, upstream_port: int, trigger: bytes) -> None:
self.port: Final = _free_port()
self.port: Final = allocate_port()
self._upstream_host: Final = upstream_host
self._upstream_port: Final = upstream_port
self._trigger: Final = trigger
@ -231,7 +223,7 @@ class FrontendStatementScanner:
class StatementCountingRelay:
def __init__(self, upstream_host: str, upstream_port: int) -> None:
self.port: Final = _free_port()
self.port: Final = allocate_port()
self._upstream_host: Final = upstream_host
self._upstream_port: Final = upstream_port
self._loop: Final = asyncio.new_event_loop()

19
tests/integration/_support/otlp_sink.py
View file

@ -13,7 +13,6 @@ import datetime
import json
import os
import signal
import socket
import ssl
import subprocess
import sys
@ -35,6 +34,8 @@ import yaml
from pydantic import JsonValue, TypeAdapter
from typing_extensions import ReadOnly, TypedDict
from integration._support.ports import allocate_port
INTERNAL_MARKERS: Final = ("gen_ai.operation.name", "mcp.method.name", "litellm.guardrail_name")
@ -421,12 +422,6 @@ class ConnectSink:
ca_pem: str
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return int(reserve.getsockname()[1])
def _pid_reachable(url: str) -> bool:
try:
return httpx.get(f"{url}/__pid", trust_env=False, timeout=2).status_code == 200
@ -439,7 +434,7 @@ def owned_sinks(directory: Path) -> Iterator[SpanSinks]:
from integration._support.process import group_members, signal_group, stop_root_process
directory.mkdir(parents=True, exist_ok=True)
ports: Final = tuple(_free_port() for _ in range(3))
ports: Final = tuple(allocate_port() for _ in range(3))
root: Final = Path(__file__).resolve().parents[3]
with ExitStack() as stack:
processes: Final = tuple(
@ -509,8 +504,8 @@ def _await_sink(url: str) -> None:
@contextmanager
def owned_grpc_sink(directory: Path) -> Iterator[GrpcSink]:
http_port: Final = _free_port()
grpc_port: Final = _free_port()
http_port: Final = allocate_port()
grpc_port: Final = allocate_port()
with _spawn_sink(directory, "otlp-grpc-sink.log", ["--port", str(http_port), "--grpc-port", str(grpc_port)]):
control_url: Final = f"http://127.0.0.1:{http_port}"
_await_sink(control_url)
@ -519,8 +514,8 @@ def owned_grpc_sink(directory: Path) -> Iterator[GrpcSink]:
@contextmanager
def owned_connect_sink(directory: Path) -> Iterator[ConnectSink]:
http_port: Final = _free_port()
tunnel_port: Final = _free_port()
http_port: Final = allocate_port()
tunnel_port: Final = allocate_port()
ca_dir: Final = directory / "mitm"
with _spawn_sink(
directory,

61
tests/integration/_support/ports.py Normal file
View file

@ -0,0 +1,61 @@
import fcntl
import os
import socket
import tempfile
import threading
from collections.abc import Iterator
from pathlib import Path
from typing import Final
FIRST_PORT: Final = 10_000
LAST_PORT: Final = 32_767
BLOCK_SIZE: Final = 200
LOCK_DIRECTORY: Final = Path(tempfile.gettempdir()) / "litellm-integration-ports"
def _blocks() -> Iterator[range]:
return (range(start, start + BLOCK_SIZE) for start in range(FIRST_PORT, LAST_PORT + 2 - BLOCK_SIZE, BLOCK_SIZE))
def _claim_until_exit(block: range, directory: Path) -> bool:
descriptor: Final = os.open(directory / f"{block.start}.lock", os.O_RDWR | os.O_CREAT, 0o600)
try:
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
os.close(descriptor)
return False
return True
def _bindable(port: int) -> bool:
with socket.socket() as probe:
try:
probe.bind(("127.0.0.1", port))
except OSError:
return False
return True
def _free_ports(directory: Path) -> Iterator[int]:
directory.mkdir(parents=True, exist_ok=True)
for block in _blocks():
if _claim_until_exit(block, directory):
yield from filter(_bindable, block)
class PortAllocator:
def __init__(self, directory: Path) -> None:
self._directory: Final = directory
self._guard: Final = threading.Lock()
self._ports: Final = _free_ports(directory)
def allocate(self) -> int:
with self._guard:
port: Final = next(self._ports, None)
assert port is not None, (
f"Every integration port block in {FIRST_PORT}-{LAST_PORT} is claimed (locks in {self._directory})"
)
return port
allocate_port: Final = PortAllocator(LOCK_DIRECTORY).allocate

77
tests/integration/_support/process.py
View file

@ -1,10 +1,8 @@
import errno
import json
import threading
import os
import re
import signal
import socket
import subprocess
import sys
import time
@ -33,6 +31,7 @@ from pydantic import TypeAdapter
from tests.integration._support.budget import import_seconds, readiness_for, readiness_seconds
from tests.integration._support.client import GATEWAY_LIMITS, Gateway
from tests.integration._support.ports import allocate_port
DB_PUSH: Final = ("--use_prisma_db_push",)
MIGRATE_DEPLOY: Final = ()
@ -406,16 +405,6 @@ def _stop(process: subprocess.Popen[bytes]) -> None:
raise
_PORT_ATTEMPTS: Final = 3
_BIND_COLLISION: Final = os.strerror(errno.EADDRINUSE).lower()
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return reserve.getsockname()[1]
@dataclass(frozen=True, slots=True)
class _Launch:
process: subprocess.Popen[bytes]
@ -458,7 +447,7 @@ def _stop_launch(launch: _Launch) -> None:
def _launch(command: tuple[str, ...], root: Path, environment: Mapping[str, str], output: Path) -> _Launch:
port: Final = _free_port()
port: Final = allocate_port()
log_path: Final = output / f"owned-proxy-{uuid.uuid4().hex}.log"
started: Final = time.monotonic()
with log_path.open("w") as log:
@ -473,10 +462,6 @@ def _launch(command: tuple[str, ...], root: Path, environment: Mapping[str, str]
return _Launch(process, port, log_path, started)
def _lost_port_race(exit_code: int | None, log: Path) -> bool:
return exit_code is not None and _BIND_COLLISION in log.read_text().lower()
def _owns_listener(launch: _Launch) -> bool:
for member in group_members(launch.process.pid):
try:
@ -600,42 +585,36 @@ def _wait_until_ready(launch: _Launch, workers: int = 1) -> BootPhases | None:
)
def _launch_until_bound(
command: tuple[str, ...], root: Path, environment: Mapping[str, str], output: Path, attempts: int, workers: int
def _launch_until_ready(
command: tuple[str, ...], root: Path, environment: Mapping[str, str], output: Path, workers: int
) -> _Launch:
launch: Final = _launch(command, root, environment, output)
try:
phases: Final = _wait_until_ready(launch, workers)
exit_code: Final = launch.process.poll()
assert exit_code is None or (attempts > 1 and _lost_port_race(exit_code, launch.log)), (
"Owned proxy exited before readiness"
)
assert launch.process.poll() is None, "Owned proxy exited before readiness"
except BaseException:
_stop_launch(launch)
raise
if exit_code is None:
assert phases is not None
launch.log.with_suffix(".startup.json").write_text(
json.dumps(
{
"root_pid": launch.process.pid,
"readiness_seconds": time.monotonic() - launch.started,
"logical_cpus": os.cpu_count(),
"workers": workers,
"spawned_at": _wall_clock(launch.started),
"phases": {
"prisma_started": phases.prisma_started,
"prisma_finished": phases.prisma_finished,
"listening": phases.listening,
"first_ready": phases.first_ready,
},
}
)
+ "\n"
assert phases is not None
launch.log.with_suffix(".startup.json").write_text(
json.dumps(
{
"root_pid": launch.process.pid,
"readiness_seconds": time.monotonic() - launch.started,
"logical_cpus": os.cpu_count(),
"workers": workers,
"spawned_at": _wall_clock(launch.started),
"phases": {
"prisma_started": phases.prisma_started,
"prisma_finished": phases.prisma_finished,
"listening": phases.listening,
"first_ready": phases.first_ready,
},
}
)
return launch
_stop_launch(launch)
return _launch_until_bound(command, root, environment, output, attempts - 1, workers)
+ "\n"
)
return launch
def _proxy_root() -> Path:
@ -785,7 +764,7 @@ def owned_proxy_process(
*database_setup,
*extra_arguments,
)
launch: Final = _launch_until_bound(command, root, environment, output, _PORT_ATTEMPTS, workers)
launch: Final = _launch_until_ready(command, root, environment, output, workers)
process: Final = launch.process
try:
with httpx.Client(
@ -825,7 +804,7 @@ def configless_proxy_process(
str(healthcheck_seconds(WorkerHealthcheck.STRICT)),
*DB_PUSH,
)
launch: Final = _launch_until_bound(command, _proxy_root(), environment, output, _PORT_ATTEMPTS, workers)
launch: Final = _launch_until_ready(command, _proxy_root(), environment, output, workers)
try:
with httpx.Client(
base_url=f"http://127.0.0.1:{launch.port}", timeout=15, trust_env=False, limits=GATEWAY_LIMITS
@ -858,7 +837,7 @@ def owned_gateway_image(
"--timeout-worker-healthcheck",
str(int(graceful_stop_seconds())),
)
launch: Final = _launch_until_bound(command, root, environment, output, _PORT_ATTEMPTS, workers)
launch: Final = _launch_until_ready(command, root, environment, output, workers)
try:
with httpx.Client(
base_url=f"http://127.0.0.1:{launch.port}", timeout=15, trust_env=False, limits=GATEWAY_LIMITS
@ -1040,7 +1019,7 @@ def owned_upstream(
directory: Path, *, tls: bool = False, port: int | None = None, inherited: Mapping[str, str] = os.environ
) -> Generator[UpstreamSlot]:
certificate: Final = self_signed_certificate(directory) if tls else None
slot: Final = UpstreamSlot(directory, port or _free_port(), _harness_root(), certificate, inherited)
slot: Final = UpstreamSlot(directory, port or allocate_port(), _harness_root(), certificate, inherited)
try:
slot.start()
yield slot

106
tests/integration/_support/redis_process.py
View file

@ -1,7 +1,6 @@
import os
import shutil
import signal
import socket
import subprocess
import time
import uuid
@ -14,18 +13,7 @@ from typing import Final, TextIO
from redis import Redis
from redis.exceptions import ConnectionError as RedisConnectionError
PORT_ATTEMPTS: Final = 5
@dataclass(frozen=True, slots=True)
class PortUnavailable:
reason: str
def free_loopback_port() -> int:
with socket.socket() as reservation:
reservation.bind(("127.0.0.1", 0))
return reservation.getsockname()[1]
from tests.integration._support.ports import allocate_port
@dataclass
@ -60,13 +48,8 @@ class OwnedRedis:
return int(result.stdout)
def start(self) -> None:
unavailable: Final = self.try_start()
assert unavailable is None, unavailable.reason
def try_start(self) -> PortUnavailable | None:
assert self.process is None
if self.serves():
return PortUnavailable(f"Another Redis already serves {self.host}:{self.port} before the owned one starts")
assert not self.serves(), f"Another Redis already serves {self.host}:{self.port} before the owned one starts"
process: Final = subprocess.Popen(
self.command, stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True
)
@ -77,7 +60,7 @@ class OwnedRedis:
code: Final = process.poll()
if code is not None:
self.process = None
return PortUnavailable(
raise AssertionError(
f"Owned Redis exited with code {code} before readiness on {self.host}:{self.port}"
)
try:
@ -86,13 +69,13 @@ class OwnedRedis:
expected: Final = self.owned_pid()
if actual == expected:
self.server_pid = actual
return None
return
try:
process.wait(timeout=max(deadline - time.monotonic(), 0))
except subprocess.TimeoutExpired as error:
raise AssertionError("Redis readiness reached a different process") from error
self.process = None
return PortUnavailable(f"A different Redis answered on {self.host}:{self.port} during startup")
raise AssertionError(f"A different Redis answered on {self.host}:{self.port} during startup")
except RedisConnectionError:
pass
assert time.monotonic() < deadline, "Owned Redis readiness deadline exceeded"
@ -148,33 +131,10 @@ class OwnedRedis:
self.signal(signal.SIGCONT)
@contextmanager
def _started(
build: Callable[[int], OwnedRedis],
choose_port: Callable[[], int],
attempts: int,
log_name: str,
) -> Iterator[OwnedRedis]:
server: Final = build(choose_port())
try:
unavailable: Final = server.try_start()
if unavailable is None:
yield server
return
assert attempts > 1, (
f"Owned Redis found no usable port after {PORT_ATTEMPTS} attempts: {unavailable.reason} (log {log_name})"
)
with _started(build, choose_port, attempts - 1, log_name) as started:
yield started
finally:
if server.process is not None:
server.stop()
@contextmanager
def owned_redis(
directory: Path,
choose_port: Callable[[], int] = free_loopback_port,
choose_port: Callable[[], int] = allocate_port,
inherited: Mapping[str, str] = os.environ,
) -> Iterator[OwnedRedis]:
binary: Final = shutil.which("redis-server")
@ -198,31 +158,31 @@ def owned_redis(
output.mkdir(parents=True, exist_ok=True)
directory.mkdir(parents=True, exist_ok=True)
with (output / f"owned-redis-{uuid.uuid4().hex}.log").open("w") as log:
def build(port: int) -> OwnedRedis:
pid_file: Final = (
str(directory / f"owned-redis-{port}.pid")
if binary
else f"/tmp/integration-redis-{uuid.uuid4().hex}.pid"
)
command: Final = (
*prefix,
"--port",
str(port),
"--set-proc-title",
"no",
"--pidfile",
pid_file,
"--bind",
"0.0.0.0" if not binary else "127.0.0.1",
"--protected-mode",
"no",
"--save",
"",
"--appendonly",
"no",
)
return OwnedRedis(host, port, command, log, pid_file)
with _started(build, choose_port, PORT_ATTEMPTS, log.name) as server:
port: Final = choose_port()
pid_file: Final = (
str(directory / f"owned-redis-{port}.pid") if binary else f"/tmp/integration-redis-{uuid.uuid4().hex}.pid"
)
command: Final = (
*prefix,
"--port",
str(port),
"--set-proc-title",
"no",
"--pidfile",
pid_file,
"--bind",
"0.0.0.0" if not binary else "127.0.0.1",
"--protected-mode",
"no",
"--save",
"",
"--appendonly",
"no",
)
server: Final = OwnedRedis(host, port, command, log, pid_file)
try:
server.start()
yield server
finally:
if server.process is not None:
server.stop()

17
tests/integration/_support/workers.py
View file

@ -1,7 +1,6 @@
from __future__ import annotations
import os
import socket
import time
import uuid
from collections.abc import Callable, Iterator, Mapping
@ -23,6 +22,7 @@ from redis.exceptions import TimeoutError as RedisTimeoutError
from tests.integration._support.client import Gateway
from tests.integration._support.database import scratch_database
from tests.integration._support.ports import allocate_port
from tests.integration._support.redis_process import owned_redis
from tests.integration._support.routing import WORKER_DATABASE_PREFIX
@ -78,15 +78,12 @@ def wait_for_services(upstream: str, proxies: tuple[Gateway, ...], cache: Redis)
def worker_endpoints() -> Mapping[str, str]:
with socket.socket() as upstream, socket.socket() as provider:
upstream.bind(("127.0.0.1", 0))
provider.bind(("127.0.0.1", 0))
return MappingProxyType(
{
"INTEGRATION_UPSTREAM_URL": f"http://127.0.0.1:{upstream.getsockname()[1]}",
"INTEGRATION_PROVIDER_PORT": str(provider.getsockname()[1]),
}
)
return MappingProxyType(
{
"INTEGRATION_UPSTREAM_URL": f"http://127.0.0.1:{allocate_port()}",
"INTEGRATION_PROVIDER_PORT": str(allocate_port()),
}
)
def _replica_urls(database: str) -> Mapping[str, str]:

9
tests/integration/authorization/test_key_guardrail_opt_out_chaos.py
View file

@ -1,5 +1,4 @@
import json
import socket
import threading
import time
import uuid
@ -15,6 +14,7 @@ from integration._support.client import Gateway, eventually, string_value
from integration._support.database import read_rows
from integration._support.process import owned_proxy, owned_proxy_process
from integration._support.timing import assert_cut_short, timed
from integration._support.ports import allocate_port
from integration.authorization._guardrail_opt_out import (
chat,
guardrail_config,
@ -34,14 +34,9 @@ class _GuardrailSink:
self._action: Final = action
self._server: ThreadingHTTPServer | None = None
self._thread: threading.Thread | None = None
self._port: Final = self._claim_port()
self._port: Final = allocate_port()
self.start()
def _claim_port(self) -> int:
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
return probe.getsockname()[1]
@property
def url(self) -> str:
return f"http://127.0.0.1:{self._port}"

6
tests/integration/authorization/test_responses_compact_id_ownership_chaos.py
View file

@ -15,6 +15,7 @@ import pytest
from integration._support.client import Gateway, eventually
from integration._support.process import graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Wire, wire_server
from integration._support.ports import allocate_port
from integration.authorization._responses_compact import (
ALIASES,
OTHER_TEAM_DETAIL,
@ -158,9 +159,9 @@ def _settled(target: _Target) -> bool:
def test_upstream_outage_mid_burst_keeps_refusals_and_lands_each_owner_compact_once(gateway: Gateway) -> None:
upstream: Final = CompactUpstream()
created: Final = marker("mint")
port: Final = allocate_port()
with gateway.scenario() as scenario:
with wire_server(upstream.respond) as first:
port: Final = urlsplit(first.url).port
with wire_server(upstream.respond, port=port) as first:
actors: Final = tenants(scenario, first)
target: Final = _Target(gateway, actors, mint(gateway, actors.owner, actors.model, created), vendor_id())
upstreamed(first)
@ -168,7 +169,6 @@ def test_upstream_outage_mid_burst_keeps_refusals_and_lands_each_owner_compact_o
before_received: Final = upstreamed(first)
during: Final = _send_all(target, _calls("during", 16))
alive: Final = gateway.request("GET", "/health/liveliness")
assert port is not None
with wire_server(upstream.respond, port=port) as second:
eventually(lambda: _settled(target), lambda settled: settled, seconds=60)
after: Final = _send_all(target, _calls("after", 16))

10
tests/integration/compatibility/test_a2a_protocol_version_override.py
View file

@ -10,13 +10,13 @@ from __future__ import annotations
import json
import os
import socket
import time
from typing import Final
import pytest
from integration._support.client import Gateway, eventually
from integration._support.database import read_rows
from integration._support.ports import allocate_port
from integration.compatibility._a2a_peers import (
HINT,
KIND_ERROR,
@ -284,14 +284,8 @@ def test_registered_agent_persists_protocol_version(gateway: Gateway) -> None:
_assert_legacy_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type]
def _closed_local_port() -> int:
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def test_pinned_agent_survives_localhost_retry(gateway: Gateway) -> None:
dead: Final = f"http://localhost:{_closed_local_port()}/a2a/unreachable"
dead: Final = f"http://localhost:{allocate_port()}/a2a/unreachable"
with peer("lgretry", "langgraph", card_url=dead) as agent, gateway.scenario() as scenario:
identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, PINNED)
_send_ok(gateway, identity, agent.marker)

12
tests/integration/compatibility/test_granian_admission_saturation.py
View file

@ -1,6 +1,5 @@
import asyncio
import os
import socket
import subprocess
import sys
import time
@ -12,22 +11,17 @@ import pytest
from tests._master_key import MASTER_KEY
from tests.integration._support.process import stop_child
from tests.integration._support.ports import allocate_port
MAX_IN_FLIGHT: Final = 8
MAX_QUEUED: Final = 8
REQUESTS: Final = 200
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
listener.bind(("127.0.0.1", 0))
return int(listener.getsockname()[1])
@pytest.mark.asyncio
async def test_granian_admission_control_saturation(tmp_path: Path) -> None:
fake_port: Final = _free_port()
proxy_port: Final = _free_port()
fake_port: Final = allocate_port()
proxy_port: Final = allocate_port()
fake_script: Final = Path(__file__).parents[2] / "_fake_openai_endpoint_server.py"
config_path: Final = tmp_path / "saturation_config.yaml"
proxy_env: Final = {**{k: v for k, v in os.environ.items() if k != "DATABASE_URL"}, "LITELLM_MASTER_KEY": MASTER_KEY}

6
tests/integration/compatibility/test_missing_body_param_status.py
View file

@ -3,7 +3,6 @@ from __future__ import annotations
import asyncio
import json
import os
import socket
import subprocess
import sys
import uuid
@ -24,6 +23,7 @@ from integration._support.database import read_rows
from integration._support.process import owned_proxy_process, paused, stop_child
from integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from openai import APIStatusError, AsyncOpenAI, BadRequestError, OpenAI
from openai.types import Video
from openai.types.eval_create_response import EvalCreateResponse as _Eval
@ -1504,9 +1504,7 @@ def _healthy(url: str) -> int:
@contextmanager
def _upstream(directory: Path) -> Iterator[tuple[subprocess.Popen[bytes], str]]:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
port: Final = int(reserve.getsockname()[1])
port: Final = allocate_port()
root: Final = Path(__file__).resolve().parents[2]
url: Final = f"http://127.0.0.1:{port}"
with (directory / "upstream.log").open("w") as log:

8
tests/integration/database/test_harness_lifecycle.py
View file

@ -44,6 +44,7 @@ from tests.integration._support.database_relay import (
from tests.integration._support.runtime import configure_executor
from tests.integration._support.sharding import ShardExecution, shard_errors
from tests.integration._support.wire import Reply, wire_server
from tests.integration._support.ports import allocate_port
from tests.integration.run import _without_provider_api_keys
@ -143,10 +144,11 @@ async def _unused_peer(reader: asyncio.StreamReader, writer: asyncio.StreamWrite
def test_relay_bind_failure_closes_its_loop_and_allows_repeated_cleanup() -> None:
port: Final = allocate_port()
with socket.socket() as occupied:
occupied.bind(("127.0.0.1", 0))
occupied.bind(("127.0.0.1", port))
occupied.listen()
relay: Final = LoopbackServer(occupied.getsockname()[1], _unused_peer)
relay: Final = LoopbackServer(port, _unused_peer)
with pytest.raises(OSError, match=os.strerror(errno.EADDRINUSE).lower()):
relay.start()
relay.stop()
@ -156,7 +158,7 @@ def test_relay_bind_failure_closes_its_loop_and_allows_repeated_cleanup() -> Non
def test_upstream_readiness_failure_reaps_the_started_process(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setattr(process_support, "_UPSTREAM_READY_SECONDS", 0)
children_before: Final = frozenset(child.pid for child in psutil.Process().children(recursive=True))
slot: Final = process_support.UpstreamSlot(tmp_path, process_support._free_port(), process_support._harness_root())
slot: Final = process_support.UpstreamSlot(tmp_path, allocate_port(), process_support._harness_root())
try:
with pytest.raises(AssertionError, match="readiness deadline exceeded"):
slot.start()

6
tests/integration/database/test_v1_migration_error_logging.py
View file

@ -25,7 +25,7 @@ from psycopg import sql
from psycopg.types.json import Jsonb
from tests.integration._support.client import eventually
from tests.integration._support.process import _free_port as free_port
from tests.integration._support.ports import allocate_port
from tests.integration._support.process import stop_process_group
REPO_ROOT: Final = Path(__file__).resolve().parents[3]
@ -94,7 +94,7 @@ def _unreachable_database_url(database_url: str) -> str:
parsed.path.lstrip("/"),
encode_password=False,
)
return _replace_port(url, free_port())
return _replace_port(url, allocate_port())
@contextmanager
@ -388,7 +388,7 @@ def test_unreachable_database_recovers_after_postgres_forwarder_starts(
port: Final = admin.port
assert hostname is not None and port is not None
target_host: Final = "127.0.0.1" if hostname == "localhost" else hostname
forwarding_port: Final = free_port()
forwarding_port: Final = allocate_port()
forwarded_url: Final = _replace_port(database_url, forwarding_port, "127.0.0.1")
command, environment = _migration_invocation(forwarded_url, tmp_path, {})
output_path: Final = _migration_log_path(request.node.name, tmp_path)

6
tests/integration/management/test_vector_store_file_list_managed_ids.py
View file

@ -22,6 +22,7 @@ from integration._support.client import Gateway, Scenario, eventually, object_va
from integration._support.database import read_rows
from integration._support.process import owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from openai import APIStatusError, AsyncOpenAI, OpenAI
from pydantic import JsonValue, TypeAdapter
@ -1202,7 +1203,8 @@ def test_provider_outage_mid_burst_fails_loudly_and_mapping_resumes_after_recove
provider_a: Final = _provider_file_id(bearer, "a.txt")
respond: Final = _provider(store, _constant_listing(store, provider_a))
with gateway.scenario() as scenario:
with wire_server(respond) as wire:
port: Final = allocate_port()
with wire_server(respond, port=port) as wire:
model: Final = scenario.model(api_base=wire.url + "/v1", api_key=bearer)
_wait_until_every_worker_serves(gateway, model)
wire.drain()
@ -1212,8 +1214,6 @@ def test_provider_outage_mid_burst_fails_loudly_and_mapping_resumes_after_recove
served: Final = _burst(gateway, store, member.key, model, 40)
assert [_ids(_listed(response)) for response in served] == [(managed_a,)] * 40
assert sum(1 for request in wire.drain() if request.method == "GET") == 40
port: Final = urlsplit(wire.url).port
assert port is not None
failed: Final = _burst(gateway, store, member.key, model, 20)
assert [response.status_code for response in failed] == [500] * 20, [r.text for r in failed[:3]]
for response in failed:

10
tests/integration/mcp/test_proxy_mcp_e2e.py
View file

@ -45,6 +45,7 @@ from tests.integration._support.process import stop_child
from tests.integration._support.timing import assert_enforced_within
from tests.integration._support.wire import Reply, Wire, wire_server
from tests.integration._support.wire import Request as WireRequest
from tests.integration._support.ports import allocate_port
CONFIG_TEMPLATE_PATH = Path("tests/integration/mcp/_config_mcp_e2e.yaml")
MCP_SERVER_SCRIPT = Path("tests/integration/mcp/_math_server.py")
@ -294,8 +295,9 @@ def _start_proxy_server(
sock: Final = socket_factory()
startup.callback(sock.close)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(("127.0.0.1", 0))
host, port = sock.getsockname()
host: Final = "127.0.0.1"
port: Final = allocate_port()
sock.bind((host, port))
config = uvicorn.Config(proxy_app, host=host, port=port, log_level="warning", lifespan="off")
server = uvicorn.Server(config)
@ -369,9 +371,7 @@ def test_proxy_startup_failure_closes_acquired_resources(tmp_path: Path) -> None
@contextmanager
def _math_http_server(offset: int) -> typing.Iterator[str]:
host = "127.0.0.1"
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind((host, 0))
_, port = sock.getsockname()
port: Final = allocate_port()
with tempfile.TemporaryFile() as server_log:
process = subprocess.Popen(

6
tests/integration/observability/test_azure_content_safety_dispatch_resilience.py
View file

@ -1,5 +1,4 @@
import concurrent.futures
import socket
import threading
import uuid
from contextlib import ExitStack
@ -13,6 +12,7 @@ from integration._support.client import Gateway, Scenario, eventually
from integration._support.database import read_rows
from integration._support.redis_process import owned_redis
from integration._support.wire import Wire, wire_server
from integration._support.ports import allocate_port
from integration.observability.azure_dispatch_support import (
ATTACK_MARKER,
AzureBehavior,
@ -251,9 +251,7 @@ def test_x1_azure_server_stop_restart_during_mixed_burst(
with ExitStack() as resources:
redis: Final = resources.enter_context(owned_redis(tmp_path))
provider: Final = resources.enter_context(wire_server(provider_handler))
with socket.socket() as reservation:
reservation.bind(("127.0.0.1", 0))
azure_port: Final = int(reservation.getsockname()[1])
azure_port: Final = allocate_port()
first_azure_lifetime: Final = ExitStack()
try:
azure: Final = first_azure_lifetime.enter_context(wire_server(azure_handler(), port=azure_port))

7
tests/integration/observability/test_guardrail_effects.py
View file

@ -1,7 +1,6 @@
import asyncio
import json
import re
import socket
import threading
import uuid
from collections.abc import Callable, Iterator, Mapping
@ -22,6 +21,7 @@ from integration._support.database import read_rows
from integration._support.mcp import mcp_peer, register_mcp, tool_names
from integration._support.process import OwnedProxy, owned_proxy, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from openai import AsyncOpenAI, OpenAI
from pydantic import BaseModel, JsonValue
@ -1285,10 +1285,7 @@ def _blocked_stream_events(text: str) -> tuple[dict[str, object], ...]:
def _dead_api_base() -> str:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
port: Final = reserve.getsockname()[1]
return f"http://127.0.0.1:{port}/v1"
return f"http://127.0.0.1:{allocate_port()}/v1"
@pytest.mark.covers("other.observability.guardrails.responses_pre_call_denial_openai_sdk_streams_typed_message")

12
tests/integration/observability/test_guardrail_stream_scope_chaos.py
View file

@ -3,7 +3,6 @@ from __future__ import annotations
import json
import os
import shutil
import socket
import subprocess
import uuid
from collections.abc import Iterator, Mapping, Sequence
@ -23,6 +22,7 @@ from integration._support.client import Gateway, JsonValue, eventually
from integration._support.database import read_rows
from integration._support.process import OwnedProxy, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import TypeAdapter
ChaosEndpoint: TypeAlias = Literal["chat", "messages", "responses"]
@ -455,12 +455,6 @@ def _owned_proxy(
yield owned
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return int(reserve.getsockname()[1])
def _assert_down_wave(
plans: Sequence[CallPlan],
responses: Sequence[httpx.Response],
@ -485,7 +479,7 @@ def _assert_down_wave(
def test_h1_sink_outage_keeps_scope_isolated_through_recovery(rig: ChaosRig, tmp_path: Path) -> None:
port_a: Final = _free_port()
port_a: Final = allocate_port()
started: Final = Event()
unavailable: Final = Event()
release: Final = Event()
@ -763,7 +757,7 @@ def _owned_postgres(directory: Path) -> Iterator[PostgresCluster]:
)
assert docker_info.returncode == 0, docker_info.stderr
directory.mkdir(parents=True, exist_ok=True)
port: Final = _free_port()
port: Final = allocate_port()
container_name: Final = f"litellm-stream-scope-h5-{uuid.uuid4().hex}"
password: Final = uuid.uuid4().hex
created: Final = subprocess.run(

4
tests/integration/observability/test_langtrace_delivery.py
View file

@ -21,6 +21,7 @@ from integration._support.client import Gateway, eventually
from integration._support.process import OwnedProxy, owned_proxy, owned_proxy_process
from integration._support.timing import assert_not_delayed_by, timed
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from openai import AsyncOpenAI, OpenAI
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
from opentelemetry.proto.trace.v1.trace_pb2 import Span, Status
@ -595,8 +596,7 @@ def test_langtrace_two_workers_deliver_every_burst_span_exactly_once(gateway: Ga
def test_langtrace_sink_outage_mid_burst_recovers_on_the_same_port(gateway: Gateway, tmp_path: Path) -> None:
key: Final = "synthetic-langtrace-key-" + uuid.uuid4().hex
with wire_server(_accepted) as probe:
port: Final = int(probe.url.rsplit(":", 1)[1])
port: Final = allocate_port()
host: Final = f"http://127.0.0.1:{port}"
with wire_server(_upstream) as provider:
overrides: Final = {"LANGTRACE_API_KEY": key, "LANGTRACE_API_HOST": host, "OTEL_BSP_SCHEDULE_DELAY": "300"}

6
tests/integration/observability/test_logging_only_scope_chaos.py
View file

@ -1,6 +1,5 @@
from __future__ import annotations
import socket
import threading
import time
import uuid
@ -46,6 +45,7 @@ from integration._support.client import Gateway, eventually
from integration._support.process import owned_proxy, owned_proxy_process
from integration._support.timing import Timed, assert_not_delayed_by, timed
from integration._support.wire import Reply, Request
from integration._support.ports import allocate_port
from openai import APIConnectionError as OpenAIAPIConnectionError
from pydantic import JsonValue
@ -60,9 +60,7 @@ def test_K1_policy_edge_restart_mid_burst_keeps_output_observation_fail_open(gat
def policy(_request: Request) -> Reply:
return Reply(body=b'{"action":"NONE"}')
with socket.socket() as reservation:
reservation.bind(("127.0.0.1", 0))
policy_port: Final = reservation.getsockname()[1]
policy_port: Final = allocate_port()
with gateway.scenario() as scenario:
deployments: Final = _chaos_models(scenario, marker)
models: Final = tuple(deployment.model_name for deployment in deployments)

5
tests/integration/observability/test_passthrough_upstream_error_chaos.py
View file

@ -15,6 +15,7 @@ from integration._support.client import Gateway, eventually, object_value
from integration._support.database import read_rows
from integration._support.process import WorkerHealthcheck, graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
_GENERATE_CONTENT: Final[dict[str, JsonValue]] = {"contents": [{"role": "user", "parts": [{"text": "hi"}]}]}
@ -120,8 +121,8 @@ async def _fire_burst(
async def test_passthrough_upstream_outage_mid_burst_still_logs_errors_once(gateway: Gateway, tmp_path: Path) -> None:
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
path: Final = tmp_path / "chaos-outage.yaml"
with wire_server(_chaos_reply) as wire:
port: Final = int(wire.url.rsplit(":", 1)[1])
port: Final = allocate_port()
with wire_server(_chaos_reply, port=port) as wire:
config["environment_variables"] = {"GEMINI_API_BASE": wire.url, "GEMINI_API_KEY": "scripted"}
path.write_text(yaml.safe_dump(config))
with owned_proxy_process(gateway, tmp_path, {}, config=path, workers=2) as owned:

7
tests/integration/observability/test_straiker_v3_platform.py
View file

@ -10,7 +10,6 @@ from __future__ import annotations
import hashlib
import itertools
import json
import socket
import threading
import uuid
from collections.abc import Callable, Iterator
@ -30,6 +29,7 @@ from integration._support.client import Gateway, eventually, gateway_from_enviro
from integration._support.database import read_rows
from integration._support.process import OwnedProxy, WorkerHealthcheck, owned_proxy, owned_proxy_process
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
V3_KEY: Final = "sk_agt_synthetic_integration_key"
V1_KEY: Final = "synthetic-v1-collection-key"
@ -426,10 +426,7 @@ class Rig:
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
root: Final = tmp_path_factory.mktemp("straiker")
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
port: Final = reserve.getsockname()[1]
sink: Final = Sink(port)
sink: Final = Sink(allocate_port())
sink.start()
with gateway_from_environment() as gateway, wire_server(_provider) as provider:
config: Final = _rig_config(sink.url, root)

7
tests/integration/providers/_assistants_route_errors_support.py
View file

@ -3,7 +3,6 @@ from __future__ import annotations
import asyncio
import json
import re
import socket
import threading
import uuid
from collections.abc import AsyncIterator, Callable, Generator, Mapping
@ -84,12 +83,6 @@ def new_marker() -> str:
return f"m{uuid.uuid4().hex}"
def free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return reserve.getsockname()[1]
def proxy_path(route: Route, marker: str) -> str:
return route.path.format(marker=marker)

7
tests/integration/providers/_coralbricks.py
View file

@ -2,7 +2,6 @@ from __future__ import annotations
import json
import re
import socket
import uuid
from collections.abc import Callable, Mapping
from dataclasses import dataclass
@ -380,12 +379,6 @@ def raw_deployment(scenario: Scenario, model_name: str, litellm_params: Mapping[
return model_name
def closed_port() -> int:
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def provider_calls(wire: Wire) -> tuple[Request, ...]:
return tuple(request for request in wire.drain() if not is_readiness_probe(request))

7
tests/integration/providers/_vertex_batch_rows.py
View file

@ -1,5 +1,4 @@
import json
import socket
import threading
from collections.abc import Callable
from email.message import Message
@ -132,12 +131,6 @@ def stored_custom_ids(request: Request) -> tuple[str, ...]:
)
def closed_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def multipart_file(request: Request) -> bytes:
envelope: Final = f"content-type: {request.headers['content-type']}\r\n\r\n".encode() + request.body
parts: Final[tuple[Message, ...]] = tuple(BytesParser(policy=HTTP).parsebytes(envelope).iter_parts())

10
tests/integration/providers/test_anthropic_count_tokens_system_lift_wire.py
View file

@ -1,6 +1,5 @@
import asyncio
import json
import socket
import threading
import uuid
from collections.abc import Callable, Iterator, Mapping
@ -18,6 +17,7 @@ import pytest
from anthropic.types import MessageParam
from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from integration.providers._count_tokens_system_lift import (
ANTHROPIC_VERSION,
ASSISTANT,
@ -131,12 +131,6 @@ def _listening(deployment: _Deployment, count: Callable[[Request], Reply] = _cou
return wire_server(_peer(deployment.provider, count), port=deployment.port)
def _reserved_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return _CLIENT_ADDRESS.validate_python(reserve.getsockname())[1]
def _cmdline(process: psutil.Process) -> tuple[str, ...]:
try:
return _ARGUMENTS.validate_python(process.cmdline())
@ -211,7 +205,7 @@ def _settled_on_every_worker(gateway: Gateway, body: Mapping[str, JsonValue]) ->
def _deploy(gateway: Gateway, scenario: Scenario, provider: _Provider) -> _Deployment:
port: Final = _reserved_port()
port: Final = allocate_port()
model: Final = scenario.model(
model_info=None, model=f"{provider.prefix}/{_MODEL}", api_base=f"http://127.0.0.1:{port}", api_key=_API_KEY
)

12
tests/integration/providers/test_anthropic_refusal_stop_details_chaos.py
View file

@ -33,6 +33,7 @@ from integration._support.client import Gateway, eventually, held
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
from integration._support.timing import assert_not_delayed_by
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
pytestmark: Final = [
pytest.mark.timeout(2 * graceful_stop_seconds() + 120),
@ -180,12 +181,6 @@ def _open_upstream_connections(pid: int, upstream: str) -> int:
)
def _port(wire: Wire) -> int:
port: Final = urlsplit(wire.url).port
assert port is not None, wire.url
return port
def _accept(_request: Request) -> Reply:
return Reply()
@ -200,8 +195,9 @@ async def _timed_chat(gateway: Gateway) -> float:
async def test_sink_outage_mid_burst_delivers_every_refusal_event_once(gateway: Gateway, tmp_path: Path) -> None:
calls: Final = _calls(30)
markers: Final = frozenset(call.marker for call in calls)
sink_port: Final = allocate_port()
with wire_server(respond) as upstream, ExitStack() as first_sink:
sink: Final = first_sink.enter_context(wire_server(_accept))
sink: Final = first_sink.enter_context(wire_server(_accept, port=sink_port))
first: Final = Ledger(sink)
config: Final = _chaos_config(tmp_path, upstream, sink.url, retries=6)
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as owned:
@ -213,7 +209,7 @@ async def test_sink_outage_mid_burst_delivers_every_refusal_event_once(gateway:
first.events()
first_sink.close()
served_during: Final = await _burst(owned.gateway, calls[10:20])
with wire_server(_accept, port=_port(sink)) as restarted:
with wire_server(_accept, port=sink_port) as restarted:
second: Final = Ledger(restarted)
served_after: Final = await _burst(owned.gateway, calls[20:])
for item in (*served_first, *served_during, *served_after):

4
tests/integration/providers/test_assistants_route_errors_wire.py
View file

@ -14,6 +14,7 @@ import pytest
from integration._support.client import Gateway, eventually, gateway_from_environment
from integration._support.process import OwnedProxy, WorkerHealthcheck, graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request
from integration._support.ports import allocate_port
from integration.providers._assistants_route_errors_support import (
ASSISTANT_MODEL,
BODY_ROUTES,
@ -39,7 +40,6 @@ from integration.providers._assistants_route_errors_support import (
assert_reached_upstream_once,
assistants_wire,
error_reply,
free_port,
held_error_peer,
held_one_by_one,
held_upstream_connections,
@ -89,7 +89,7 @@ class _Scripted:
@pytest.fixture(scope="module")
def scripted(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Scripted]:
directory: Final = tmp_path_factory.mktemp("assistants-route-errors-scripted")
port: Final = free_port()
port: Final = allocate_port()
config: Final = openai_assistants_config(directory, port, _DEPLOYMENT_TIMEOUT_SECONDS)
with (
gateway_from_environment() as shared,

10
tests/integration/providers/test_bedrock_kimi_k3_cache_point_wire.py
View file

@ -3,7 +3,6 @@ import base64
import json
import os
import re
import socket
import threading
import time
import uuid
@ -27,6 +26,7 @@ from integration._support.process import OwnedProxy, owned_proxy_process
from integration._support.timing import assert_within_factor_of_control, timed
from integration._support.upstream import _aws_event_frame
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_if_encrypted_with
@ -1406,17 +1406,11 @@ def _assert_rows_once(served: tuple[_Served, ...]) -> None:
assert all(row["status"] == "success" for row in rows), rows
def _reserved_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return reserve.getsockname()[1]
@pytest.mark.timeout(900)
async def test_c01_a_peer_outage_mid_traffic_fails_cleanly_and_recovery_lands_every_id_once(
gateway: Gateway, tmp_path: Path
) -> None:
port: Final = _reserved_port()
port: Final = allocate_port()
prompt: Final = _prompt()
with wire_server(_peer(), port=port) as first_peer:
config: Final = _owned_config(first_peer, tmp_path, names=("kimi-us", "claude-control"))

5
tests/integration/providers/test_bedrock_mantle_context_overflow_wire.py
View file

@ -12,6 +12,7 @@ import pytest
from integration._support.client import Gateway, eventually, object_value
from integration._support.database import read_rows
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
_BACKEND: Final = "openai.gpt-5.6-luna"
@ -776,9 +777,9 @@ def test_messages_stream_overflow_logs_one_failure_row(gateway: Gateway) -> None
def test_chaos_mantle_peer_outage_mid_burst_logs_every_call_once_and_keeps_the_proxy_alive(gateway: Gateway) -> None:
port: Final = allocate_port()
with gateway.scenario() as scenario:
with wire_server(_chaos_peer) as wire:
port: Final = int(wire.url.rsplit(":", 1)[1])
with wire_server(_chaos_peer, port=port) as wire:
model: Final = scenario.model(model=_MODEL, api_base=wire.url, api_key=_API_KEY)
before: Final = _fire(gateway, _burst_bodies(model, "before"))
assert len(wire.drain()) == len(before)

12
tests/integration/providers/test_bedrock_mantle_count_tokens_wire.py
View file

@ -1,7 +1,6 @@
import asyncio
import json
import re
import socket
import threading
import uuid
from collections.abc import Callable, Iterator, Mapping, Sequence
@ -23,6 +22,7 @@ from integration._support.bedrock_runtime_peer import respond as runtime_generat
from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment, object_value
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from integration.providers._count_tokens_system_lift import (
ASSISTANT,
FOLLOW_UP,
@ -181,19 +181,13 @@ def _mantle_environment(port: int) -> Mapping[str, str]:
_INHERITED_BEARER: Final = ("AWS_BEARER_TOKEN_BEDROCK",)
def _reserved_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return reserve.getsockname()[1]
def _closed_port_url() -> str:
return f"http://127.0.0.1:{_reserved_port()}"
return f"http://127.0.0.1:{allocate_port()}"
@pytest.fixture(scope="module")
def mantle_port() -> int:
return _reserved_port()
return allocate_port()
@pytest.fixture(scope="module")

13
tests/integration/providers/test_bedrock_runtime_chat_completions_chaos.py
View file

@ -11,7 +11,7 @@ import socket
import threading
import uuid
from collections.abc import Callable, Iterator, Mapping
from contextlib import ExitStack, contextmanager
from contextlib import contextmanager
from dataclasses import dataclass
from multiprocessing.process import BaseProcess
from multiprocessing.sharedctypes import Synchronized
@ -30,6 +30,7 @@ from integration._support.client import Gateway, eventually, gateway_from_enviro
from integration._support.database import read_rows
from integration._support.process import kill_stray_process, owned_proxy_process
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
BEDROCK_MODEL: Final = "us.openai.gpt-5.6-sol"
@ -227,14 +228,6 @@ def _calls(count: int, endpoints: tuple[Endpoint, ...], stream: Callable[[int],
)
def _free_ports(count: int) -> tuple[int, ...]:
with ExitStack() as reserved:
sockets: Final = tuple(reserved.enter_context(socket.socket()) for _ in range(count))
for reserve in sockets:
reserve.bind(("127.0.0.1", 0))
return tuple(reserve.getsockname()[1] for reserve in sockets)
def _accepts_connections(port: int) -> bool:
try:
with socket.create_connection(("127.0.0.1", port), timeout=0.2):
@ -283,7 +276,7 @@ def _chaos_config(endpoints: Mapping[str, str], directory: Path) -> Path:
def chaos_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_ChaosProxy]:
directory: Final = tmp_path_factory.mktemp("bedrock-gpt-chat-completions-chaos")
burst, peer_killed, slow_peer = (
_Deployment(f"bedrock-gpt-chat-completions-chaos-{uuid.uuid4().hex}", port) for port in _free_ports(3)
_Deployment(f"bedrock-gpt-chat-completions-chaos-{uuid.uuid4().hex}", allocate_port()) for _ in range(3)
)
endpoints: Final = {
deployment.model: f"http://127.0.0.1:{deployment.peer_port}" for deployment in (burst, peer_killed, slow_peer)

10
tests/integration/providers/test_cache_control_tool_call_marks_owned_proxy.py
View file

@ -1,6 +1,5 @@
import asyncio
import re
import socket
import threading
from collections import Counter
from collections.abc import Callable, Sequence
@ -17,6 +16,7 @@ from integration._support.client import Gateway, eventually
from integration._support.database import read_rows
from integration._support.process import graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from integration.providers._cache_control_marks_support import (
CITIES,
POINTS,
@ -57,12 +57,6 @@ class _Sent:
call_id: str
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return int(reserve.getsockname()[1])
def _surface_request(surface: str, marker: str) -> tuple[str, dict[str, JsonValue]]:
if surface == "messages":
return "/v1/messages", messages_body(_MODEL, marker, stream=False)
@ -150,7 +144,7 @@ def _single_spend_row(item: _Sent) -> None:
async def test_capped_burst_rides_out_a_provider_outage_and_logs_every_request_once(
gateway: Gateway, tmp_path: Path
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
config: Final = owned_config(
tmp_path, [anthropic_deployment(_MODEL, f"http://127.0.0.1:{port}", cache_control_injection_points=POINTS)]
)

10
tests/integration/providers/test_chat_completions_tool_search_chaos.py
View file

@ -1,6 +1,5 @@
import asyncio
import re
import socket
import threading
import uuid
from collections.abc import Callable, Mapping
@ -22,6 +21,7 @@ from integration._support.database import read_rows
from integration._support.openai_wire import chat_reply
from integration._support.process import graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from integration.providers.test_chat_completions_tool_search_wire import (
DEFERRED_NAMES,
UNLISTED_CLAUDE,
@ -191,12 +191,6 @@ def _assert_every_call_forwarded(wire: Wire, calls: tuple[_Call, ...]) -> None:
_assert_forwarded(request, kinds[marker])
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def _identities(served: tuple[_Served, ...], kind: Kind) -> tuple[str, ...]:
return tuple(response_id(item.answer) for item in served if item.call.kind == kind)
@ -236,7 +230,7 @@ def _spend_status(model: str, expected: int) -> dict[str, str]:
async def test_a_vendor_outage_fails_each_call_cleanly_and_the_recovered_vendor_gets_the_regex_tool(
gateway: Gateway,
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
while_down: Final = _calls(15, ("search",))
after: Final = _calls(15, ("search",))
with gateway.scenario() as scenario:

10
tests/integration/providers/test_chatgpt_responses_caller_stream_chaos.py
View file

@ -20,6 +20,7 @@ from integration._support.client import Gateway, eventually, gateway_from_enviro
from integration._support.database import read_rows
from integration._support.process import OwnedProxy, owned_proxy_process
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
pytestmark: Final = pytest.mark.timeout(240)
@ -62,17 +63,10 @@ class _Rig:
return f"http://127.0.0.1:{self.port}"
def _free_port() -> int:
with wire_server(cv.CodexVendor().respond) as probe:
port: Final = urlsplit(probe.url).port
assert port is not None, probe.url
return port
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
directory: Final = tmp_path_factory.mktemp("chatgpt-caller-stream-chaos")
port: Final = _free_port()
port: Final = allocate_port()
overrides: Final = {"CHATGPT_TOKEN_DIR": str(cv.login(directory)), "CHATGPT_API_BASE": f"http://127.0.0.1:{port}"}
config: Final = cv.proxy_config(directory, probe=False)
with gateway_from_environment() as gateway:

10
tests/integration/providers/test_cloudflare_clef_omni_media_chaos.py
View file

@ -1,7 +1,6 @@
import asyncio
import json
import re
import socket
import threading
import uuid
from dataclasses import dataclass
@ -19,6 +18,7 @@ from integration._support.client import Gateway, eventually, object_value, strin
from integration._support.database import read_rows
from integration._support.process import graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
_API_KEY: Final = "synthetic-cloudflare-key"
@ -163,12 +163,6 @@ def _assert_each_billed_once(
assert by_call[item.call_id]["status"] == "success", (item.call_id, by_call[item.call_id])
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def _assert_failed_cleanly(served: _Served) -> None:
assert served.status >= 500, (served.status, served.text)
assert "answers" not in served.text, served.text
@ -180,7 +174,7 @@ def _assert_failed_cleanly(served: _Served) -> None:
async def test_an_upstream_dropping_connections_mid_burst_then_stopped_and_restarted_lands_each_success_once(
gateway: Gateway,
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
first_wave: Final = _markers(12)
dropped: Final = frozenset(marker for index, marker in enumerate(first_wave) if index % 2 == 1)
while_down: Final = _markers(6)

6
tests/integration/providers/test_coralbricks_provider_wire.py
View file

@ -8,6 +8,7 @@ import pytest
from integration._support.client import JSON_OBJECT, Gateway, eventually, object_value, string_value
from integration._support.responses_vendor import same_response
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from integration.providers._coralbricks import (
ANSWER,
API_KEY,
@ -24,7 +25,6 @@ from integration.providers._coralbricks import (
catalog_rows,
async_anthropic_client,
async_openai_client,
closed_port,
deployment,
expected_spend,
failing_provider,
@ -365,7 +365,7 @@ def test_unreachable_api_base_reports_the_connection_error_and_leaves_the_proxy_
with gateway.scenario() as scenario:
control: Final = scenario.model()
alias: Final = scenario.model(
model=f"{PROVIDER}/{MODEL}", api_base=f"http://127.0.0.1:{closed_port()}/v1", api_key=API_KEY
model=f"{PROVIDER}/{MODEL}", api_base=f"http://127.0.0.1:{allocate_port()}/v1", api_key=API_KEY
)
response: Final = eventually(
lambda: gateway.request(
@ -593,7 +593,7 @@ def test_fallback_from_a_refused_primary_lands_on_coralbricks(gateway: Gateway)
with wire_server(provider(identity)) as wire, gateway.scenario() as scenario:
alias: Final = deployment(scenario, wire)
primary: Final = scenario.model(
model=f"{PROVIDER}/{MODEL}", api_base=f"http://127.0.0.1:{closed_port()}/v1", api_key=API_KEY
model=f"{PROVIDER}/{MODEL}", api_base=f"http://127.0.0.1:{allocate_port()}/v1", api_key=API_KEY
)
wire.drain()
response: Final = eventually(

6
tests/integration/providers/test_decisions_box_deployment_flag_chaos.py
View file

@ -18,6 +18,7 @@ from integration._support.client import Gateway, eventually
from integration._support.database import read_rows
from integration._support.process import graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
_API_KEY: Final = "synthetic-decisions-key"
@ -180,10 +181,9 @@ async def test_a_flagged_deployment_recovers_box_answers_after_an_upstream_outag
gateway: Gateway,
) -> None:
base_url: Final = str(gateway.client.base_url)
port: Final = allocate_port()
with gateway.scenario() as scenario:
with wire_server(_reply) as wire:
port: Final = urlsplit(wire.url).port
assert port is not None
with wire_server(_reply, port=port) as wire:
model: Final = scenario.model(
model=_backend(), api_base=wire.url, api_key=_API_KEY, model_info={"supports_box_questions": True}
)

6
tests/integration/providers/test_decisions_chaos.py
View file

@ -18,6 +18,7 @@ from integration._support.client import Gateway, eventually
from integration._support.database import read_rows
from integration._support.process import owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
_BACKEND: Final = "pplx-decider-v1-27b"
@ -239,10 +240,9 @@ async def test_worker_sigkill_mid_burst_leaves_the_sibling_serving_the_default_m
@pytest.mark.timeout(180)
async def test_upstream_outage_fails_its_calls_and_recovery_on_the_same_port_restores_them(gateway: Gateway) -> None:
base_url: Final = str(gateway.client.base_url)
port: Final = allocate_port()
with gateway.scenario() as scenario:
with wire_server(_reply) as wire:
port: Final = urlsplit(wire.url).port
assert port is not None
with wire_server(_reply, port=port) as wire:
model: Final = scenario.model(model=f"perplexity/{_BACKEND}", api_base=wire.url, api_key=_API_KEY)
before: Final = await _burst(base_url, gateway.key, model, _calls(5, fail=False))
assert _health(gateway, model) == (1, 0)

18
tests/integration/providers/test_decisions_wire.py
View file

@ -1,6 +1,5 @@
import json
import math
import socket
import uuid
from collections.abc import Sequence
from dataclasses import dataclass
@ -14,6 +13,7 @@ from integration._support.client import Gateway, Scenario, eventually, object_va
from integration._support.database import read_rows
from integration._support.process import owned_proxy_process
from integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario
from integration._support.ports import allocate_port
from integration.cost_calculation.cost_tracking_case import JsonResponse
from pydantic import JsonValue, TypeAdapter
@ -235,12 +235,6 @@ def _assert_connection_error(response: httpx.Response) -> None:
assert _GENERIC_API_ERROR not in response.text, response.text
def _free_closed_port() -> int:
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
return probe.getsockname()[1]
def _pass_through_config(directory: Path, pass_through_target: str, native_api_base: str) -> Path:
base: Final = _JSON_OBJECT.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()))
config: Final = {
@ -431,8 +425,8 @@ def test_request_body_api_base_is_refused_like_chat_without_an_upstream_call(gat
with gateway.scenario() as scenario:
handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
model: Final = _deployment(scenario, handle, _PERPLEXITY)
decisions: Final = _decide(gateway, model, api_base=f"http://127.0.0.1:{_free_closed_port()}")
chat: Final = _chat(gateway, model, api_base=f"http://127.0.0.1:{_free_closed_port()}")
decisions: Final = _decide(gateway, model, api_base=f"http://127.0.0.1:{allocate_port()}")
chat: Final = _chat(gateway, model, api_base=f"http://127.0.0.1:{allocate_port()}")
assert 400 <= decisions.status_code < 500, decisions.text
assert decisions.status_code == chat.status_code, (decisions.text, chat.text)
assert _upstream_calls(gateway, handle) == []
@ -542,7 +536,7 @@ def test_unreachable_upstream_fails_only_its_own_deployment(gateway: Gateway, pr
handle: Final = _register(scenario, _answer_body(provider))
healthy: Final = _deployment(scenario, handle, provider)
dead: Final = scenario.model(
model=provider.model, api_base=f"http://127.0.0.1:{_free_closed_port()}", api_key=provider.api_key
model=provider.model, api_base=f"http://127.0.0.1:{allocate_port()}", api_key=provider.api_key
)
failed: Final = _decide(gateway, dead, num_retries=0)
_assert_connection_error(failed)
@ -556,7 +550,7 @@ def test_an_unreachable_openrouter_deployment_reports_a_connection_error_on_chat
gateway: Gateway,
) -> None:
with gateway.scenario() as scenario:
dead_api_base: Final = f"http://127.0.0.1:{_free_closed_port()}"
dead_api_base: Final = f"http://127.0.0.1:{allocate_port()}"
chat_model: Final = scenario.model(model=_OPENROUTER_CHAT_MODEL, api_base=dead_api_base, api_key=_API_KEY)
decisions_model: Final = scenario.model(model=_OPENROUTER.model, api_base=dead_api_base, api_key=_API_KEY)
chat: Final = _chat(gateway, chat_model, num_retries=0)
@ -572,7 +566,7 @@ def test_an_unreachable_openrouter_deployment_reports_a_connection_error_on_chat
async def test_sdk_openrouter_connection_failures_raise_a_connection_error(gateway: Gateway) -> None:
dead_api_base: Final = f"http://127.0.0.1:{_free_closed_port()}"
dead_api_base: Final = f"http://127.0.0.1:{allocate_port()}"
with pytest.raises(litellm.APIConnectionError):
litellm.completion(
model=_OPENROUTER_CHAT_MODEL,

10
tests/integration/providers/test_gemini_tts_voice_mapping_wire.py
View file

@ -11,7 +11,6 @@ import asyncio
import base64
import dataclasses
import json
import socket
import threading
import time
from collections.abc import Callable, Mapping
@ -24,6 +23,7 @@ import openai
import pytest
from integration._support.client import JSON_OBJECT, Gateway, Scenario, eventually, object_value, string_value
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from integration.providers._gemini_tts_voice_support import (
AUDIO_TOKENS,
BACKEND,
@ -514,12 +514,6 @@ def test_vertex_live_default_mode_sends_one_setup_without_the_client_voice(gatew
assert voice_in(setups[0]) == NO_VOICE, setups[0]
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return int(reserve.getsockname()[1])
@dataclass(frozen=True, slots=True)
class Fired:
provider: str
@ -586,7 +580,7 @@ def _assert_served(items: tuple[Fired, ...], received: tuple[Request, ...]) -> N
def test_peer_outage_mid_burst_ends_each_request_once_and_recovers_on_the_same_port(gateway: Gateway) -> None:
port: Final = _free_port()
port: Final = allocate_port()
with gateway.scenario() as scenario:
url: Final = f"http://127.0.0.1:{port}"
models: Final = {

5
tests/integration/providers/test_idle_upstream_reconnect_chaos_wire.py
View file

@ -8,6 +8,7 @@ from typing import Final
import pytest
from integration._support.client import Gateway
from integration._support.idle_gateway import idle_gateway
from integration._support.ports import allocate_port
from integration._support.process import graceful_stop_seconds
from integration.providers._idle_upstream_support import (
DROP_AFTER_SECONDS,
@ -108,7 +109,9 @@ def test_upstream_restarted_on_the_same_port_with_connections_pooled_serves_a_bu
before_restart: Final = stack.enter_context(ExitStack())
first: Final = MappingProxyType(
{
row.name: before_restart.enter_context(idle_gateway(answer, drop_after=DROP_AFTER_SECONDS))
row.name: before_restart.enter_context(
idle_gateway(answer, drop_after=DROP_AFTER_SECONDS, port=allocate_port())
)
for row in RESTART_ROWS
}
)

11
tests/integration/providers/test_responses_bridge_message_cache_control_wire_chaos.py
View file

@ -8,7 +8,6 @@ from pathlib import Path
from queue import SimpleQueue
from types import MappingProxyType
from typing import Final, TypeAlias
from urllib.parse import urlsplit
import httpx
import pytest
@ -25,6 +24,7 @@ from integration._support.client import (
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
from integration._support.openai_wire import answering_model_discovery, chat_reply
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from integration.providers._cache_control_marks_support import (
ANTHROPIC_MODEL,
ASK,
@ -236,13 +236,6 @@ def _health(gateway: Gateway, model: str) -> Mapping[str, JsonValue]:
return _JSON_OBJECT.validate_json(response.text)
def _free_port() -> int:
with wire_server(anthropic_peer) as probe:
port: Final = urlsplit(probe.url).port
assert port is not None, probe.url
return port
def _deployment(scenario: Scenario, api_base: str) -> str:
return scenario.model(model=f"anthropic/{ANTHROPIC_MODEL}", api_base=api_base, api_key=PROVIDER_KEY)
@ -262,7 +255,7 @@ async def test_mixed_burst_carries_the_system_mark_once_per_call(gateway: Gatewa
async def test_peer_outage_fails_cleanly_and_the_restarted_peer_receives_the_marks(
gateway: Gateway, spend: pcb.SpendLogs
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
while_down: Final = _calls(12)
after: Final = _calls(12)
with gateway.scenario() as scenario:

8
tests/integration/providers/test_responses_bridge_prompt_cache_breakpoint_chaos.py
View file

@ -8,7 +8,6 @@ from contextlib import ExitStack
from pathlib import Path
from queue import SimpleQueue
from typing import Final
from urllib.parse import urlsplit
import httpx
import psutil
@ -17,6 +16,7 @@ import yaml
from integration._support.client import Gateway, eventually
from integration._support.process import owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from integration.providers._responses_bridge_prompt_cache_breakpoint import (
_Call,
_JSON_OBJECT,
@ -86,8 +86,9 @@ async def test_worker_and_peer_outages_preserve_markers_and_recover(
assert gate.wait(timeout=60), "The worker-kill burst was never released"
return _responses_reply(request)
upstream_port: Final = allocate_port()
with ExitStack() as peer_stack:
wire: Final = peer_stack.enter_context(wire_server(held))
wire: Final = peer_stack.enter_context(wire_server(held, port=upstream_port))
config: Final = _chaos_config(wire, tmp_path)
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as owned:
try:
@ -113,9 +114,6 @@ async def test_worker_and_peer_outages_preserve_markers_and_recover(
early_successful: Final = tuple(item for item in early_served if item.status == 200)
for item in early_successful:
_assert_spend_for_result(item, _CONFIG_MODEL)
upstream_port_value: Final = urlsplit(wire.url).port
assert upstream_port_value is not None
upstream_port: Final = upstream_port_value
active_by_worker: Final = eventually(
lambda: {pid: _open_upstream_connections(pid, upstream_port) for pid in workers},
lambda counts: sum(counts.values()) == len(calls) - len(early_calls),

11
tests/integration/providers/test_responses_bridge_prompt_cache_breakpoint_wire_chaos.py
View file

@ -9,7 +9,6 @@ from pathlib import Path
from queue import SimpleQueue
from types import MappingProxyType
from typing import Final, Literal, TypeAlias
from urllib.parse import urlsplit
import httpx
import pytest
@ -19,6 +18,7 @@ from integration._support import responses_vendor as rv
from integration._support.client import Gateway, eventually, gateway_from_environment, string_value
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
@ -213,13 +213,6 @@ def _health(gateway: Gateway, model: str) -> Mapping[str, JsonValue]:
return rv.JSON_OBJECT.validate_json(response.text)
def _free_port() -> int:
with wire_server(pcb.respond) as probe:
port: Final = urlsplit(probe.url).port
assert port is not None, probe.url
return port
def _chat(gateway: Gateway, model: str, marker: str, breakpoint: JsonValue) -> httpx.Response:
return gateway.request("POST", "/v1/chat/completions", dict(_body(model, _Call("chat", False, marker), breakpoint)))
@ -285,7 +278,7 @@ async def test_mixed_burst_carries_every_marker_once(gateway: Gateway, spend: pc
async def test_upstream_outage_fails_cleanly_and_the_restarted_upstream_serves_marked_calls(
gateway: Gateway, spend: pcb.SpendLogs
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
while_down: Final = _calls(12, _ENDPOINTS)
after: Final = _calls(12, _ENDPOINTS)
with gateway.scenario() as scenario:

10
tests/integration/providers/test_responses_bridge_tool_search_chaos.py
View file

@ -2,7 +2,6 @@ import asyncio
import itertools
import json
import re
import socket
import threading
import uuid
from collections.abc import Callable, Iterator, Mapping, Sequence
@ -28,6 +27,7 @@ from integration._support.openai_wire import answering_model_discovery
from integration._support.database import read_rows
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
_CLAUDE: Final = cc.OPUS
@ -339,12 +339,6 @@ def _assert_each_lands_once(
assert by_status[match] == "success", rows
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def _health_counts(gateway: Gateway, model: str) -> tuple[int, int]:
response: Final = gateway.request("GET", f"/health?model={model}", None)
assert response.status_code in (200, 503), response.text
@ -382,7 +376,7 @@ async def test_mixed_burst_carries_the_search_call_on_every_endpoint_and_lands_e
async def test_vendor_outage_fails_each_call_cleanly_and_the_recovered_vendor_gets_the_tool_search(
gateway: Gateway,
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
while_down: Final = _calls(15, ("responses", "chat", "messages"))
after: Final = _calls(15, ("responses", "chat", "messages"))
with gateway.scenario() as scenario:

10
tests/integration/providers/test_responses_minted_reasoning_replay_chaos.py
View file

@ -1,7 +1,6 @@
import asyncio
import json
import re
import socket
import threading
import time
import uuid
@ -25,6 +24,7 @@ from integration._support.database import read_rows
from integration._support.process import OwnedProxy, owned_proxy_process
from integration._support.tls import server_context, write_self_signed_cert
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
_GPT: Final = "gpt-5.6"
@ -226,12 +226,6 @@ def _assert_each_lands_once(
assert by_status[match] == "success", rows
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def _health_counts(gateway: Gateway, model: str) -> tuple[int, int]:
response: Final = gateway.request("GET", f"/health?model={model}", None)
assert response.status_code in (200, 503), response.text
@ -249,7 +243,7 @@ def _marked(received: tuple[Request, ...]) -> dict[str, Request]:
async def test_vendor_outage_fails_each_replay_cleanly_and_the_recovered_vendor_gets_them_without_minted_items(
gateway: Gateway,
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
while_down: Final = _calls(15, ("responses", "chat", "messages"))
after: Final = _calls(15, ("responses", "chat", "messages"))
with gateway.scenario() as scenario:

10
tests/integration/providers/test_sagemaker_embedding_chaos.py
View file

@ -1,6 +1,5 @@
import asyncio
import re
import socket
import threading
import uuid
from collections.abc import Iterator
@ -29,6 +28,7 @@ from integration._support.sagemaker_embedding import (
respond,
)
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
pytestmark = pytest.mark.timeout(300)
@ -54,12 +54,6 @@ class _Served:
call_id: str
def _free_port() -> int:
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
return probe.getsockname()[1]
def _config(directory: Path) -> Path:
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
config["model_list"] = [openai_deployment(_MODEL, "openai", model_info={"mode": "embedding"})]
@ -80,7 +74,7 @@ def rig() -> Iterator[Gateway]:
@pytest.fixture(scope="module")
def peer_port() -> int:
return _free_port()
return allocate_port()
@pytest.fixture(scope="module")

9
tests/integration/providers/test_vertex_anthropic_structured_output_wire.py
View file

@ -28,6 +28,7 @@ from integration._support.process import OwnedProxy, graceful_stop_seconds, owne
from integration._support.upstream import aws_event_stream_frame
from integration._support.vertex import service_account_json
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
_JSON_TOOL: Final = "json_tool_call"
@ -1734,9 +1735,9 @@ def test_vertex_chat_peer_error_reaches_the_caller_and_the_deployment_keeps_serv
) -> None:
marker, marker_hex = _marker("wrapped")
after, after_hex = _marker("wrapped")
port: Final = allocate_port()
with gateway.scenario() as scenario:
with wire_server(_failing(status)) as failing:
port: Final = int(failing.url.rsplit(":", 1)[1])
with wire_server(_failing(status), port=port) as failing:
model: Final = _vertex_deployment(gateway, scenario, failing.url, _TOOL_MODEL)
response: Final = gateway.request("POST", "/v1/chat/completions", _chat_body(model, marker))
assert response.status_code == status, response.text
@ -1830,8 +1831,8 @@ def test_upstream_outage_mid_run_recovers_with_envelopes_still_unwrapped(gateway
assert observed == _payload(marker_hex), response.text
_seen_once(_delivered(wire), tuple(marker_hex for _, marker_hex, _ in results))
with wire_server(_peer) as wire:
port: Final = int(wire.url.rsplit(":", 1)[1])
port: Final = allocate_port()
with wire_server(_peer, port=port) as wire:
model: Final = _vertex_deployment(gateway, scenario, wire.url, _TOOL_MODEL)
unwrapped(wire, wave())
outage: Final = wave()

4
tests/integration/providers/test_vertex_batch_row_line_errors_chaos.py
View file

@ -8,12 +8,12 @@ from typing import Final
import httpx
from integration._support.client import Gateway, eventually
from integration._support.wire import Request, wire_server
from integration._support.ports import allocate_port
from integration.providers._vertex_batch_rows import (
ROUTES,
Route,
chat_row,
closed_port,
encoded,
error_of,
gcs_peer,
@ -124,7 +124,7 @@ def test_concurrent_mixed_burst_names_each_bad_files_own_line_and_lands_each_goo
def test_bucket_outage_answers_500_without_blaming_a_row_and_recovers_on_restart(gateway: Gateway) -> None:
port: Final = closed_port()
port: Final = allocate_port()
with gateway.scenario() as scenario, _burst_client(gateway) as client:
model: Final = vertex_model(gateway, scenario, f"http://127.0.0.1:{port}")
with wire_server(gcs_peer, port=port) as before:

6
tests/integration/providers/test_vertex_batch_row_line_errors_owned_proxy.py
View file

@ -17,12 +17,12 @@ import yaml
from integration._support.client import Gateway, eventually, gateway_from_environment
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Request, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
from integration.providers._vertex_batch_rows import (
Route,
chat_row,
closed_port,
encoded,
error_of,
gcs_peer,
@ -97,8 +97,8 @@ def _config(gateway: Gateway, directory: Path, vertex_port: int, openai_port: in
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
directory: Final = tmp_path_factory.mktemp("vertex-batch-row-lines")
vertex_port: Final = closed_port()
openai_port: Final = closed_port()
vertex_port: Final = allocate_port()
openai_port: Final = allocate_port()
with gateway_from_environment() as shared:
config: Final = _config(shared, directory, vertex_port, openai_port)
with owned_proxy_process(shared, directory, {}, config=config, workers=2) as owned:

6
tests/integration/providers/test_vertex_batch_row_line_errors_wire.py
View file

@ -7,6 +7,7 @@ import openai
import pytest
from integration._support.client import Gateway, Scenario
from integration._support.wire import Reply, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
from integration.providers._vertex_batch_rows import (
@ -16,7 +17,6 @@ from integration.providers._vertex_batch_rows import (
ROUTES,
Route,
chat_row,
closed_port,
encoded,
error_of,
gcs_peer,
@ -264,7 +264,7 @@ def test_bucket_failures_keep_their_status_and_never_blame_a_row(
gateway: Gateway, route: Route, reply: Reply | None, status: int
) -> None:
with wire_server(lambda _request: reply or Reply()) as wire, gateway.scenario() as scenario:
url: Final = wire.url if reply is not None else f"http://127.0.0.1:{closed_port()}"
url: Final = wire.url if reply is not None else f"http://127.0.0.1:{allocate_port()}"
model: Final = vertex_model(gateway, scenario, url)
response: Final = upload(gateway, route, model, jsonl(encoded(chat_row("first", model, "one"))))
assert response.status_code == status, response.text
@ -330,7 +330,7 @@ def _unreachable_anthropic_model(scenario: Scenario) -> str:
return scenario.model(
model="anthropic/claude-haiku-4-5",
api_key="synthetic-anthropic-key",
api_base=f"http://127.0.0.1:{closed_port()}",
api_base=f"http://127.0.0.1:{allocate_port()}",
num_retries=0,
)

6
tests/integration/providers/test_vertex_partner_count_tokens_wire.py
View file

@ -1,6 +1,5 @@
import json
import re
import socket
import threading
import uuid
from collections.abc import Callable, Mapping, Sequence
@ -18,6 +17,7 @@ from integration._support.client import Gateway, Scenario, eventually
from integration._support.process import owned_proxy_process
from integration._support.vertex import service_account_json
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
_BACKEND: Final = "claude-sonnet-4-6"
@ -224,9 +224,7 @@ def _local_count(gateway: Gateway, body: Mapping[str, JsonValue]) -> int:
def _closed_port_url() -> str:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return f"http://127.0.0.1:{reserve.getsockname()[1]}"
return f"http://127.0.0.1:{allocate_port()}"
def _clients(stack: ExitStack, base_url: str, count: int) -> tuple[httpx.Client, ...]:

11
tests/integration/providers/test_voyage_mongodb_host_wire.py
View file

@ -2,7 +2,6 @@ import asyncio
import json
import os
import re
import socket
import uuid
from collections.abc import Callable, Iterator
from dataclasses import dataclass
@ -17,6 +16,7 @@ from integration._support.database import read_rows
from integration._support.forward_proxy import ForwardProxy, refusing_forward_proxy
from integration._support.process import owned_proxy, owned_proxy_process
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from openai import APIStatusError, AsyncOpenAI, OpenAI
from pydantic import JsonValue, TypeAdapter
@ -617,15 +617,8 @@ def _spend_rows(call_id: str) -> list[dict[str, JsonValue]]:
return read_rows('SELECT request_id FROM "LiteLLM_SpendLogs" WHERE request_id=%s', (call_id,))
def _reserve_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
address: Final[Callable[[], tuple[str, int]]] = reserve.getsockname
return address()[1]
async def test_forward_proxy_outage_mid_traffic_recovers_with_the_right_hosts(gateway: Gateway, tmp_path: Path) -> None:
port: Final = _reserve_port()
port: Final = allocate_port()
config: Final = _write_config(tmp_path, "outage", _isolated_deployments(gateway.upstream_url))
with owned_proxy(
gateway,

10
tests/integration/routing/test_declared_provider_bare_model_routes.py
View file

@ -9,7 +9,6 @@ from __future__ import annotations
import asyncio
import json
import socket
import time
import uuid
from collections.abc import Awaitable, Callable
@ -29,6 +28,7 @@ from websockets.exceptions import ConnectionClosed
from tests.integration._support.client import JSON_OBJECT, Gateway, Scenario, object_value, string_value
from tests.integration._support.upstream import delete_scenario, register_scenario
from tests.integration._support.wire import Reply, Request, Wire, wire_server
from tests.integration._support.ports import allocate_port
from tests.integration.cost_calculation.cost_tracking_case import RealtimeResponse
_KEY: Final = "synthetic-declared-provider-key"
@ -497,14 +497,8 @@ def _marker(request: Request) -> str:
return string_value(body["prompt"] if request.target == _IMAGES else body["input"])
def _closed_port() -> int:
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def test_burst_across_declared_and_dead_deployments_reaches_each_live_provider_once(gateway: Gateway) -> None:
dead_base: Final = f"http://127.0.0.1:{_closed_port()}/v1"
dead_base: Final = f"http://127.0.0.1:{allocate_port()}/v1"
with wire_server(_burst_reply) as wire, gateway.scenario() as scenario:
live: Final = tuple(
(route, _deployment(scenario, wire, form, _bare()))

5
tests/integration/routing/test_fallback_hop_foreign_encrypted_reasoning.py
View file

@ -46,6 +46,7 @@ from integration._support.database import read_rows, scratch_database
from integration._support.process import graceful_stop_seconds, owned_proxy_process
from integration._support.upstream import delete_scenario, register_scenario
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from integration.cost_calculation.cost_tracking_case import JsonResponse, RoutedResponse, SseResponse, StoredResponse
from pydantic import JsonValue
@ -1214,10 +1215,10 @@ def test_order_one_outage_mid_burst_hops_stripped_and_recovers_after_restart(gat
order_two: Final = deployment(
scenario, name, scenario_api_base(scenario, two_scenario, healthy_json()), order=2
)
with wire_server(peer_reply) as peer:
port: Final = allocate_port()
with wire_server(peer_reply, port=port) as peer:
order_one: Final = deployment(scenario, name, peer.url, order=1)
target: Final = Target(name, frozenset({order_one, order_two}))
port: Final = int(peer.url.rsplit(":", 1)[-1])
warm: Final = marker()
assert served_by_order_one(responses_answer(gateway, HTTPX, target, history(warm)), order_one), warm
assert [json.loads(request.body)["input"] for request in peer.drain()] == [history(warm)]

7
tests/integration/sdk/test_http2_wire.py
View file

@ -10,7 +10,7 @@ import time
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Final, cast
from typing import Final
import pytest
from cryptography import x509
@ -22,6 +22,7 @@ from hypercorn.asyncio import (
)
from hypercorn.config import Config
from hypercorn.typing import ASGIReceiveCallable, ASGISendCallable, HTTPResponseBodyEvent, HTTPResponseStartEvent, Scope
from integration._support.ports import allocate_port
STREAM_CHUNKS: Final = 3
@ -102,9 +103,7 @@ async def _peer(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallab
@pytest.fixture(scope="module")
def http2_tls_peer(tmp_path_factory: pytest.TempPathFactory) -> Iterator[str]:
cert_file, key_file = _write_self_signed_cert(tmp_path_factory.mktemp("h2certs"))
with socket.socket() as sock:
sock.bind(("127.0.0.1", 0))
port: Final = cast(int, sock.getsockname()[1])
port: Final = allocate_port()
shutdown: Final = threading.Event()
def _serve() -> None:

33
tests/integration/sdk/test_owned_redis_ports.py
View file

@ -1,12 +1,10 @@
from __future__ import annotations
import re
from itertools import chain
from pathlib import Path
from typing import Final
import pytest
from integration._support.redis_process import PORT_ATTEMPTS, free_loopback_port, owned_redis
from integration._support.redis_process import owned_redis
from redis import Redis
@ -15,20 +13,6 @@ def _redis_pid(host: str, port: int) -> int:
return int(client.info("server")["process_id"])
def test_owned_redis_moves_off_a_port_another_owned_redis_already_serves(tmp_path: Path) -> None:
with owned_redis(tmp_path / "first") as first:
candidates: Final = chain((first.port,), iter(free_loopback_port, None))
with owned_redis(tmp_path / "second", lambda: next(candidates)) as second:
assert second.port != first.port
assert first.serves()
assert second.serves()
assert second.server_pid != first.server_pid
assert _redis_pid(second.host, second.port) == second.server_pid
assert _redis_pid(first.host, first.port) == first.server_pid
assert not second.serves()
assert first.serves()
def test_owned_redis_restart_refuses_a_port_another_redis_took_while_it_was_down(tmp_path: Path) -> None:
with owned_redis(tmp_path / "first") as first:
first.stop()
@ -39,18 +23,3 @@ def test_owned_redis_restart_refuses_a_port_another_redis_took_while_it_was_down
first.start()
assert first.process is None
assert _redis_pid(squatter.host, squatter.port) == squatter.server_pid
def test_owned_redis_gives_up_with_the_last_cause_after_a_bounded_number_of_ports(tmp_path: Path) -> None:
with owned_redis(tmp_path / "first") as first:
ports: Final = iter((first.port,) * PORT_ATTEMPTS)
with pytest.raises(
AssertionError,
match=(
f"after {PORT_ATTEMPTS} attempts: Another Redis already serves {re.escape(first.host)}:{first.port}"
),
):
with owned_redis(tmp_path / "second", lambda: next(ports)):
pytest.fail("Owned Redis unexpectedly reused a port already serving Redis")
assert next(ports, None) is None
assert first.serves()

15
tests/integration/streaming/test_faked_responses_stream_chaos.py
View file

@ -1,6 +1,5 @@
import asyncio
import json
import socket
import threading
import uuid
from collections.abc import Iterator, Mapping, Sequence
@ -27,6 +26,7 @@ from integration._support.client import (
from integration._support.database import read_rows
from integration._support.process import OwnedProxy, 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 (
ANTHROPIC_TOOLS,
API_KEY,
@ -39,7 +39,7 @@ from integration.streaming.faked_responses_case import (
message_item,
response_body,
)
from pydantic import JsonValue, TypeAdapter
from pydantic import JsonValue
pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
@ -68,9 +68,6 @@ class _Served:
call_id: str
_ADDRESS: Final = TypeAdapter(tuple[str, int])
@dataclass(frozen=True, slots=True)
class _Rig:
port: int
@ -81,12 +78,6 @@ class _Rig:
return self.proxy.gateway
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return _ADDRESS.validate_python(reserve.getsockname())[1]
def _deployment(name: str, model: str, port: int) -> dict[str, JsonValue]:
return {
"model_name": name,
@ -117,7 +108,7 @@ def _config(port: int, directory: Path) -> Path:
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
directory: Final = tmp_path_factory.mktemp("faked-stream-chaos")
port: Final = _free_port()
port: Final = allocate_port()
with (
gateway_from_environment() as shared,
owned_proxy_process(shared, directory, {}, config=_config(port, directory), workers=2) as owned,

12
tests/integration/streaming/test_responses_bridge_stream_chaos.py
View file

@ -1,6 +1,5 @@
import asyncio
import re
import socket
import threading
import uuid
from collections.abc import Iterator, Mapping, Sequence
@ -35,6 +34,7 @@ from integration._support.responses_stream import (
rate_limited_stream,
)
from integration._support.wire import Reply, Request, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
@ -94,12 +94,6 @@ class _Rig:
return self.proxy.gateway
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return reserve.getsockname()[1]
def _config(port: int, directory: Path) -> Path:
stock: Final = JSON_OBJECT.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()))
router_settings: Final = object_value(stock.get("router_settings") or {})
@ -128,7 +122,7 @@ def _config(port: int, directory: Path) -> Path:
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
directory: Final = tmp_path_factory.mktemp("bridged-stream-chaos")
port: Final = _free_port()
port: Final = allocate_port()
with (
gateway_from_environment() as shared,
owned_proxy_process(shared, directory, {}, config=_config(port, directory), workers=2) as owned,
@ -382,7 +376,7 @@ async def test_worker_sigkill_mid_burst_leaves_the_sibling_answering_the_bridged
async def test_proxy_sigterm_mid_burst_drains_the_spend_log_queue_and_the_restarted_proxy_serves(
gateway: Gateway, tmp_path: Path
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
config: Final = _config(port, tmp_path)
calls: Final = _calls(20, _CHAT_KINDS)
held: Final = _Held(threading.Event(), SimpleQueue())

13
tests/integration/streaming/test_responses_bridge_stream_errors.py
View file

@ -1,8 +1,6 @@
import json
import socket
import uuid
from collections.abc import Iterator, Mapping, Sequence
from contextlib import ExitStack
from dataclasses import dataclass
from pathlib import Path
from typing import Final
@ -47,6 +45,7 @@ from integration._support.responses_stream import (
sse_events,
)
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue
_PRIMARY: Final = "bridged-primary"
@ -110,14 +109,6 @@ class FallbackProxy:
spare_port: int
def _free_ports(count: int) -> tuple[int, ...]:
with ExitStack() as reserved:
sockets: Final = tuple(reserved.enter_context(socket.socket()) for _ in range(count))
for reserve in sockets:
reserve.bind(("127.0.0.1", 0))
return tuple(reserve.getsockname()[1] for reserve in sockets)
def _fallback_config(directory: Path, primary_port: int, spare_port: int) -> Path:
base: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
deployments: Final = [
@ -140,7 +131,7 @@ def _fallback_config(directory: Path, primary_port: int, spare_port: int) -> Pat
@pytest.fixture(scope="module")
def fallback_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[FallbackProxy]:
directory: Final = tmp_path_factory.mktemp("bridged-fallbacks")
primary_port, spare_port = _free_ports(2)
primary_port, spare_port = allocate_port(), allocate_port()
with (
gateway_from_environment() as shared,
owned_proxy(shared, directory, {}, config=_fallback_config(directory, primary_port, spare_port)) as owned,

12
tests/integration/streaming/test_stream_open_error_provider_headers_proxy.py
View file

@ -1,7 +1,6 @@
import asyncio
import json
import re
import socket
import threading
import uuid
from collections.abc import Iterator, Mapping, Sequence
@ -29,6 +28,7 @@ from integration._support.client import (
from integration._support.database import read_rows
from integration._support.process import OwnedProxy, configless_proxy_process, graceful_stop_seconds
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from openai.types.responses import ResponseFailedEvent
from pydantic import JsonValue
@ -76,12 +76,6 @@ class _Served:
call_id: str
def _free_port() -> int:
with socket.socket() as reserve:
reserve.bind(("127.0.0.1", 0))
return reserve.getsockname()[1]
@contextmanager
def _configless(gateway: Gateway, directory: Path, port: int) -> Iterator[_Rig]:
url: Final = f"http://127.0.0.1:{port}"
@ -99,7 +93,7 @@ def _configless(gateway: Gateway, directory: Path, port: int) -> Iterator[_Rig]:
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
with (
gateway_from_environment() as shared,
_configless(shared, tmp_path_factory.mktemp("configless"), _free_port()) as owned,
_configless(shared, tmp_path_factory.mktemp("configless"), allocate_port()) as owned,
):
yield owned
@ -543,7 +537,7 @@ async def test_worker_sigkill_mid_burst_leaves_the_sibling_answering_each_stream
async def test_proxy_sigterm_mid_burst_answers_each_in_flight_stream_and_the_restarted_proxy_serves(
gateway: Gateway, tmp_path: Path
) -> None:
port: Final = _free_port()
port: Final = allocate_port()
calls: Final = _stream_calls(20)
held: Final = _held()
with wire_server(held.respond, port=port) as wire:

10
tests/integration/test_oci_proxy_integration.py
View file

@ -32,7 +32,6 @@ from __future__ import annotations
import json
import os
import socket
import subprocess
import sys
import time
@ -43,6 +42,7 @@ import httpx
import pytest
from tests._master_key import MASTER_KEY
from tests.integration._support.process import stop_child
from tests.integration._support.ports import allocate_port
# ---------------------------------------------------------------------------
# Skip gate
@ -59,12 +59,6 @@ STARTUP_TIMEOUT_S = 90.0
REQUEST_TIMEOUT_S = 120.0
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def _wait_for_health(base_url: str, proc: subprocess.Popen, deadline: float) -> None:
"""Poll /health/liveliness until the proxy answers or the deadline expires."""
while time.monotonic() < deadline:
@ -118,7 +112,7 @@ def _serve(config_path: str) -> Iterator[str]:
env.pop("DATABASE_URL", None)
env["STORE_MODEL_IN_DB"] = "False"
port = _free_port()
port = allocate_port()
base_url = f"http://127.0.0.1:{port}"
# Prefer the `litellm` console script that lives next to the active

44
tests/unit/integration_support/test_process.py
View file

@ -1,10 +1,7 @@
from __future__ import annotations
import asyncio
import errno
import importlib
import os
import socket
import subprocess
import sys
import threading
@ -37,7 +34,6 @@ from tests.integration._support.process import (
)
TESTS_DIR: Final = Path(__file__).resolve().parents[2]
UNRELATED_CRASH: Final = "Traceback (most recent call last):\nModuleNotFoundError: No module named 'litellm'\n"
def test_worker_configuration_routes_stock_edges_to_its_own_servers_and_preserves_other_targets(
@ -111,46 +107,6 @@ def process_module(monkeypatch: pytest.MonkeyPatch) -> ModuleType:
return importlib.import_module("integration._support.process")
async def _refused_bind(port: int) -> OSError | None:
try:
await asyncio.get_running_loop().create_server(asyncio.Protocol, "127.0.0.1", port)
except OSError as refused:
return refused
return None
@pytest.fixture
def bind_error_line() -> Iterator[str]:
with socket.socket() as held:
held.bind(("127.0.0.1", 0))
held.listen()
refused: Final = asyncio.run(_refused_bind(held.getsockname()[1]))
assert refused is not None and refused.errno == errno.EADDRINUSE
yield f"ERROR: {refused}\n"
def _written_log(directory: Path, text: str) -> Path:
log: Final = directory / "owned-proxy.log"
log.write_text(text)
return log
def test_lost_port_race_matches_the_bind_error_the_server_logs(
process_module: ModuleType, bind_error_line: str, tmp_path: Path
) -> None:
assert process_module._lost_port_race(1, _written_log(tmp_path, bind_error_line))
def test_lost_port_race_ignores_an_exit_for_another_reason(process_module: ModuleType, tmp_path: Path) -> None:
assert not process_module._lost_port_race(1, _written_log(tmp_path, UNRELATED_CRASH))
def test_lost_port_race_needs_the_process_to_have_exited(
process_module: ModuleType, bind_error_line: str, tmp_path: Path
) -> None:
assert not process_module._lost_port_race(None, _written_log(tmp_path, bind_error_line))
@pytest.mark.parametrize("failure", (psutil.AccessDenied, psutil.NoSuchProcess))
def test_an_unreadable_process_cannot_establish_listener_ownership(
process_module: ModuleType, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, failure: type[psutil.Error]

18
tests/unit/test_check_test_quality.py
View file

@ -640,6 +640,10 @@ _VIOLATING_SNIPPETS: Final = MappingProxyType(
),
"TQ010": ("tests/integration/chaos/test_snippet.py", "import os, signal\n\nos.kill(1, signal.SIGKILL)\n"),
"TQ011": ("tests/integration/chaos/test_snippet.py", "def test_fast():\n elapsed = run()\n assert elapsed < 2\n"),
"TQ012": (
"tests/integration/_support/snippet.py",
"def port():\n probe.bind(('127.0.0.1', 0))\n return probe.getsockname()[1]\n",
),
}
)
@ -792,6 +796,20 @@ def test_lower_bounds_relative_bounds_counts_and_polling_deadlines_are_left_alon
assert _chaos_codes(tmp_path, source) == []
@pytest.mark.parametrize(
("source", "expected"),
(
("def port():\n probe.bind(('127.0.0.1', 0))\n return probe.getsockname()[1]\n", ["TQ012"]),
("def port():\n probe.bind((host, 0))\n _, number = probe.getsockname()\n return number\n", ["TQ012"]),
("def serve(port):\n probe.bind(('127.0.0.1', port))\n return probe.getsockname()[1]\n", []),
("def client():\n probe.bind(('127.0.0.1', 0))\n\n\ndef other():\n return probe.getsockname()\n", []),
),
)
def test_a_port_read_back_from_a_port_zero_bind_is_flagged_anywhere_under_integration(tmp_path, source, expected):
assert _chaos_codes(tmp_path, source, "tests/integration/_support/snippet.py") == expected
assert _chaos_codes(tmp_path, source, "tests/unit/test_snippet.py") == []
def test_a_clock_bound_name_is_only_an_elapsed_time_in_the_function_that_binds_it(tmp_path):
source = (
"import time\n\n\n"