✨ Feat: add Thin Waist address validation utilities and integrate into echo example

tests/examples/test_echo_thin_waist.py

@@ -0,0 +1,51 @@
+import asyncio
+import contextlib
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+# This test is intentionally lightweight and can be marked as 'integration'.
+# It ensures the echo example runs and prints the new Thin Waist lines.
+
+EXAMPLES_DIR = Path(__file__).parent.parent.parent / "examples" / "echo"
+
+
+@pytest.mark.timeout(20)
+def test_echo_example_starts_and_prints_thin_waist(monkeypatch, tmp_path):
+    # We run: python examples/echo/echo.py -p 0
+    cmd = [sys.executable, str(EXAMPLES_DIR / "echo.py"), "-p", "0"]
+    proc = subprocess.Popen(
+        cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    assert proc.stdout is not None
+
+    found_selected = False
+    found_interfaces = False
+    start = time.time()
+
+    try:
+        while time.time() - start < 10:
+            line = proc.stdout.readline()
+            if not line:
+                time.sleep(0.1)
+                continue
+            if "Selected binding address:" in line:
+                found_selected = True
+            if "Available candidate interfaces:" in line:
+                found_interfaces = True
+            if "Waiting for incoming connections..." in line:
+                break
+    finally:
+        with contextlib.suppress(ProcessLookupError):
+            proc.terminate()
+        with contextlib.suppress(ProcessLookupError):
+            proc.kill()
+
+    assert found_selected, "Did not capture Thin Waist binding log line"
+    assert found_interfaces, "Did not capture Thin Waist interfaces log line"

tests/utils/test_address_validation.py

@@ -0,0 +1,56 @@
+import os
+
+import pytest
+from multiaddr import Multiaddr
+
+from libp2p.utils.address_validation import (
+    get_available_interfaces,
+    get_optimal_binding_address,
+    expand_wildcard_address,
+)
+
+
+@pytest.mark.parametrize("proto", ["tcp"])
+def test_get_available_interfaces(proto: str) -> None:
+    interfaces = get_available_interfaces(0, protocol=proto)
+    assert len(interfaces) > 0
+    for addr in interfaces:
+        assert isinstance(addr, Multiaddr)
+        assert f"/{proto}/" in str(addr)
+
+
+def test_get_optimal_binding_address() -> None:
+    addr = get_optimal_binding_address(0)
+    assert isinstance(addr, Multiaddr)
+    # At least IPv4 or IPv6 prefix present
+    s = str(addr)
+    assert ("/ip4/" in s) or ("/ip6/" in s)
+
+
+def test_expand_wildcard_address_ipv4() -> None:
+    wildcard = Multiaddr("/ip4/0.0.0.0/tcp/0")
+    expanded = expand_wildcard_address(wildcard)
+    assert len(expanded) > 0
+    for e in expanded:
+        assert isinstance(e, Multiaddr)
+        assert "/tcp/" in str(e)
+
+
+def test_expand_wildcard_address_port_override() -> None:
+    wildcard = Multiaddr("/ip4/0.0.0.0/tcp/7000")
+    overridden = expand_wildcard_address(wildcard, port=9001)
+    assert len(overridden) > 0
+    for e in overridden:
+        assert str(e).endswith("/tcp/9001")
+
+
+@pytest.mark.skipif(
+    os.environ.get("NO_IPV6") == "1",
+    reason="Environment disallows IPv6",
+)
+def test_expand_wildcard_address_ipv6() -> None:
+    wildcard = Multiaddr("/ip6/::/tcp/0")
+    expanded = expand_wildcard_address(wildcard)
+    assert len(expanded) > 0
+    for e in expanded:
+        assert "/ip6/" in str(e)