andreasandClaude Sonnet 5
52968201fa
Capture RTT on every heartbeat using the existing plugin_data history mechanism, recording under rtt_ipv4 or rtt_ipv6 keys. Maintains 100-sample retention per host and address family. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DEimzMv4Q5EjFg3hoiZ69T
65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
"""Tests for RTT history capture in udp.py's handle_datagram."""
|
|
from hbd.common.proto import dicttos
|
|
from hbd.server import hbdclass
|
|
from hbd.server.udp import handle_datagram, parse_message
|
|
|
|
|
|
class _FakeTransport:
|
|
def __init__(self):
|
|
self.sent = []
|
|
|
|
def sendto(self, data, addr):
|
|
self.sent.append((data, addr))
|
|
|
|
|
|
def _htb(name, rtt=None, interval=0):
|
|
d = {"name": name, "interval": interval, "id": 0}
|
|
if rtt is not None:
|
|
d["rtt"] = rtt
|
|
return parse_message(dicttos("HTB", d))
|
|
|
|
|
|
def _base_ctx():
|
|
return {
|
|
"config": {},
|
|
"hbdclass": hbdclass,
|
|
"msg_to_websockets": None,
|
|
"DEBUG": 0,
|
|
"verbose": False,
|
|
}
|
|
|
|
|
|
def test_handle_datagram_records_rtt_history_for_new_connection():
|
|
hbdclass.Host.hosts.pop("rtt-hist-host", None)
|
|
handle_datagram(_htb("rtt-hist-host", rtt=42.5), ("127.0.0.1", 50000),
|
|
_FakeTransport(), _base_ctx())
|
|
|
|
host = hbdclass.Host.hosts["rtt-hist-host"]
|
|
samples = host.plugin_data.get("rtt_ipv4")
|
|
assert samples is not None
|
|
assert len(samples) == 1
|
|
ts, data = samples[0]
|
|
assert data == {"rtt": 42.5}
|
|
assert isinstance(ts, float)
|
|
|
|
|
|
def test_handle_datagram_appends_rtt_history_across_heartbeats():
|
|
hbdclass.Host.hosts.pop("rtt-hist-host2", None)
|
|
transport = _FakeTransport()
|
|
ctx = _base_ctx()
|
|
handle_datagram(_htb("rtt-hist-host2", rtt=10.0), ("127.0.0.1", 50000), transport, ctx)
|
|
handle_datagram(_htb("rtt-hist-host2", rtt=20.0), ("127.0.0.1", 50000), transport, ctx)
|
|
|
|
host = hbdclass.Host.hosts["rtt-hist-host2"]
|
|
samples = host.plugin_data["rtt_ipv4"]
|
|
assert [d["rtt"] for _, d in samples] == [10.0, 20.0]
|
|
|
|
|
|
def test_handle_datagram_skips_rtt_history_when_rtt_missing():
|
|
hbdclass.Host.hosts.pop("rtt-hist-host3", None)
|
|
handle_datagram(_htb("rtt-hist-host3", rtt=None), ("127.0.0.1", 50000),
|
|
_FakeTransport(), _base_ctx())
|
|
|
|
host = hbdclass.Host.hosts["rtt-hist-host3"]
|
|
assert "rtt_ipv4" not in host.plugin_data
|