feat: record RTT history per heartbeat for charting

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
2026-08-17 07:18:01 -04:00
parent a7b698e6b9
commit 52968201fa
2 changed files with 69 additions and 0 deletions
@@ -0,0 +1,64 @@
+"""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