mirror of
https://github.com/varun-r-mallya/py-libp2p.git
synced 2025-12-31 20:36:24 +00:00
Add tests for echo example and address validation utilities
- Introduced `test_echo_thin_waist.py` to validate the echo example's output for Thin Waist lines. - Added `test_address_validation.py` to cover functions for available interfaces, optimal binding addresses, and wildcard address expansion. - Included parameterized tests and environment checks for IPv6 support.
This commit is contained in:
51
tests/examples/test_echo_thin_waist.py
Normal file
51
tests/examples/test_echo_thin_waist.py
Normal file
@ -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"
|
||||
56
tests/utils/test_address_validation.py
Normal file
56
tests/utils/test_address_validation.py
Normal file
@ -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)
|
||||
Reference in New Issue
Block a user