andreasandClaude Sonnet 5
2f1cad9156
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Uc9V1v7iXruubpoLiSE24M
95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
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 == {}
|