#!/usr/bin/env python3 """Exercise patched common.sh without building GnuTLS (Python standard library only). Usage: python3 verify-readiness.py /path/to/patched/tests/scripts/common.sh Set SHELL to test another shell, and NETSTAT to test a netstat outside PATH. """ # Use unittest so this upstream companion tool needs no pytest installation. # ruff: noqa: PT009 import os import shutil import socket import subprocess import sys import tempfile import time import unittest from pathlib import Path COMMON = str(Path(sys.argv.pop(1)).resolve()) SHELL = os.environ.get("SHELL", "/bin/sh") class ReadinessTests(unittest.TestCase): """Check endpoint parsing and the server startup lifecycle.""" def setUp(self) -> None: """Create a socket-listing fixture for each check.""" self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.fixture = self.root / "sockets" self.fixture.write_text("") self.finder = self.root / "port-finder" self.finder.write_text('#!/bin/sh\ncat "$SOCKET_FIXTURE"\n') self.finder.chmod(0o755) def run_shell(self, body: str, **env: str) -> subprocess.CompletedProcess[str]: """Source the actual helper and run a shell scenario.""" return subprocess.run( [SHELL, "-c", '. "$COMMON"\n' + body], env={ **os.environ, "COMMON": COMMON, "SOCKET_FIXTURE": str(self.fixture), "PFCMD": str(self.finder), "PORT": "12345", **env, }, capture_output=True, text=True, timeout=20, check=False, ) def test_socket_formats_and_false_matches(self) -> None: """Accept IPv4 UDP local endpoints and reject unrelated sockets.""" cases = [ ("udp UNCONN 0 0 0.0.0.0:12345 0.0.0.0:*", True), ("udp UNCONN 0 0 127.0.0.1:12345 0.0.0.0:*", True), ("udp 0 0 0.0.0.0:12345 0.0.0.0:*", True), ("udp4 0 0 *.12345 *.*", True), ("udp 0 0 127.0.0.1.12345 *.*", True), ("udp 0 0 *.12345 *.*", True), ("udp UNCONN 0 0 0.0.0.0:123456 0.0.0.0:*", False), ("udp 0 0 0.0.0.0:123456 0.0.0.0:*", False), ("udp ESTAB 0 0 127.0.0.1:54321 127.0.0.1:12345", False), ("udp 0 0 127.0.0.1:54321 127.0.0.1:12345", False), ("tcp LISTEN 0 128 0.0.0.0:12345 0.0.0.0:*", False), ("tcp 0 0 0.0.0.0:12345 0.0.0.0:* LISTEN", False), ("udp UNCONN 0 0 [::]:12345 [::]:*", False), ("udp UNCONN 0 0 *:12345 *:*", False), ("udp6 0 0 :::12345 :::*", False), ("udp6 0 0 *.12345 *.*", False), ("", False), ] for row, ready in cases: with self.subTest(row=row): self.fixture.write_text(row + "\n") result = self.run_shell('check_if_udp_port_bound "$PORT"') self.assertEqual(result.returncode, 0 if ready else 1, result.stderr) def test_exited_server_fails_immediately(self) -> None: """Fail without sleeping when the server has already exited.""" result = self.run_shell( 'true &\npid=$!\nwait "$pid"\nsleep() { echo "unexpected sleep" >&2; }\nwait_udp_server "$pid"' ) self.assertEqual(result.returncode, 1) self.assertIn("exited before binding", result.stderr) self.assertNotIn("unexpected sleep", result.stderr) def test_ready_socket_does_not_sleep(self) -> None: """Check readiness before the first sleep.""" self.fixture.write_text("udp UNCONN 0 0 0.0.0.0:12345 0.0.0.0:*\n") result = self.run_shell('sleep() { echo "unexpected sleep" >&2; }\nwait_udp_server "$$"') self.assertEqual(result.returncode, 0, result.stderr) self.assertNotIn("unexpected sleep", result.stderr) def test_missing_port_finders_skip(self) -> None: """Skip and stop the live server when no finder is available.""" # have_port_finder also tries these paths independently of PATH. if any(os.access(f"{directory}/ss", os.X_OK) for directory in ("/sbin", "/usr/sbin", "/usr/local/sbin")): self.skipTest("an absolute ss path cannot be hidden by this PATH-only fixture") with subprocess.Popen(["sleep", "60"]) as server: try: result = self.run_shell( 'unset PFCMD\nPATH=/nonexistent\nwait_udp_server "$SERVER_PID"', SERVER_PID=str(server.pid), ) self.assertEqual(result.returncode, 77) self.assertIn("neither ss nor netstat found", result.stderr) server.wait(timeout=3) self.assertLess(server.returncode, 0) finally: if server.poll() is None: server.kill() def test_timeout_is_bounded_and_cleans_up(self) -> None: """Stop polling after the retry budget and terminate the server.""" # Only accelerate the polling delay; keep a real live server process. self.finder.write_text('#!/bin/sh\necho probe >&2\ncat "$SOCKET_FIXTURE"\n') with subprocess.Popen(["sleep", "60"]) as server: try: result = self.run_shell( 'sleep() { echo polling-sleep; }\nwait_udp_server "$SERVER_PID"', SERVER_PID=str(server.pid), ) self.assertEqual(result.returncode, 1) self.assertIn("did not come up", result.stderr) self.assertEqual(result.stderr.count("probe\n"), 90) self.assertEqual(result.stdout.count("polling-sleep"), 89) server.wait(timeout=3) self.assertLess(server.returncode, 0) finally: if server.poll() is None: server.kill() def test_server_exits_while_waiting(self) -> None: """Detect a startup failure that happens after polling begins.""" result = self.run_shell('sleep 1 &\npid=$!\nwait_udp_server "$pid"') self.assertEqual(result.returncode, 1) self.assertIn("exited before binding", result.stderr) self.assertIn("waiting for UDP port", result.stdout) def test_real_socket_delayed_beyond_four_seconds(self) -> None: """Wait for a real delayed bind with each installed port finder.""" finders = [shutil.which("ss"), os.environ.get("NETSTAT") or shutil.which("netstat")] finders = [finder for finder in finders if finder] if not finders: self.skipTest("neither ss nor netstat available") for finder in finders: with self.subTest(finder=finder): with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] code = ( "import socket,time,sys; time.sleep(6); " "s=socket.socket(socket.AF_INET,socket.SOCK_DGRAM); " "s.bind(('127.0.0.1',int(sys.argv[1]))); time.sleep(30)" ) with subprocess.Popen([sys.executable, "-c", code, str(port)]) as server: try: started = time.monotonic() result = self.run_shell( 'wait_udp_server "$SERVER_PID"', SERVER_PID=str(server.pid), PORT=str(port), PFCMD=finder, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertGreaterEqual(time.monotonic() - started, 6) self.assertIsNone(server.poll()) finally: server.terminate() server.wait(timeout=3) if __name__ == "__main__": unittest.main()