From 18713e613acb26da2a24df6359c0d76e3783d74c Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Thu, 6 Aug 2026 22:05:41 +0000 Subject: [PATCH] Fix flaky moto_server fixture by using an OS-assigned ephemeral port The `moto_server` session fixture hardcoded port 5001 and pre-bound a socket to it purely to detect conflicts (added in #292). This is still flaky: when tests run in parallel (e.g. multiple CI jobs on a shared runner) or when a previous run leaves the port in TIME_WAIT, the pre-bind raises `OSError: [Errno 98] Address already in use`, failing the whole test session at fixture setup. Bind to port 0 instead so the OS assigns a free ephemeral port, and read the actual bound port back via moto's `get_host_and_port()`. This removes the collision and the now-unnecessary pre-bind check (and the `socket` import). The fixture return annotation is corrected to `Generator[...]` since it yields. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/conftest.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 51a663434f..cca9146c16 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -27,7 +27,6 @@ import os import re -import socket import string import time import uuid @@ -2341,14 +2340,14 @@ def fixture_aws_credentials() -> Generator[None, None, None]: @pytest.fixture(scope="session") -def moto_server() -> "ThreadedMotoServer": +def moto_server() -> Generator["ThreadedMotoServer", None, None]: from moto.server import ThreadedMotoServer - server = ThreadedMotoServer(ip_address="localhost", port=5001) - - # this will throw an exception if the port is already in use - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind((server._ip_address, server._port)) + # Bind to port 0 so the OS assigns a free ephemeral port. A hardcoded port + # collides when tests run in parallel (e.g. on shared CI agents) or when a + # previous run leaves the port in TIME_WAIT, raising + # "OSError: [Errno 98] Address already in use". + server = ThreadedMotoServer(ip_address="localhost", port=0) server.start() yield server @@ -2357,7 +2356,8 @@ def moto_server() -> "ThreadedMotoServer": @pytest.fixture(scope="session") def moto_endpoint_url(moto_server: "ThreadedMotoServer") -> str: - _url = f"http://{moto_server._ip_address}:{moto_server._port}" + host, port = moto_server.get_host_and_port() + _url = f"http://{host}:{port}" return _url