Files
qtalker---/backend/tests/test_telemetry.py
Indiana c372427ced feat: matrix/graph data views, wire-generated anomalies, http cookie fix
- auth: session cookie Secure only over https — plain-http LAN access was
  silently dropping the cookie, killing WS auth ('connection unstable')
- wire ghost: server-side spike detection on jitter baseline (3σ + 2.5×mean,
  20KB/s floor, 20s throttle) — wire anomalies now flood every session with
  zero hardware, pushed to clients as {type:'anomaly'} frames
- telemetry cadence 3-5s for live graphs; ambient whispers unchanged
- frontend: MATRIX view (data-rain interleaved with live utterances/anomaly/
  telemetry strings), GRAPHS view (scrolling jitter/variance/dns lines +
  anomaly markers + counters), BOARD/MATRIX/GRAPHS switcher
- connection banner: 'connecting' is now neutral 'tuning the veil…', only
  unstable/closed warns
- db: recreated quantumancy(+_test) as UTF8 (was SQL_ASCII — crashed on
  non-ASCII spirit text); README quickstart updated
- i18n: seance.views.* EN/ES
2026-07-21 00:02:41 +00:00

49 lines
1.9 KiB
Python

from app.telemetry import detect_wire_spike, parse_proc_net_dev
PROC_NET_DEV = """Inter-| Receive | Transmit
face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed
lo: 1234567 1000 0 0 0 0 0 0 1234567 1000 0 0 0 0 0 0
eth0: 9876543 5000 0 0 0 0 0 0 1111111 4000 0 0 0 0 0 0
"""
def test_parse_proc_net_dev_extracts_counters():
counters = parse_proc_net_dev(PROC_NET_DEV)
assert counters == {
"lo": (1234567, 1234567),
"eth0": (9876543, 1111111),
}
def test_parse_proc_net_dev_ignores_malformed_lines():
assert parse_proc_net_dev("garbage\nno colon here\n") == {}
def test_spike_needs_history():
assert detect_wire_spike([], 1_000_000) is None
assert detect_wire_spike([10_000.0] * 3, 1_000_000) is None
def test_spike_fires_on_real_surge():
history = [20_000.0, 25_000.0, 22_000.0, 21_000.0, 23_000.0, 24_000.0, 22_500.0]
ratio = detect_wire_spike(history, 400_000.0)
assert ratio is not None
assert ratio > 10
def test_spike_ignores_normal_fluctuation():
history = [20_000.0, 25_000.0, 22_000.0, 21_000.0, 23_000.0, 24_000.0, 22_500.0]
assert detect_wire_spike(history, 30_000.0) is None
def test_spike_has_absolute_floor_for_silent_links():
# A nearly idle link jittering by a few KB/s must never cry ghost.
history = [100.0, 150.0, 120.0, 90.0, 110.0, 130.0, 140.0]
assert detect_wire_spike(history, 5_000.0) is None
def test_spike_adapts_to_loud_baseline():
# Once the line is genuinely busy, the same surge is no longer anomalous.
history = [350_000.0, 380_000.0, 360_000.0, 370_000.0, 355_000.0, 375_000.0, 365_000.0]
assert detect_wire_spike(history, 400_000.0) is None