dotfiles/common/optional/x86-64-v3/patches/gnutls/verify-readiness.py

#!/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()