import time import pytest from hbd.server import hbdclass from hbd.server.hbdclass import Connection as C def _host(name): hbdclass.Host.hosts.pop(name, None) return hbdclass.Host(name) def _conn(host, afam, state): addr = "192.0.2.1" if afam == "IPv4" else "2001:db8::1" conn, _ = host.conndata(0, addr, None, 1000.0) conn.state = state return conn CASES = [ # v4 state, v6 state, presence flags, expected status (C.UP, C.UP, {}, "up"), (C.UP, C.OVERDUE, {}, "simplex"), # both present (legacy) => simplex (C.UP, C.OVERDUE, {"IPv6": True}, "simplex"), (C.UP, C.OVERDUE, {"IPv6": False}, "up"), # v6 not present => not monitored (C.OVERDUE, C.OVERDUE, {}, "overdue"), (C.OVERDUE, C.ABSENT, {"IPv6": False}, "overdue"), # only present family is overdue (C.UP, C.ABSENT, {"IPv6": False}, "up"), (C.DOWN, C.UP, {}, "down"), (C.UNKNOWN, C.UNKNOWN, {}, "unknown"), (C.OVERDUE, C.UNKNOWN, {}, "overdue"), (C.UP, C.UNKNOWN, {}, "up"), (C.OVERDUE, C.UP, {"IPv6": False}, "simplex"), # HTB arriving => present despite flag ] @pytest.mark.parametrize("s4,s6,flags,expected", CASES) def test_status_matrix(s4, s6, flags, expected): h = _host("status-matrix") _conn(h, "IPv4", s4) _conn(h, "IPv6", s6) for afam, present in flags.items(): h.set_presence(afam, present, time.time()) assert h.status() == expected def test_status_single_family_host(): h = _host("status-v4only") _conn(h, "IPv4", C.UP) assert h.status() == "up" h.connections["IPv4"].state = C.OVERDUE assert h.status() == "overdue" def test_status_no_connections_is_unknown(): assert _host("status-empty").status() == "unknown" def test_family_present_rules(): h = _host("fp") v6 = _conn(h, "IPv6", C.OVERDUE) assert h.family_present("IPv6") is True # never reported => present h.set_presence("IPv6", False, 1.0) assert h.family_present("IPv6") is False v6.state = C.UP # real traffic overrides the flag assert h.family_present("IPv6") is True assert h.flag_present("IPv6") is False def test_stateinfo_and_statedict_expose_status(): h = _host("status-exposed") _conn(h, "IPv4", C.UP) assert h.stateinfo()["status"] == "up" assert h.statedict()["status"] == "up" def test_absent_connection_statedict_is_neutral(): h = _host("absent-dict") conn = _conn(h, "IPv6", C.ABSENT) conn.rtts.append(42.0) d = conn.statedict() assert d["state"] == "absent" and d["rtt"] == "" and d["rttstate"] == "absent" def test_presence_survives_pickle_and_old_pickles_get_default(): import pickle h = _host("pickled") h.set_presence("IPv6", False, 5.0) h2 = pickle.loads(pickle.dumps(h)) assert h2.presence == {"IPv6": {"present": False, "ts": 5.0}} del h2.__dict__["presence"] # simulate a pre-feature pickle h2.fixup() assert h2.presence == {} def _all_absent(name, s4, s6=None): h = _host(name) _conn(h, "IPv4", s4) h.set_presence("IPv4", False, time.time()) if s6 is not None: _conn(h, "IPv6", s6) h.set_presence("IPv6", False, time.time()) return h def test_all_absent_single_family_overdue_is_overdue(): assert _all_absent("abs-single", C.OVERDUE).status() == "overdue" def test_all_absent_dual_stack_one_absent_one_overdue_is_overdue(): assert _all_absent("abs-dual-od", C.ABSENT, C.OVERDUE).status() == "overdue" def test_all_absent_dual_stack_v6_up_is_up(): assert _all_absent("abs-dual-up", C.ABSENT, C.UP).status() == "up" def test_all_absent_all_unknown_is_unknown(): assert _all_absent("abs-unknown", C.UNKNOWN, C.UNKNOWN).status() == "unknown"