andreasandClaude Sonnet 5
997279dce8
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Uc9V1v7iXruubpoLiSE24M
307 lines
9.6 KiB
Python
307 lines
9.6 KiB
Python
"""Connectivity probe in the standalone scripts/hbc_mini.py (stdlib-only single-file client)."""
|
|
import asyncio
|
|
import importlib.util
|
|
import pathlib
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
_PATH = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "hbc_mini.py"
|
|
_SPEC = importlib.util.spec_from_file_location("hbc_mini", _PATH)
|
|
mini = importlib.util.module_from_spec(_SPEC)
|
|
sys.modules["hbc_mini"] = mini # dataclasses resolve their module by name
|
|
_SPEC.loader.exec_module(mini)
|
|
|
|
PROC_V4 = (
|
|
"Iface\tDestination\tGateway \tFlags\tRefCnt\tUse\tMetric\tMask\t\tMTU\tWindow\tIRTT\n"
|
|
"wlan0\t00000000\t0100A8C0\t0003\t0\t0\t600\t00000000\t0\t0\t0\n"
|
|
"eth0\t00000000\t0101A8C0\t0003\t0\t0\t100\t00000000\t0\t0\t0\n"
|
|
"wlan0\t0000A8C0\t00000000\t0001\t0\t0\t600\t00FFFFFF\t0\t0\t0\n"
|
|
)
|
|
PROC_V6 = (
|
|
"00000000000000000000000000000000 00 00000000000000000000000000000000 00 "
|
|
"fe800000000000000000000000000001 00000400 00000001 00000000 00000003 wlan0\n"
|
|
"00000000000000000000000000000000 00 00000000000000000000000000000000 00 "
|
|
"00000000000000000000000000000000 ffffffff 00000001 00000000 00200200 lo\n"
|
|
"fe800000000000000000000000000000 40 00000000000000000000000000000000 00 "
|
|
"00000000000000000000000000000000 00000100 00000001 00000000 00000001 wlan0\n"
|
|
)
|
|
ROUTE_GET = """ route to: 0.0.0.0
|
|
destination: 0.0.0.0
|
|
mask: 0.0.0.0
|
|
gateway: 192.0.2.1
|
|
fib: 0
|
|
interface: vtnet0
|
|
"""
|
|
WIN_V4 = """IPv4 Route Table
|
|
Active Routes:
|
|
Network Destination Netmask Gateway Interface Metric
|
|
0.0.0.0 0.0.0.0 192.168.1.1 192.168.1.50 25
|
|
0.0.0.0 0.0.0.0 10.0.0.1 10.0.0.9 50
|
|
127.0.0.0 255.0.0.0 On-link 127.0.0.1 331
|
|
"""
|
|
WIN_V6 = """IPv6 Route Table
|
|
Active Routes:
|
|
If Metric Network Destination Gateway
|
|
12 281 ::/0 fe80::1
|
|
1 331 ::1/128 On-link
|
|
"""
|
|
LINUX_PING = """PING 1.1.1.1 (1.1.1.1) 56(84) bytes of data.
|
|
64 bytes from 1.1.1.1: icmp_seq=1 ttl=57 time=11.2 ms
|
|
|
|
--- 1.1.1.1 ping statistics ---
|
|
1 packets transmitted, 1 received, 0% packet loss, time 0ms
|
|
rtt min/avg/max/mdev = 11.234/11.234/11.234/0.000 ms
|
|
"""
|
|
FREEBSD_PING = """PING 1.1.1.1 (1.1.1.1): 56 data bytes
|
|
64 bytes from 1.1.1.1: icmp_seq=0 ttl=59 time=4.006 ms
|
|
|
|
--- 1.1.1.1 ping statistics ---
|
|
1 packets transmitted, 1 packets received, 0.0% packet loss, 1 packets out of wait time
|
|
round-trip min/avg/max/stddev = 4.006/4.006/4.006/0.000 ms
|
|
"""
|
|
WIN_PING = """Pinging 1.1.1.1 with 32 bytes of data:
|
|
Reply from 1.1.1.1: bytes=32 time=12ms TTL=57
|
|
|
|
Ping statistics for 1.1.1.1:
|
|
Packets: Sent = 1, Received = 1, Lost = 0 (0% loss),
|
|
"""
|
|
LOST_PING = """--- 1.1.1.1 ping statistics ---
|
|
1 packets transmitted, 0 received, 100% packet loss, time 0ms
|
|
"""
|
|
|
|
|
|
# ---- gateway parsers ---------------------------------------------------------------
|
|
|
|
def test_proc_v4_lowest_metric_wins():
|
|
assert mini._parse_proc_route_v4(PROC_V4) == "192.168.1.1" # eth0, metric 100
|
|
|
|
|
|
def test_proc_v4_none_without_default_route():
|
|
assert mini._parse_proc_route_v4(PROC_V4.splitlines()[0] + "\n") is None
|
|
|
|
|
|
def test_proc_v6_skips_reject_and_non_default():
|
|
assert mini._parse_proc_route_v6(PROC_V6) == "fe80::1"
|
|
|
|
|
|
def test_route_get_bsd_and_macos():
|
|
assert mini._parse_route_get(ROUTE_GET) == "192.0.2.1"
|
|
assert mini._parse_route_get("route: writing to routing socket: not in table") is None
|
|
|
|
|
|
def test_windows_routes():
|
|
assert mini._parse_windows_route(WIN_V4, "ipv4") == "192.168.1.1"
|
|
assert mini._parse_windows_route(WIN_V6, "ipv6") == "fe80::1"
|
|
|
|
|
|
def test_default_gateway_never_raises(monkeypatch):
|
|
monkeypatch.setattr(sys, "platform", "linux")
|
|
monkeypatch.setattr(mini, "_parse_proc_route_v4", lambda t: 1 / 0)
|
|
assert asyncio.run(mini._default_gateway("ipv4")) is None
|
|
|
|
|
|
# ---- ping ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("platform,family,expected", [
|
|
("linux", "ipv4", ["ping", "-c", "1", "-W", "2", "-4", "h"]),
|
|
("linux", "ipv6", ["ping", "-c", "1", "-W", "2", "-6", "h"]),
|
|
("freebsd15", "ipv4", ["ping", "-c", "1", "-W", "2000", "h"]),
|
|
("freebsd15", "ipv6", ["ping6", "-c", "1", "h"]),
|
|
("darwin", "ipv4", ["ping", "-c", "1", "-W", "2000", "h"]),
|
|
("darwin", "ipv6", ["ping6", "-c", "1", "h"]),
|
|
("win32", "ipv4", ["ping", "-n", "1", "-w", "2000", "-4", "h"]),
|
|
("win32", "ipv6", ["ping", "-n", "1", "-w", "2000", "-6", "h"]),
|
|
])
|
|
def test_probe_ping_command(monkeypatch, platform, family, expected):
|
|
monkeypatch.setattr(sys, "platform", platform)
|
|
assert mini._probe_ping_command("h", family) == expected
|
|
|
|
|
|
def test_parse_ping_rtt():
|
|
assert mini._parse_ping_rtt(LINUX_PING) == 11.234
|
|
assert mini._parse_ping_rtt(FREEBSD_PING) == 4.006
|
|
assert mini._parse_ping_rtt(WIN_PING) == 12.0
|
|
assert mini._parse_ping_rtt(LOST_PING) is None
|
|
assert mini._parse_ping_rtt("ping: connect: Network is unreachable") is None
|
|
|
|
|
|
# ---- probe state machine ------------------------------------------------------------
|
|
|
|
class Net:
|
|
"""Scriptable network: addr None = no route, 'unavailable' = family unsupported."""
|
|
|
|
def __init__(self):
|
|
self.addr = {"ipv4": "192.0.2.5", "ipv6": "2001:db8::5"}
|
|
self.gw = {"ipv4": "192.0.2.1", "ipv6": "fe80::1"}
|
|
self.rtt = {"ipv4": 10.0, "ipv6": 20.0} # None = ping fails, 'unavailable' = no ping binary
|
|
self.pings = []
|
|
self.now = 1000.0
|
|
|
|
def local(self, fam, target):
|
|
a = self.addr[fam]
|
|
if a == "unavailable":
|
|
raise mini._ProbeUnavailable("no such family")
|
|
return a
|
|
|
|
async def ping(self, target, fam):
|
|
self.pings.append(fam)
|
|
r = self.rtt[fam]
|
|
if r == "unavailable":
|
|
raise mini._PingUnavailable("no ping")
|
|
return r
|
|
|
|
async def gateway(self, fam):
|
|
return self.gw[fam]
|
|
|
|
def clock(self):
|
|
return self.now
|
|
|
|
|
|
def make(net, **kw):
|
|
c = mini._Connectivity(local_address_fn=net.local, ping_fn=net.ping, gateway_fn=net.gateway,
|
|
clock=net.clock, **kw)
|
|
c._cheap_gateway = True
|
|
return c
|
|
|
|
|
|
def refresh(c):
|
|
asyncio.run(c.refresh())
|
|
|
|
|
|
def test_present_with_addr_gateway_rtt():
|
|
net = Net()
|
|
c = make(net)
|
|
refresh(c)
|
|
s = c.snapshot()["ipv4"]
|
|
assert (s.present, s.addr, s.gateway, s.rtt) == (True, "192.0.2.5", "192.0.2.1", 10.0)
|
|
assert c.flags() == {"p4": True, "p6": True}
|
|
assert c.as_metric()["ipv6"]["gateway"] == "fe80::1"
|
|
|
|
|
|
def test_no_route_is_immediately_absent():
|
|
net = Net()
|
|
net.addr["ipv6"] = None
|
|
c = make(net)
|
|
refresh(c)
|
|
assert c.flags() == {"p4": True, "p6": False}
|
|
assert net.pings == ["ipv4"]
|
|
assert c.snapshot()["ipv6"].addr is None
|
|
|
|
|
|
def test_ping_is_debounced_and_rate_limited():
|
|
net = Net()
|
|
c = make(net, probe_interval=60.0, retry_interval=10.0)
|
|
refresh(c)
|
|
refresh(c)
|
|
assert net.pings == ["ipv4", "ipv6"]
|
|
net.now += 60
|
|
net.rtt["ipv6"] = None
|
|
refresh(c)
|
|
assert c.flags()["p6"] is True
|
|
net.now += 10
|
|
refresh(c)
|
|
assert c.flags()["p6"] is False
|
|
|
|
|
|
def test_first_probe_failing_once_is_unknown():
|
|
net = Net()
|
|
net.rtt["ipv4"] = None
|
|
c = make(net)
|
|
refresh(c)
|
|
assert "p4" not in c.flags() and c.flags()["p6"] is True
|
|
|
|
|
|
def test_recovers_on_first_success():
|
|
net = Net()
|
|
net.rtt["ipv6"] = None
|
|
c = make(net, retry_interval=10.0)
|
|
refresh(c)
|
|
net.now += 10
|
|
refresh(c)
|
|
assert c.flags()["p6"] is False
|
|
net.rtt["ipv6"] = 25.0
|
|
net.now += 10
|
|
refresh(c)
|
|
assert c.flags()["p6"] is True
|
|
|
|
|
|
def test_address_change_invalidates_cache():
|
|
net = Net()
|
|
c = make(net)
|
|
refresh(c)
|
|
net.addr["ipv4"] = "198.51.100.7"
|
|
refresh(c)
|
|
assert net.pings.count("ipv4") == 2
|
|
assert c.snapshot()["ipv4"].addr == "198.51.100.7"
|
|
|
|
|
|
def test_gateway_change_invalidates_cache():
|
|
net = Net()
|
|
c = make(net)
|
|
refresh(c)
|
|
net.gw["ipv4"] = "192.0.2.254"
|
|
refresh(c)
|
|
assert net.pings.count("ipv4") == 2
|
|
|
|
|
|
def test_route_loss_then_return_pings_immediately():
|
|
net = Net()
|
|
c = make(net)
|
|
refresh(c)
|
|
net.addr["ipv6"] = None
|
|
refresh(c)
|
|
net.addr["ipv6"] = "2001:db8::5"
|
|
refresh(c)
|
|
assert net.pings.count("ipv6") == 2 and c.flags()["p6"] is True
|
|
|
|
|
|
def test_missing_ping_binary_leaves_presence_unknown_and_stops_retrying():
|
|
net = Net()
|
|
net.rtt["ipv4"] = "unavailable"
|
|
c = make(net)
|
|
refresh(c)
|
|
net.now += 600
|
|
refresh(c)
|
|
assert "p4" not in c.flags()
|
|
assert net.pings.count("ipv4") == 1
|
|
|
|
|
|
def test_ping_less_family_stays_unknown_after_route_returns():
|
|
net = Net()
|
|
net.rtt["ipv4"] = "unavailable"
|
|
c = make(net)
|
|
refresh(c)
|
|
assert "p4" not in c.flags()
|
|
net.addr["ipv4"] = None
|
|
refresh(c)
|
|
assert c.flags()["p4"] is False # no route is definitive
|
|
net.addr["ipv4"] = "192.0.2.5"
|
|
refresh(c)
|
|
assert "p4" not in c.flags() and net.pings.count("ipv4") == 1
|
|
|
|
|
|
def test_unsupported_family_is_unknown():
|
|
net = Net()
|
|
net.addr["ipv6"] = "unavailable"
|
|
c = make(net)
|
|
refresh(c)
|
|
assert "p6" not in c.flags() and c.flags()["p4"] is True
|
|
|
|
|
|
def test_from_config():
|
|
c = mini._Connectivity.from_config({"probe_v4": "9.9.9.9", "probe_interval": 30})
|
|
assert c.probes["ipv4"] == "9.9.9.9"
|
|
assert c.probes["ipv6"] == "2606:4700:4700::1111"
|
|
assert c.probe_interval == 30.0
|
|
|
|
|
|
def test_local_address_on_this_host_never_raises():
|
|
# Real UDP connect(): either an address string, None (no route) or _ProbeUnavailable.
|
|
for fam, target in (("ipv4", "1.1.1.1"), ("ipv6", "2606:4700:4700::1111")):
|
|
try:
|
|
r = mini._local_address(fam, target)
|
|
except mini._ProbeUnavailable:
|
|
continue
|
|
assert r is None or isinstance(r, str)
|