"""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) # ---- heartbeat message and plugin wiring --------------------------------------------- class _Conn: ackcount = 3 rtts = [0.0, 12.5] class _FlagsProbe: def __init__(self, flags): self._flags = flags def flags(self): return self._flags def test_heartbeat_without_probe_is_unchanged(): assert mini._build_heartbeat_msg(_Conn(), 60) == {"acks": 3, "rtt": 12.5, "interval": 60} def test_heartbeat_carries_presence_flags(): msg = mini._build_heartbeat_msg(_Conn(), 60, _FlagsProbe({"p4": True, "p6": False})) assert msg == {"acks": 3, "rtt": 12.5, "interval": 60, "p4": True, "p6": False} def test_heartbeat_omits_unknown_flags(): msg = mini._build_heartbeat_msg(_Conn(), 60, _FlagsProbe({})) assert "p4" not in msg and "p6" not in msg def test_flags_survive_the_wire_encoding(): msg = mini._build_heartbeat_msg(_Conn(), 60, _FlagsProbe({"p4": True, "p6": False})) out = mini._stodict(mini._dicttos("HTB", dict(msg, name="h", id=1))) assert out["p4"] == 1 and out["p6"] == 0 class _FakeProbe: def as_metric(self): return {"ipv4": {"present": True, "addr": "192.0.2.5", "gateway": "192.0.2.1", "rtt": 9.0}, "ipv6": {"present": False, "addr": None, "gateway": None, "rtt": None}} def test_plugin_reports_connectivity_alongside_interfaces(monkeypatch): monkeypatch.setattr(mini, "_probe", _FakeProbe()) monkeypatch.setattr(mini.platform, "system", lambda: "Linux") plugin = mini.NetworkMonitorPlugin({}) assert asyncio.run(plugin.initialize()) is True data = asyncio.run(plugin._collect_metrics()) assert data["connectivity"]["ipv4"]["present"] is True assert data["connectivity"]["ipv6"]["present"] is False def test_plugin_probe_only_mode_on_non_linux(monkeypatch): monkeypatch.setattr(mini, "_probe", _FakeProbe()) monkeypatch.setattr(mini.platform, "system", lambda: "FreeBSD") plugin = mini.NetworkMonitorPlugin({}) assert asyncio.run(plugin.initialize()) is True assert list(asyncio.run(plugin._collect_metrics())) == ["connectivity"] def test_plugin_still_skips_on_non_linux_without_probe(monkeypatch): monkeypatch.setattr(mini, "_probe", None) monkeypatch.setattr(mini.platform, "system", lambda: "FreeBSD") plugin = mini.NetworkMonitorPlugin({}) assert asyncio.run(plugin.initialize()) is False assert plugin.skip_reason def test_plugin_omits_connectivity_without_probe(monkeypatch): monkeypatch.setattr(mini, "_probe", None) monkeypatch.setattr(mini.platform, "system", lambda: "Linux") plugin = mini.NetworkMonitorPlugin({}) assert asyncio.run(plugin.initialize()) is True assert "connectivity" not in asyncio.run(plugin._collect_metrics())