diff --git a/.agents/skills/e2e/SKILL.md b/.agents/skills/e2e/SKILL.md index 7d626cf6..99a8f8fe 100644 --- a/.agents/skills/e2e/SKILL.md +++ b/.agents/skills/e2e/SKILL.md @@ -84,7 +84,9 @@ Keep URL template coverage split by resolver responsibility. Do not recreate a m - Import helpers only from `helpers`; update both imports and `__all__` in `helpers/__init__.py` when adding a helper. - Never hardcode ports. Use `find_free_port()`, `find_free_udp_port()`, or - `find_free_udp_port_pair()`. + `find_free_udp_port_pair()`. These allocate from a per-xdist-worker range + below typical ephemeral ports (14000-32399) so parallel client connections + cannot steal a listen port before rtp2httpd binds it. - Prefer module-scoped or class-scoped `R2HProcess` fixtures when tests share config and args. - Use per-test `R2HProcess` only for mutually exclusive configs, port-range behavior, timeout or log-capture cases, Unix socket cases, or tests whose process state must be isolated. @@ -121,9 +123,11 @@ Port and process helpers: - `find_free_port()` - `find_free_udp_port()` - `find_free_udp_port_pair()` +- `worker_port_range()` - `wait_for_port(port, host="127.0.0.1", timeout=5.0)` - `wait_for_unix_socket(path, timeout=5.0)` - `R2HProcess(binary, port, extra_args=None, config_content=None, capture_log=False, listen=None)` + (`start()` fail-fasts if the process exits and includes rtp2httpd logs) - `make_m3u_rtsp_config(port, rtsp_port, service_name="Test RTSP")` Config and file helpers: @@ -242,7 +246,8 @@ assert_etag_cache_behavior("127.0.0.1", shared_r2h.port, "/playlist.m3u") - Missing marker or marker missing from `pyproject.toml`. - Helper added but not re-exported in `helpers/__init__.py`. - - Hardcoded port or port pair collision. + - Hardcoded port or port pair collision, or a listen port from `bind(0)` + in the ephemeral range (use `find_free_port()`). - Mock server started after rtp2httpd needs it. - `stream_get()` timeout too short for streaming or multicast. - External M3U fetch via `-M http://...` or `-M file://...` needs a short async wait. diff --git a/e2e/helpers/__init__.py b/e2e/helpers/__init__.py index a1671d4d..a860825e 100644 --- a/e2e/helpers/__init__.py +++ b/e2e/helpers/__init__.py @@ -6,7 +6,7 @@ Sub-modules: constants - project paths and platform constants - ports - free port allocation and wait_for_port + ports - per-worker listen-port allocation and wait_for_port http - HTTP client helpers (http_get, http_request, stream_get) rtp - RTP packet crafting and MulticastSender r2h_process - R2HProcess server wrapper @@ -56,6 +56,7 @@ ipv6_loopback_available, wait_for_port, wait_for_unix_socket, + worker_port_range, ) from .r2h_process import R2HProcess, make_m3u_rtsp_config from .rtp import MulticastSender, make_rtp_packet @@ -100,5 +101,6 @@ "wait_for_port", "wait_for_status_payload", "wait_for_unix_socket", + "worker_port_range", "write_temp_file", ] diff --git a/e2e/helpers/ports.py b/e2e/helpers/ports.py index d8e2e372..27cf3219 100644 --- a/e2e/helpers/ports.py +++ b/e2e/helpers/ports.py @@ -1,63 +1,185 @@ -"""Port allocation helpers for E2E tests.""" +"""Port allocation helpers for E2E tests. + +``bind(0)`` draws from the kernel ephemeral range (Linux typically 32768-60999, +macOS 49152-65535). That races with pytest-xdist workers: a port returned by +``bind(0)`` can be reused as another worker's client source port before +rtp2httpd binds it. CI failures such as ``rtp2httpd did not start on port +49234`` were almost all macOS ephemeral ports. + +Listen ports are therefore taken from a per-xdist-worker range below those +ephemeral windows. A short cooldown avoids immediately recycling a port that +may still be in TIME_WAIT after the previous process exits. +""" from __future__ import annotations +import os import socket +import threading import time +from collections import deque +# Below Linux ip_local_port_range (32768+) and macOS net.inet.ip.portrange (49152+). +_RANGE_BASE = 14000 +_RANGE_SIZE = 2300 +_MAX_RANGES = 8 # 14000-32399 even with more than 8 xdist workers +_COOLDOWN = 64 + +_lock = threading.Lock() +_recent_tcp: deque[int] = deque(maxlen=_COOLDOWN) +_recent_udp: deque[int] = deque(maxlen=_COOLDOWN) +_tcp_cursor = 0 +_udp_cursor = 0 -def find_free_port(host: str = "127.0.0.1") -> int: - """Find a free TCP port on *host* (use "::1" for IPv6 loopback).""" - family = socket.AF_INET6 if ":" in host else socket.AF_INET - with socket.socket(family, socket.SOCK_STREAM) as s: - s.bind((host, 0)) - return s.getsockname()[1] + +def xdist_worker_index() -> int: + """Return this process's pytest-xdist worker index (0 for serial runs).""" + worker = os.environ.get("PYTEST_XDIST_WORKER", "master") + if worker == "master": + return 0 + if worker.startswith("gw") and worker[2:].isdigit(): + return int(worker[2:]) + 1 + return 0 + + +def worker_port_range() -> tuple[int, int]: + """Return the inclusive-exclusive TCP/UDP listen range for this worker.""" + start = _RANGE_BASE + (xdist_worker_index() % _MAX_RANGES) * _RANGE_SIZE + return start, start + _RANGE_SIZE def ipv6_loopback_available() -> bool: """Return True when binding a TCP socket to ::1 works on this host.""" try: with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: + s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) s.bind(("::1", 0)) return True except OSError: return False +def _tcp_probe_targets(host: str) -> list[tuple[int, str]]: + """Addresses whose bind success means rtp2httpd can listen here. + + Tests usually wait on 127.0.0.1, but rtp2httpd binds 0.0.0.0 (and ::). + Probe the IPv4 wildcard so we do not hand out a port that is only free + on the loopback address. + """ + if host in ("127.0.0.1", "0.0.0.0", ""): + return [(socket.AF_INET, "0.0.0.0")] + if ":" in host: + return [(socket.AF_INET6, host)] + return [(socket.AF_INET, host)] + + +def _can_bind_tcp(host: str, port: int) -> bool: + sockets: list[socket.socket] = [] + try: + for family, bind_host in _tcp_probe_targets(host): + sock = socket.socket(family, socket.SOCK_STREAM) + sockets.append(sock) + if family == socket.AF_INET6: + sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) + sock.bind((bind_host, port)) + return True + except OSError: + return False + finally: + for sock in sockets: + sock.close() + + +def _can_bind_udp(port: int) -> bool: + try: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.bind(("", port)) + return True + except OSError: + return False + + +def find_free_port(host: str = "127.0.0.1") -> int: + """Find a free TCP port on *host* (use "::1" for IPv6 loopback).""" + start, end = worker_port_range() + size = end - start + with _lock: + global _tcp_cursor + for step in range(size): + port = start + ((_tcp_cursor + step) % size) + if port in _recent_tcp: + continue + if _can_bind_tcp(host, port): + _tcp_cursor = (_tcp_cursor + step + 1) % size + _recent_tcp.append(port) + return port + raise RuntimeError(f"No free TCP port in {start}-{end - 1} for {host}") + + def find_free_udp_port() -> int: """Find a free UDP port.""" - with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: - s.bind(("", 0)) - return s.getsockname()[1] + start, end = worker_port_range() + size = end - start + with _lock: + global _udp_cursor + for step in range(size): + port = start + ((_udp_cursor + step) % size) + if port in _recent_udp: + continue + if _can_bind_udp(port): + _udp_cursor = (_udp_cursor + step + 1) % size + _recent_udp.append(port) + return port + raise RuntimeError(f"No free UDP port in {start}-{end - 1}") def find_free_udp_port_pair() -> tuple[int, int]: """Find a free even/odd UDP port pair (for RTP/RTCP).""" - for _ in range(100): - with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s1: - s1.bind(("", 0)) - p = s1.getsockname()[1] - if p % 2 != 0: + start, end = worker_port_range() + size = end - start + with _lock: + global _udp_cursor + for step in range(size): + port = start + ((_udp_cursor + step) % size) + if port % 2 != 0 or port + 1 >= end: continue - try: - with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s2: - s2.bind(("", p + 1)) - return p, p + 1 - except OSError: + if port in _recent_udp or (port + 1) in _recent_udp: continue - # fallback - return find_free_udp_port(), find_free_udp_port() + if _can_bind_udp(port) and _can_bind_udp(port + 1): + _udp_cursor = (_udp_cursor + step + 2) % size + _recent_udp.append(port) + _recent_udp.append(port + 1) + return port, port + 1 + raise RuntimeError(f"No free UDP port pair in {start}-{end - 1}") + + +def port_connectable(port: int, host: str = "127.0.0.1", timeout: float = 0.05) -> bool: + """Return True if *host:port* accepts a TCP connection right now.""" + try: + with socket.create_connection((host, port), timeout=timeout): + return True + except OSError: + return False + + +def unix_socket_connectable(path: str, timeout: float = 0.05) -> bool: + """Return True if *path* accepts a Unix stream connection right now.""" + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.settimeout(timeout) + sock.connect(path) + return True + except OSError: + return False def wait_for_port(port: int, host: str = "127.0.0.1", timeout: float = 5.0) -> bool: """Block until *port* is accepting TCP connections (or *timeout* expires).""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: - try: - with socket.create_connection((host, port), timeout=0.5): - return True - except TimeoutError, ConnectionRefusedError, OSError: - time.sleep(0.05) + if port_connectable(port, host=host, timeout=0.5): + return True + time.sleep(0.05) return False @@ -65,11 +187,7 @@ def wait_for_unix_socket(path: str, timeout: float = 5.0) -> bool: """Block until *path* is accepting Unix stream socket connections.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: - try: - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: - sock.settimeout(0.5) - sock.connect(path) - return True - except OSError: - time.sleep(0.05) + if unix_socket_connectable(path, timeout=0.5): + return True + time.sleep(0.05) return False diff --git a/e2e/helpers/r2h_process.py b/e2e/helpers/r2h_process.py index f861ae2b..497481f4 100644 --- a/e2e/helpers/r2h_process.py +++ b/e2e/helpers/r2h_process.py @@ -5,9 +5,10 @@ import os import subprocess import tempfile +import time from pathlib import Path -from .ports import wait_for_port, wait_for_unix_socket +from .ports import port_connectable, unix_socket_connectable def make_m3u_rtsp_config(r2h_port: int, rtsp_port: int, channel_name: str, configured_url_query: str = "") -> str: @@ -47,30 +48,18 @@ def __init__( def start(self, wait: bool = True) -> None: args = self._build_args() - if self.capture_log: - log_fd, self._log_path = tempfile.mkstemp(suffix=".log", prefix="r2h_log_") - self._log_handle = os.fdopen(log_fd, "w") + log_fd, self._log_path = tempfile.mkstemp(suffix=".log", prefix="r2h_log_") + self._log_handle = os.fdopen(log_fd, "w") + try: self.process = subprocess.Popen(args, stdout=self._log_handle, stderr=self._log_handle) - else: - self.process = subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + except OSError as exc: + raise RuntimeError(f"failed to spawn rtp2httpd: {exc}.\nCommand: {' '.join(args)}") from exc if wait: - if self.wait_socket_path: - if not wait_for_unix_socket(self.wait_socket_path, timeout=6.0): - self.stop() - raise RuntimeError( - "rtp2httpd did not start on Unix socket {}.\nCommand: {}".format( - self.wait_socket_path, " ".join(args) - ) - ) - elif self.listen and self.listen.startswith("/"): - if not wait_for_unix_socket(self.listen, timeout=6.0): - self.stop() - raise RuntimeError( - "rtp2httpd did not start on Unix socket {}.\nCommand: {}".format(self.listen, " ".join(args)) - ) - elif self.port is not None and not wait_for_port(self.port, timeout=6.0): + error = self._wait_until_ready() + if error: + detail = self._startup_failure_detail(args, error) self.stop() - raise RuntimeError(f"rtp2httpd did not start on port {self.port}.\nCommand: {' '.join(args)}") + raise RuntimeError(detail) def stop(self) -> None: if self.process and self.process.poll() is None: @@ -98,12 +87,60 @@ def stop(self) -> None: def read_log(self) -> str: """Return the captured rtp2httpd stdout/stderr.""" - assert self._log_path is not None, "read_log requires capture_log=True at construction time" + assert self._log_path is not None, "read_log requires the process to have been started" + if self._log_handle is not None: + self._log_handle.flush() with open(self._log_path) as f: return f.read() # -- internals ----------------------------------------------------------- + def _ready_timeout(self) -> float: + return 10.0 if os.environ.get("CI") else 6.0 + + def _is_listening(self) -> bool: + if self.wait_socket_path: + return unix_socket_connectable(self.wait_socket_path) + if self.listen and self.listen.startswith("/"): + return unix_socket_connectable(self.listen) + if self.port is not None: + return port_connectable(self.port) + return True + + def _wait_until_ready(self) -> str | None: + """Return None when ready, or a short reason if startup failed.""" + assert self.process is not None + timeout = self._ready_timeout() + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if self.process.poll() is not None: + return f"process exited with code {self.process.returncode}" + if self._is_listening(): + return None + time.sleep(0.05) + return f"timed out after {timeout:.1f}s waiting for listen" + + def _startup_failure_detail(self, args: list[str], error: str) -> str: + if self.wait_socket_path: + target = f"Unix socket {self.wait_socket_path}" + elif self.listen and self.listen.startswith("/"): + target = f"Unix socket {self.listen}" + elif self.port is not None: + target = f"port {self.port}" + else: + target = "the configured listener" + log = "" + try: + log = self.read_log().strip() + except OSError: + pass + if len(log) > 4000: + log = log[-4000:] + detail = f"rtp2httpd did not start on {target}: {error}.\nCommand: {' '.join(args)}" + if log: + detail += f"\n--- rtp2httpd log ---\n{log}\n--- end log ---" + return detail + def _build_args(self) -> list[str]: if self.config_content is not None: fd, path = tempfile.mkstemp(suffix=".conf", prefix="r2h_test_") diff --git a/e2e/test_pages.py b/e2e/test_pages.py index 9e8ca472..34559471 100644 --- a/e2e/test_pages.py +++ b/e2e/test_pages.py @@ -487,17 +487,31 @@ def test_manifest_updates_after_config_reload(self, r2h_binary): assert r2h.process is not None os.kill(r2h.process.pid, signal.SIGHUP) - deadline = time.monotonic() + 5 + deadline = time.monotonic() + 8 + last_status = None + last_error = None + body = b"" while True: - status, _, body = http_get( - "127.0.0.1", - port, - "/new/status.webmanifest?r2h-token=new-token", - ) + try: + status, _, body = http_get( + "127.0.0.1", + port, + "/new/status.webmanifest?r2h-token=new-token", + ) + last_status = status + last_error = None + except (OSError, TimeoutError) as exc: + # Workers briefly drop the listen socket during SIGHUP reload. + last_status = None + last_error = exc + status = 0 if status == 200: break if time.monotonic() >= deadline: - raise AssertionError(f"reloaded manifest did not become available; last status was {status}") + raise AssertionError( + f"reloaded manifest did not become available; last status was {last_status}; " + f"last error: {last_error}" + ) time.sleep(0.05) manifest = _parse_manifest(body) diff --git a/e2e/test_port_helpers.py b/e2e/test_port_helpers.py new file mode 100644 index 00000000..b5976922 --- /dev/null +++ b/e2e/test_port_helpers.py @@ -0,0 +1,71 @@ +"""Unit coverage for listen-port allocation. + +These tests do not start rtp2httpd. They guard the ephemeral-port race that +caused CI failures such as ``rtp2httpd did not start on port 49234``. +""" + +import os +import socket + +import pytest +from helpers import ( + R2HProcess, + find_free_port, + find_free_udp_port, + find_free_udp_port_pair, + wait_for_port, + worker_port_range, +) + + +def test_find_free_port_stays_in_worker_range(): + start, end = worker_port_range() + port = find_free_port() + assert start <= port < end + assert port < 32768 + + +def test_find_free_port_does_not_reuse_immediately(): + first = find_free_port() + second = find_free_port() + assert first != second + + +def test_find_free_port_is_bindable_on_wildcard(): + port = find_free_port() + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("0.0.0.0", port)) + sock.listen(1) + assert wait_for_port(port, timeout=1.0) + + +def test_find_free_udp_port_stays_in_worker_range(): + start, end = worker_port_range() + port = find_free_udp_port() + assert start <= port < end + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.bind(("", port)) + + +def test_find_free_udp_port_pair_is_adjacent_and_bindable(): + start, end = worker_port_range() + even, odd = find_free_udp_port_pair() + assert even % 2 == 0 + assert odd == even + 1 + assert start <= even < odd < end + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s1, socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s2: + s1.bind(("", even)) + s2.bind(("", odd)) + + +def test_r2h_start_fail_fast_includes_exit_code(tmp_path): + # FreeBSD has no /bin/false; use a portable immediate-exit script. + script = tmp_path / "immediate_exit" + script.write_text("#!/bin/sh\nexit 7\n") + os.chmod(script, 0o755) + r2h = R2HProcess(script, find_free_port()) + try: + with pytest.raises(RuntimeError, match="exited with code 7"): + r2h.start() + finally: + r2h.stop()