MythEclipse/omarchy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
omarchy/plugins/io.github.x3me.nexthop/test/test_linkevents.py
T
asepharyana 1cdb82a76f Sync config from arch 
- hypr/apps.lua
- hypr/autostart.lua
- hypr/envs.lua
- hypr/hyprland.lua
- hypr/hyprsunset.conf
- hypr/input.lua
- hypr/looknfeel.lua
- hypr/omasettings.lua
- hypr/xdph.conf
- omarchy/branding/about.txt
- omarchy/branding/screensaver.txt
- omarchy/extensions/omarchy-menu.jsonc
- omarchy/hooks/battery-low.d/play-warning-sound.sample
- omarchy/hooks/font-set.d/show-font-notification.sample
- omarchy/hooks/post-boot.d/weather.sample
- omarchy/hooks/post-update.d/install-voxtype.hook
- omarchy/hooks/post-update.d/setup-agent.hook
- omarchy/hooks/post-update.d/setup-fingerprint.hook
- omarchy/hooks/post-update.d/show-update-notification.sample
- omarchy/hooks/pre-refresh-pacman.d/add-custom-repo.sample
- omarchy/hooks/theme-set.d/show-theme-notification.sample
- omarchy/shell.json
- omarchy/shell.toml
- omarchy/theme.name
- omarchy/themes/azure-glow/README.md
- omarchy/themes/azure-glow/alacritty.toml
- omarchy/themes/azure-glow/btop.theme
- omarchy/themes/azure-glow/hyprland.conf
- omarchy/themes/azure-glow/hyprlock.conf
- omarchy/themes/azure-glow/icons.theme
- … 269 more
2026-09-23 15:19:12 +07:00

359 lines
15 KiB
Python
Raw Permalink Blame History

"""Tests for linkevents.py and LinkWatch — who ended the association.
Run: python3 -m unittest discover -s test
"""
import sys
import tempfile
import time
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from nexthopd.daemon import LinkWatch # noqa: E402
from nexthopd.linkevents import NlEvents, reason_text # noqa: E402
from nexthopd.store import Store # noqa: E402
from support import StubEvents # noqa: E402
class LinkEvents(unittest.TestCase):
def setUp(self):
self.dir = tempfile.TemporaryDirectory()
self.store = Store(Path(self.dir.name) / "t.db")
self.watch = LinkWatch(self.store)
def tearDown(self):
self.store.close()
self.dir.cleanup()
def kinds(self):
return [e["kind"] for e in self.store.events()]
def test_first_sighting_is_not_an_association(self):
# A daemon restart sees an existing link — that is not an event.
t = time.time()
self.watch.sample(t, {"bssid": "aa:aa:aa:aa:aa:aa", "ssid": "Office"})
self.assertEqual(self.kinds(), [])
def test_roam_and_detail(self):
t = time.time()
self.watch.sample(t, {"bssid": "aa:aa:aa:aa:aa:aa", "ssid": "Office",
"channel": 149, "signal_dbm": -61})
self.watch.sample(t + 2, {"bssid": "bb:bb:bb:bb:bb:bb", "ssid": "Office",
"channel": 44, "signal_dbm": -47})
kinds = self.kinds()
self.assertIn("roam", kinds)
self.assertNotIn("associate", kinds)
roam = [e for e in self.store.events() if e["kind"] == "roam"][0]
self.assertIn("149", roam["detail"])
self.assertIn("44", roam["detail"])
self.assertIn("-61", roam["detail"])
def test_associate_needs_a_confirmed_gap(self):
t = time.time()
link = {"bssid": "aa:aa:aa:aa:aa:aa", "ssid": "Office"}
self.watch.sample(t, link)
# A brief iw hiccup (fewer empty reads than the threshold) is not a
# disassociation, so the recovery is not an association.
for i in range(LinkWatch.GAP_SAMPLES - 1):
self.watch.sample(t + 2 + i * 2, {})
self.watch.sample(t + 20, link)
self.assertEqual(self.kinds(), [])
# A confirmed gap is, and the recovery is logged once.
for i in range(LinkWatch.GAP_SAMPLES):
self.watch.sample(t + 30 + i * 2, {})
self.watch.sample(t + 60, link)
self.assertEqual(self.kinds(), ["associate"])
BUSY = 1_000_000 # bytes/sec, well above the idle floor
def test_rate_drop_needs_sustain(self):
t = time.time()
link = {"bssid": "aa:aa:aa:aa:aa:aa", "ssid": "X", "channel": 44}
for i in range(5):
self.watch.sample(t + i * 2, dict(link, tx_mbps=500), self.BUSY)
# A momentary dip is not an event.
self.watch.sample(t + 12, dict(link, tx_mbps=120), self.BUSY)
self.watch.sample(t + 14, dict(link, tx_mbps=500), self.BUSY)
self.assertNotIn("rate-drop", self.kinds())
# A sustained one is, and it closes with the recovery.
for i in range(8):
self.watch.sample(t + 20 + i * 2, dict(link, tx_mbps=110), self.BUSY)
self.assertIn("rate-drop", self.kinds())
self.watch.sample(t + 40, dict(link, tx_mbps=480), self.BUSY)
drop = [e for e in self.store.events() if e["kind"] == "rate-drop"][0]
self.assertIsNotNone(drop["ended_ts"])
self.assertIn("110", drop["detail"])
def test_idle_rate_drops_are_not_events(self):
# Power save renegotiates a low bitrate the moment the link idles;
# with no traffic that is invisible to the user and must not log.
t = time.time()
link = {"bssid": "aa:aa:aa:aa:aa:aa", "ssid": "X", "channel": 44}
for i in range(5):
self.watch.sample(t + i * 2, dict(link, tx_mbps=2000), self.BUSY)
for i in range(20):
self.watch.sample(t + 20 + i * 2, dict(link, tx_mbps=120), 500)
self.assertNotIn("rate-drop", self.kinds())
def test_throttled_to_one_hz(self):
t = time.time()
self.watch.sample(t, {"bssid": "aa:aa:aa:aa:aa:aa", "ssid": "X"})
# Two samples inside the same second: the second is ignored, so a
# bssid flap faster than 1 Hz cannot spam the log.
self.watch.sample(t + 0.4, {"bssid": "bb:bb:bb:bb:bb:bb", "ssid": "X"})
self.assertNotIn("roam", self.kinds())
class LinkAttribution(unittest.TestCase):
"""A BSSID change says who ended the previous association."""
OLD = "aa:aa:aa:aa:aa:aa"
NEW = "bb:bb:bb:bb:bb:bb"
def setUp(self):
self.dir = tempfile.TemporaryDirectory()
self.store = Store(Path(self.dir.name) / "t.db")
def tearDown(self):
self.store.close()
self.dir.cleanup()
@staticmethod
def cause(by_ap, reason, gap_s):
return {"by_ap": by_ap, "reason": reason, "gap_s": gap_s,
"frame": "deauth", "t": 0.0}
def change(self, cause, raise_=False):
events = StubEvents(cause, raise_)
watch = LinkWatch(self.store, events)
t = time.time()
watch.sample(t, {"bssid": self.OLD, "ssid": "Office",
"channel": 149, "signal_dbm": -30})
watch.sample(t + 2, {"bssid": self.NEW, "ssid": "Office",
"channel": 13, "signal_dbm": -53})
evs = self.store.events()
self.assertEqual(len(evs), 1)
return evs[0], events
def test_ap_deauth_is_a_kick(self):
e, events = self.change(self.cause(True, 2, 3.0))
self.assertEqual(e["kind"], "kick")
self.assertIn("Kicked by AP " + self.OLD, e["detail"])
self.assertIn("reason 2: previous authentication no longer valid",
e["detail"])
self.assertIn("rejoined via " + self.NEW + " after 3 s", e["detail"])
self.assertIn("channel 149 \u2192 13", e["detail"])
self.assertIn("-30 \u2192 -53 dBm", e["detail"])
# It asked about the AP we left, over a window reaching back to
# when that link was last seen up.
bssid, now, window = events.calls[0]
self.assertEqual(bssid, self.OLD)
self.assertGreaterEqual(window, 2.0)
def test_local_deauth_with_instant_reauth_is_a_roam(self):
# mac80211 emits the deauth from inside the call that starts the
# new authentication, so a roam's gap is milliseconds.
e, _ = self.change(self.cause(False, 1, 0.004))
self.assertEqual(e["kind"], "roam")
self.assertTrue(e["detail"].startswith("Roamed to " + self.NEW))
def test_local_deauth_with_a_scan_first_is_a_drop(self):
e, _ = self.change(self.cause(False, 4, 2.8))
self.assertEqual(e["kind"], "drop")
self.assertIn("Dropped by this machine (reason 4: beacon loss)",
e["detail"])
self.assertIn("rejoined via " + self.NEW + " after 3 s", e["detail"])
def test_unknown_cause_stays_a_roam(self):
e, _ = self.change(None)
self.assertEqual(e["kind"], "roam")
self.assertTrue(e["detail"].startswith("Roamed to " + self.NEW))
def test_without_an_event_source_nothing_changes(self):
watch = LinkWatch(self.store)
t = time.time()
watch.sample(t, {"bssid": self.OLD, "ssid": "Office"})
watch.sample(t + 2, {"bssid": self.NEW, "ssid": "Office"})
self.assertEqual([e["kind"] for e in self.store.events()], ["roam"])
def test_attribution_failure_never_costs_the_event(self):
e, _ = self.change(self.cause(True, 2, 3.0), raise_=True)
self.assertEqual(e["kind"], "roam")
def test_kick_with_a_confirmed_gap_is_one_row(self):
# Kicked, off the air long enough to count as disassociated, back
# on the same radio: one row says all of that, not a bare
# "Associated with".
events = StubEvents(self.cause(True, 2, None))
watch = LinkWatch(self.store, events)
t = time.time()
link = {"bssid": self.OLD, "ssid": "Office"}
watch.sample(t, link)
for i in range(LinkWatch.GAP_SAMPLES):
watch.sample(t + 2 + i * 2, {})
watch.sample(t + 14, link)
evs = self.store.events()
self.assertEqual([e["kind"] for e in evs], ["kick"])
self.assertIn("rejoined after 14 s", evs[0]["detail"])
self.assertNotIn(" via ", evs[0]["detail"])
# The lookup window reached back to the last time the link was up.
self.assertGreaterEqual(events.calls[-1][2], 14.0)
def test_gap_nobody_claimed_is_a_plain_association(self):
watch = LinkWatch(self.store, StubEvents(None))
t = time.time()
link = {"bssid": self.OLD, "ssid": "Office"}
watch.sample(t, link)
for i in range(LinkWatch.GAP_SAMPLES):
watch.sample(t + 2 + i * 2, {})
watch.sample(t + 14, link)
self.assertEqual([e["kind"] for e in self.store.events()], ["associate"])
class NlEventParsing(unittest.TestCase):
"""`iw event -t` lines become causes; everything else is ignored."""
AP = "02:11:22:33:44:05"
ME = "02:aa:bb:cc:dd:ee"
NEW = "02:11:22:33:44:09"
def feed(self, lines):
ev = NlEvents()
for line in lines:
ev.consume(line)
return ev
def test_ap_kick_then_rejoin_after_a_scan(self):
ev = self.feed([
"1000.000100: wlo1 (phy #0): deauth %s -> %s reason 2: "
"Previous authentication no longer valid" % (self.AP, self.ME),
"1000.000200: wlo1 (phy #0): disconnected (by AP) reason: 2: "
"Previous authentication no longer valid",
"1000.100000: wlo1 (phy #0): scan started",
'1002.900000: wlo1 (phy #0): scan finished: 2412 2437, ""',
"1003.000100: wlo1 (phy #0): auth %s -> %s status: 0: Successful"
% (self.NEW, self.ME),
"1003.010000: wlo1 (phy #0): assoc %s -> %s status: 0: Successful"
% (self.NEW, self.ME),
"1003.012000: wlo1 (phy #0): connected to %s" % self.NEW,
])
c = ev.cause_for(self.AP, 1004.0, 30.0)
self.assertTrue(c["by_ap"])
self.assertEqual(c["reason"], 2)
self.assertEqual(c["frame"], "deauth")
self.assertAlmostEqual(c["gap_s"], 3.0, places=2)
# The new AP ended nothing; and a cause ages out of the window.
self.assertIsNone(ev.cause_for(self.NEW, 1004.0, 30.0))
self.assertIsNone(ev.cause_for(self.AP, 1100.0, 30.0))
def test_client_roam_is_local_and_instant(self):
ev = self.feed([
"2000.000000: wlo1 (phy #0): deauth %s -> %s reason 1: Unspecified"
% (self.ME, self.AP),
"2000.004000: wlo1 (phy #0): auth %s -> %s status: 0: Successful"
% (self.NEW, self.ME),
])
c = ev.cause_for(self.AP, 2001.0, 30.0)
self.assertFalse(c["by_ap"])
self.assertLess(c["gap_s"], LinkWatch.ROAM_FOLLOW_S)
def test_ap_disassoc_counts_and_case_does_not_matter(self):
ev = self.feed([
"3000.5: wlo1 (phy #0): disassoc %s -> %s reason 4: Disassociated "
"due to inactivity" % (self.AP.upper(), self.ME.upper()),
])
c = ev.cause_for(self.AP, 3001.0, 10.0)
self.assertEqual((c["by_ap"], c["reason"], c["frame"], c["gap_s"]),
(True, 4, "disassoc", None))
def test_forged_and_junk_lines_are_ignored(self):
ev = self.feed([
"4000.0: wlo1 (phy #0): unprotected deauth %s -> %s reason 7: x"
% (self.AP, self.ME),
"4000.1: wlo1 (phy #0): deauth %s -> %s reason 99999999: x"
% (self.AP, self.ME),
"4000.2: wlo1 (phy #0): deauth not-a-mac -> %s reason 2: x" % self.ME,
"garbage", "", "x" * 5000,
])
self.assertIsNone(ev.cause_for(self.AP, 4001.0, 1e9))
def test_untimestamped_lines_take_the_clock_given(self):
ev = NlEvents()
ev.consume("wlo1: deauth %s -> %s reason 3: Leaving" % (self.ME, self.AP),
now=5000.0)
ev.consume("wlo1: connected to %s" % self.NEW, now=5002.5)
c = ev.cause_for(self.AP, 5003.0, 10.0)
self.assertEqual((c["by_ap"], c["reason"]), (False, 3))
self.assertAlmostEqual(c["gap_s"], 2.5, places=3)
def test_follow_up_binds_to_the_newest_cause_only(self):
ev = self.feed([
"6000.0: wlo1 (phy #0): deauth %s -> %s reason 2: x" % (self.AP, self.ME),
"6001.0: wlo1 (phy #0): auth %s -> %s status: 0: Successful"
% (self.NEW, self.ME),
"6002.0: wlo1 (phy #0): assoc %s -> %s status: 0: Successful"
% (self.NEW, self.ME),
])
self.assertAlmostEqual(ev.cause_for(self.AP, 6003.0, 10.0)["gap_s"], 1.0)
def test_reason_words_depend_on_who_sent_it(self):
self.assertEqual(reason_text(4, True), "reason 4: inactivity")
self.assertEqual(reason_text(4, False), "reason 4: beacon loss")
self.assertEqual(reason_text(3, True), "reason 3: the AP is leaving")
self.assertEqual(reason_text(8, True), "reason 8: the AP is leaving the BSS")
self.assertEqual(reason_text(8, False), "reason 8: leaving the BSS")
self.assertIn("previous authentication", reason_text(2, True))
self.assertEqual(reason_text(250, True), "reason 250")
class NlEventRecording(unittest.TestCase):
"""A real kick, recorded with `iw event -t` on an Intel BE201 while the
gateway ran `iwpriv ra1 set DisConnectSta=<this laptop>`: the AP's
deauth, a scan, and the re-association on another radio 2.9 s later.
Addresses are substituted; timings and formats are as recorded."""
OLD = "02:11:22:33:44:01"
NEW = "02:11:22:33:44:05"
def setUp(self):
path = Path(__file__).parent / "fixtures" / "iw-event.txt"
self.lines = path.read_text().splitlines()
self.events = NlEvents()
for line in self.lines:
self.events.consume(line)
def test_recording_parses_as_an_ap_kick(self):
c = self.events.cause_for(self.OLD, 1788278083.0, 30.0)
self.assertEqual((c["by_ap"], c["reason"], c["frame"]),
(True, 8, "deauth"))
self.assertAlmostEqual(c["gap_s"], 2.93, places=1)
# The radio we landed on ended nothing.
self.assertIsNone(self.events.cause_for(self.NEW, 1788278083.0, 30.0))
def test_link_watch_logs_it_as_one_kick(self):
d = tempfile.TemporaryDirectory()
self.addCleanup(d.cleanup)
store = Store(Path(d.name) / "t.db")
self.addCleanup(store.close)
watch = LinkWatch(store, self.events)
watch.sample(1788278078.0, {"bssid": self.OLD, "ssid": "x",
"channel": 1, "signal_dbm": -30})
watch.sample(1788278083.0, {"bssid": self.NEW, "ssid": "x",
"channel": 149, "signal_dbm": -40})
# Read the log on the recording's own clock. store.events() windows
# on wall-clock now, so anchoring it anywhere else makes this test
# pass until the fixture is a week old and fail every day after.
evs = store.events(now=1788278083.0)
self.assertEqual([e["kind"] for e in evs], ["kick"])
self.assertEqual(
evs[0]["detail"],
"Kicked by AP 02:11:22:33:44:01 (reason 8: the AP is leaving the "
"BSS), rejoined via 02:11:22:33:44:05 after 3 s, "
"channel 1 \u2192 149, -30 \u2192 -40 dBm")
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink