heartbeat/tests/test_connectivity.py

import asyncio

from hbd.client.connectivity import Connectivity, ProbeUnavailable
from hbd.client.pingutil import PingUnavailable


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 ProbeUnavailable("no such family")
        return a

    async def ping(self, target, fam):
        self.pings.append(fam)
        r = self.rtt[fam]
        if r == "unavailable":
            raise 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 = 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"]          # no ping wasted on a family without a route
    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)                            # same instant: cached, no second ping
    assert net.pings == ["ipv4", "ipv6"]
    net.now += 60
    net.rtt["ipv6"] = None                # first failure: still present (debounce)
    refresh(c)
    assert c.flags()["p6"] is True
    net.now += 10                         # retry_interval elapsed: second failure flips it
    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"     # moved networks, no time has passed
    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_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 = 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