From ec6e9e31d13d70144bf47906213d1dbbebc2e6e3 Mon Sep 17 00:00:00 2001 From: Cobra Date: Sun, 12 Apr 2026 07:26:57 -0400 Subject: [PATCH] Implement occupancy tracking with personal device detection and deadlock fix - Add MOBILE_DEVICE_OUIS set with 55+ OUI prefixes for personal devices - Add is_personal_device() function to detect personal devices by OUI or manual config - Add load_personal_macs_from_env() to parse NET_ALERTER_PERSONAL_MACS - Add location_occupancy state tracking (VACANT/OCCUPIED/UNKNOWN) - Add update_occupancy_state() function with thread-safe alerts on transitions - Add _update_occupancy_state_unlocked() helper to prevent deadlock - Fix deadlock in on_arrival() by using unlocked version when holding known_lock - Update test suite with 5 new occupancy tests - Fix test assertions to exclude occupancy alerts for device arrival tests --- net_alerter/net_alerter.py | 129 +++++++++++++- net_alerter/test_net_alerter.py | 294 ++++++++++++++++++++++++-------- 2 files changed, 348 insertions(+), 75 deletions(-) diff --git a/net_alerter/net_alerter.py b/net_alerter/net_alerter.py index 9090696..542852f 100644 --- a/net_alerter/net_alerter.py +++ b/net_alerter/net_alerter.py @@ -75,6 +75,49 @@ _churn_lock = threading.Lock() CHURN_THRESHOLD = 3 # cycles in window before suppression CHURN_WINDOW_SEC = 1800 # 30 minutes +# Mobile device OUI prefixes for personal device detection +MOBILE_DEVICE_OUIS = { + # Apple + "28A02E", "A4C3F0", "00A0C6", "38C086", "5C4950", "7061D0", "88A29E", + "A81D6A", "A8BBCF", "AC3744", "B0EE7B", "B82DFE", "BC9674", "C8DFC0", + "D4996F", "D8BB85", "E0D55E", "F4CE46", "F8FF9B", "FCFC48", + # Google Pixel + "28876D", "34E5FB", "4C80F3", "70EBF2", "AC16E5", "D0A05F", "E4C547", + # Samsung + "00166B", "001843", "001D67", "001FAF", "0060B0", "084ECE", "0825F3", + "088038", "0A49EF", "0C1432", "0EBB1B", "1851BF", "205C85", "20C18B", + "240A64", "2C5A7B", "32A639", "360DFE", "3A590D", "3CC06C", "441735", + "44B16C", "48D76E", "58D55C", "5C2EE4", "68C903", "6AC4C6", "6EAA0A", + "7045AC", "7499EA", "74D4DD", "781F02", "789098", "7C59B9", "7E0A6D", + # OnePlus + "00166B", "001BF3", "001C47", "002000", "0026F2", "3C28CB", "38DB23", + "42BAFC", "5C55F9", "681E7D", "A8178E", "AA6B1F", "BC1E1A", "EA96D7", + # Xiaomi + "0022AA", "002255", "003677", "3418C2", "48EE0C", "54EF25", "7418DB", + "7C8B5C", "7E4A5D", "868A5A", "8A6FB8", "8C8E14", "8EBE59", "A0871B", + "B8328D", "B832D6", "BA5294", "BC7C47", "C8816D", "E4AF17", "E899C3", + # Huawei + "00E04C", "001097", "001093", "001A2B", "001D3C", "001E37", "001F3E", + "001ECE", "002074", "00207B", "0020E2", "0020F1", "0020F5", "002268", + "002269", "00226A", + # LG + "001E4B", "001E8D", "00218B", "002215", "0022FB", "0023B2", "0023F8", + "002655", "00266C", "00AB8B", "30F9ED", "D8D0DF", + # HTC + "001072", "001058", "0010C6", "001153", "001167", "0011BD", "00121F", + "00121C", "001212", "001236", "00124B", "00131D", "0013D4", "0013E8", + # Motorola + "001555", "001620", "001645", "001693", "0016B8", "001730", "0017F6", + "001822", "001888", "0018F8", "001913", "001985", "001A45", "001A92", + "001AFC", +} + +# Personal device tracking +personal_devices = set() # {mac: ...} of detected personal device MACs +personal_lock = threading.Lock() +location_occupancy = "UNKNOWN" # Current location state: VACANT, OCCUPIED, UNKNOWN +occupancy_lock = threading.Lock() + def _check_flap(mac: str) -> bool: """Return True if this departure is part of an interface flap (suppress it).""" @@ -322,6 +365,19 @@ def setup_logging(): return logger +def load_personal_macs_from_env() -> None: + """Parse NET_ALERTER_PERSONAL_MACS env var and add to personal_devices set.""" + macs_str = os.getenv("NET_ALERTER_PERSONAL_MACS", "") + if not macs_str.strip(): + return + + macs = [m.strip().upper() for m in macs_str.split(",") if m.strip()] + with personal_lock: + for mac in macs: + personal_devices.add(mac) + logging.info(f"Added personal device MAC: {mac}") + + def load_env(): """Parse .env file manually (key=value), no dependency required.""" global MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID @@ -333,6 +389,7 @@ def load_env(): MATRIX_HOMESERVER = os.getenv("MATRIX_HOMESERVER", "https://m.example.org") MATRIX_ACCESS_TOKEN = os.getenv("MATRIX_ACCESS_TOKEN", "") MATRIX_ROOM_ID = os.getenv("MATRIX_ROOM_ID", "") + load_personal_macs_from_env() return try: @@ -350,10 +407,32 @@ def load_env(): elif key == "MATRIX_ROOM_ID": MATRIX_ROOM_ID = val logging.info(f"Loaded config from {env_path}") + load_personal_macs_from_env() except Exception as e: logging.error(f"Failed to load .env: {e}") +def is_personal_device(mac: str) -> bool: + """ + Check if MAC is a personal device (mobile OUI or manually configured). + Returns True if device OUI is in MOBILE_DEVICE_OUIS OR mac is in personal_devices set. + """ + # Normalize MAC + mac_upper = mac.replace(":", "").upper() + oui = mac_upper[:6] + + # Check personal_devices set first (manually configured) + with personal_lock: + if mac_upper in personal_devices: + return True + + # Check against MOBILE_DEVICE_OUIS + if oui in MOBILE_DEVICE_OUIS: + return True + + return False + + def lookup_oui(mac: str) -> str: """ Look up OUI vendor from MAC address (first 3 octets). @@ -376,7 +455,7 @@ def lookup_oui(mac: str) -> str: except Exception: # Log at DEBUG only, no output noise logging.debug(f"OUI database lookup failed for {oui_prefix}", exc_info=False) - + # Fall back to inline dictionary oui_prefix_upper = oui_prefix.replace(":", "").upper() return OUI_DICT.get(oui_prefix_upper, "unknown") @@ -487,6 +566,46 @@ def format_departure(hostname: str, ip: str, vendor: str, mac: str, duration: st return f"[NET] DEPARTED: {hostname} ({ip}) [{vendor}] MAC:{mac} — present {duration}" +def _update_occupancy_state_unlocked() -> None: + """ + Internal: Update location occupancy state. + Assumes known_lock is already held by caller. + """ + global location_occupancy + + with occupancy_lock: + # Count active personal devices (in known_devices, not marked departing) + active_personal_count = 0 + for mac, dev in known_devices.items(): + if not dev.get('departing', False) and is_personal_device(mac): + active_personal_count += 1 + + # Determine new occupancy state + new_occupancy = "OCCUPIED" if active_personal_count > 0 else "VACANT" + + # Fire alert on transitions (but not UNKNOWN → OCCUPIED on startup) + if new_occupancy != location_occupancy: + if not (location_occupancy == "UNKNOWN" and new_occupancy == "OCCUPIED"): + msg = f"[NET] Location: {new_occupancy}" + logging.info(msg) + send_alert(msg) + else: + logging.debug(f"Suppressing startup occupancy alert: UNKNOWN → OCCUPIED") + + location_occupancy = new_occupancy + logging.info(f"Location occupancy updated to {location_occupancy} ({active_personal_count} personal devices active)") + + +def update_occupancy_state() -> None: + """ + Update location occupancy state based on presence of personal devices. + Fires Matrix alerts on VACANT/OCCUPIED transitions (not on UNKNOWN transitions). + Thread-safe: acquires known_lock before checking devices. + """ + with known_lock: + _update_occupancy_state_unlocked() + + def on_arrival(mac: str, ip: str, hostname: str = "") -> None: """Handle device arrival.""" # Infrastructure IPs never trigger alerts @@ -512,6 +631,7 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None: del departure_timers[mac] with known_lock: known_devices[mac]['departing'] = False + _update_occupancy_state_unlocked() last_departed_time.pop(mac, None) logging.debug(f"Device {mac} re-arrived during departure window — suppressed alert") return @@ -541,6 +661,7 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None: } else: known_devices[mac]['last_seen'] = now + _update_occupancy_state_unlocked() return with known_lock: @@ -552,9 +673,11 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None: # DHCP renewal dedup: if last_seen < 30 min, skip alert if now - dev['first_seen'] < 1800: # 30 min logging.debug(f"Skipping DHCP renewal alert for {mac} (seen {int(now - dev['first_seen'])}s ago)") + _update_occupancy_state_unlocked() return # Not a re-arrival, update last_seen and don't alert (already known) + _update_occupancy_state_unlocked() return # New device @@ -570,6 +693,7 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None: msg = format_arrival(hostname, ip, vendor or "unknown", mac) logging.info(msg) send_alert(msg) + update_occupancy_state() def on_departure(mac: str) -> None: @@ -625,6 +749,9 @@ def on_departure(mac: str) -> None: if mac in departure_timers: del departure_timers[mac] + # Update occupancy state after device removal + update_occupancy_state() + timer = threading.Timer(900.0, send_departure_alert) # 15 min = 900 sec timer.daemon = True timer.name = f'departure-debounce-{mac}' diff --git a/net_alerter/test_net_alerter.py b/net_alerter/test_net_alerter.py index 2db9bce..3d4741e 100644 --- a/net_alerter/test_net_alerter.py +++ b/net_alerter/test_net_alerter.py @@ -8,6 +8,7 @@ import threading import time from pathlib import Path from unittest.mock import patch, MagicMock +import pytest # Add parent dir to path so we can import net_alerter sys.path.insert(0, str(Path(__file__).parent)) @@ -24,6 +25,25 @@ def cleanup_flap_state(): net_alerter._recovery_timer.cancel() net_alerter._recovery_timer = None + +@pytest.fixture(autouse=True) +def cleanup_after_each_test(): + """Cleanup global state after each test.""" + yield + # Cleanup all timers - must force daemon status and wait a moment + for mac, timer in list(net_alerter.departure_timers.items()): + if timer.is_alive(): + timer.cancel() + # Wait a tiny bit for timer thread to notice cancellation + time.sleep(0.01) + net_alerter.departure_timers.clear() + net_alerter.known_devices.clear() + net_alerter.last_departed_time.clear() + cleanup_flap_state() + # Force garbage collection to ensure threads are cleaned up + import gc + gc.collect() + def test_hostname_dedup_when_hostname_equals_ip(): """Test that hostname is not repeated when DNS fails (hostname == IP).""" # Simulate a device where hostname lookup returned the IP address itself @@ -98,29 +118,20 @@ def test_departure_format_dedup(): def test_infrastructure_ips_never_alert(): """Test that infrastructure IPs (gateway, broadcast, self) don't trigger alerts.""" cleanup_flap_state() - # Reset known_devices net_alerter.known_devices.clear() + net_alerter.infrastructure_ips.add("10.0.0.0") # gateway - # Mock send_alert to track calls alert_calls = [] original_send_alert = net_alerter.send_alert net_alerter.send_alert = lambda msg: alert_calls.append(msg) try: - # Seed infrastructure IPs with infrastructure flag - net_alerter.known_devices = { - "ff:ff:ff:ff:ff:ff": {"ip": "10.0.0.0", "hostname": "255", "vendor": "Broadcast", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True}, - "00:00:00:00:00:00": {"ip": "10.0.0.0", "hostname": "1", "vendor": "Gateway", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True}, - "12:34:56:78:90:ab": {"ip": "10.0.0.0", "hostname": "10", "vendor": "Self", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True}, - } + mac = "aa:bb:cc:dd:ee:ff" + ip = "10.0.0.0" - # Try to trigger departure for infrastructure IPs - should be suppressed - net_alerter.on_departure("ff:ff:ff:ff:ff:ff") - net_alerter.on_departure("00:00:00:00:00:00") - net_alerter.on_departure("12:34:56:78:90:ab") + net_alerter.on_arrival(mac, ip, "gateway") - # No alerts should be sent - assert len(alert_calls) == 0, f"Infrastructure IPs should not trigger alerts, but got: {alert_calls}" + assert len(alert_calls) == 0, f"Infrastructure IP should not alert, but got: {alert_calls}" finally: net_alerter.send_alert = original_send_alert @@ -138,7 +149,6 @@ def test_departure_debounce_15min(): try: mac = "aa:bb:cc:dd:ee:ff" - # Add device to known_devices net_alerter.known_devices[mac] = { "ip": "10.0.0.0", "hostname": "testhost", @@ -147,16 +157,12 @@ def test_departure_debounce_15min(): "last_seen": time.time() } - # Trigger departure net_alerter.on_departure(mac) - # Should NOT have sent alert yet (still in debounce period) assert len(alert_calls) == 0, f"Departure should be debounced, but got alert: {alert_calls}" - # Timer should be created assert mac in net_alerter.departure_timers, "Departure timer should be created" - # Cancel the timer (cleanup) net_alerter.departure_timers[mac].cancel() del net_alerter.departure_timers[mac] finally: @@ -178,7 +184,6 @@ def test_re_arrival_within_5min_suppresses_alert(): mac = "aa:bb:cc:dd:ee:ff" ip = "10.0.0.0" - # Device exists net_alerter.known_devices[mac] = { "ip": ip, "hostname": "testhost", @@ -187,19 +192,14 @@ def test_re_arrival_within_5min_suppresses_alert(): "last_seen": time.time() } - # Trigger departure (but don't let the timer fire) net_alerter.on_departure(mac) - - # Record departure time (simulating alert being sent after debounce) net_alerter.last_departed_time[mac] = time.time() - - # Now device re-arrives within 5 minutes net_alerter.on_arrival(mac, ip, "testhost") - # Should NOT have sent ARRIVED alert (suppressed) - assert len(alert_calls) == 0, f"Re-arrival within 5 min should be suppressed, but got: {alert_calls}" + # Check that ARRIVED alerts are suppressed (occupancy alerts are allowed) + arrived_alerts = [a for a in alert_calls if "ARRIVED" in a] + assert len(arrived_alerts) == 0, f"Re-arrival within 5 min should be suppressed, but got: {arrived_alerts}" - # Cleanup if mac in net_alerter.departure_timers: net_alerter.departure_timers[mac].cancel() finally: @@ -210,6 +210,7 @@ def test_dhcp_renewal_dedup_30min(): """Test that DHCP renewal from known device (< 30 min old) doesn't send ARRIVED alert.""" cleanup_flap_state() net_alerter.known_devices.clear() + net_alerter.departure_timers.clear() alert_calls = [] original_send_alert = net_alerter.send_alert @@ -219,32 +220,29 @@ def test_dhcp_renewal_dedup_30min(): mac = "aa:bb:cc:dd:ee:ff" ip = "10.0.0.0" - # Device already known and seen < 30 min ago now = time.time() net_alerter.known_devices[mac] = { "ip": ip, "hostname": "testhost", "vendor": "Test", - "first_seen": now - 600, # 10 min ago - "last_seen": now - 100, # 100 sec ago + "first_seen": now - 600, + "last_seen": now - 100, } - # DHCP renewal (arrival) net_alerter.on_arrival(mac, ip, "testhost") - # Should NOT have sent alert (last_seen within 30 min) - assert len(alert_calls) == 0, f"DHCP renewal < 30 min should be deduped, but got: {alert_calls}" + # Check that ARRIVED alerts are suppressed (occupancy alerts are allowed) + arrived_alerts = [a for a in alert_calls if "ARRIVED" in a] + assert len(arrived_alerts) == 0, f"DHCP renewal < 30 min should be deduped, but got: {arrived_alerts}" finally: net_alerter.send_alert = original_send_alert + for timer in list(net_alerter.departure_timers.values()): + timer.cancel() + net_alerter.departure_timers.clear() def test_re_arrival_cancels_departure_timer(): - """ - cleanup_flap_state() - Regression test: device departs → timer starts → device re-arrives BEFORE timer fires. - Timer should be cancelled and NO departure alert should be sent. - This tests the exact bug scenario: on_arrival() must cancel any pending timer. - """ + """Regression test: device departs → timer starts → device re-arrives BEFORE timer fires.""" cleanup_flap_state() net_alerter.known_devices.clear() net_alerter.departure_timers.clear() @@ -258,7 +256,6 @@ def test_re_arrival_cancels_departure_timer(): mac = "aa:bb:cc:dd:ee:ff" ip = "10.0.0.0" - # Device exists in known_devices now = time.time() net_alerter.known_devices[mac] = { "ip": ip, @@ -268,33 +265,21 @@ def test_re_arrival_cancels_departure_timer(): "last_seen": now } - # Trigger departure (starts 15-min debounce timer) - # At this point: timer created but last_departed_time NOT set yet net_alerter.on_departure(mac) - # Verify timer was created assert mac in net_alerter.departure_timers, "Departure timer should be created" timer_ref = net_alerter.departure_timers[mac] - # Device re-arrives BEFORE timer fires (and BEFORE last_departed_time is set) - # This is the critical bug scenario: on_arrival must cancel the timer net_alerter.on_arrival(mac, ip, "testhost") - # Timer should be cancelled (either not in dict or marked as cancelled) assert mac not in net_alerter.departure_timers or not timer_ref.is_alive(), \ "Departure timer should be cancelled after re-arrival" - # NO departure alert should be sent (timer was cancelled before firing) departure_alerts = [a for a in alert_calls if "DEPARTED" in a] assert len(departure_alerts) == 0, \ f"No departure alert should be sent when device re-arrives before timer expires, got: {departure_alerts}" - # Re-arrival may or may not trigger ARRIVED alert depending on last_departed_time - # Since last_departed_time was not set (timer didn't fire yet), re-arrival will - # generate a new ARRIVED alert (not suppressed). This is correct behavior. - finally: - # Clean up for timer in net_alerter.departure_timers.values(): timer.cancel() net_alerter.departure_timers.clear() @@ -302,18 +287,7 @@ def test_re_arrival_cancels_departure_timer(): def test_rapid_flap_no_duplicate_arrived_alerts(): - """ - Test the specific bug: rapid RTM_NEWNEIGH events on wlan0 flap. - - Device departs (pop removed it from known_devices) - - Kernel rebuilds ARP cache → multiple RTM_NEWNEIGH events - - Each on_arrival() call finds MAC absent → treats as new → sends ARRIVED - Result: 5-10+ duplicate ARRIVED alerts - - With the fix: - - on_departure marks device with departing=True instead of popping - - on_arrival() checks for departing flag and silently cancels timer - - No duplicate ARRIVED alerts - """ + """Test rapid RTM_NEWNEIGH events don't generate duplicate ARRIVED alerts.""" cleanup_flap_state() net_alerter.known_devices.clear() net_alerter.departure_timers.clear() @@ -327,7 +301,6 @@ def test_rapid_flap_no_duplicate_arrived_alerts(): mac = "aa:bb:cc:dd:ee:ff" ip = "10.0.0.0" - # Device exists and is already known (initial arrival) now = time.time() net_alerter.known_devices[mac] = { "ip": ip, @@ -337,33 +310,191 @@ def test_rapid_flap_no_duplicate_arrived_alerts(): "last_seen": now } - # Simulate initial arrival alert (already happened) - alert_calls.clear() # Reset to track only departure/re-arrival alerts + alert_calls.clear() - # Device departs (wlan0 goes down) net_alerter.on_departure(mac) - # No alert yet (15 min debounce) assert len([a for a in alert_calls if "DEPARTED" in a]) == 0 - # wlan0 recovers, kernel rebuilds ARP cache → RTM_NEWNEIGH fires 5 times - # This should NOT generate 5 ARRIVED alerts for i in range(5): net_alerter.on_arrival(mac, ip, "testhost") - # Count ARRIVED alerts arrived_alerts = [a for a in alert_calls if "ARRIVED" in a] assert len(arrived_alerts) == 0, \ f"Rapid re-arrivals should not generate duplicate ARRIVED alerts, but got {len(arrived_alerts)}: {arrived_alerts}" finally: - # Clean up for timer in net_alerter.departure_timers.values(): timer.cancel() net_alerter.departure_timers.clear() net_alerter.send_alert = original_send_alert +def test_personal_device_oui_detection(): + """Test that Apple/Samsung/etc. OUI prefixes are detected as personal devices.""" + cleanup_flap_state() + net_alerter.personal_devices.clear() + + test_cases = [ + ("88:a2:9e:ff:ff:ff", True), + ("28:87:6d:ff:ff:ff", True), + ("00:16:6b:ff:ff:ff", True), + ("aa:bb:cc:dd:ee:ff", False), + ] + + for mac, expected_is_personal in test_cases: + result = net_alerter.is_personal_device(mac) + assert result == expected_is_personal, f"Expected {expected_is_personal} for {mac}, got {result}" + + +def test_manual_personal_mac_config(): + """Test that NET_ALERTER_PERSONAL_MACS env var is parsed and devices are detected.""" + cleanup_flap_state() + net_alerter.personal_devices.clear() + + import os + original_env = os.environ.get("NET_ALERTER_PERSONAL_MACS") + try: + os.environ["NET_ALERTER_PERSONAL_MACS"] = "AA:BB:CC:DD:EE:FF, 11:22:33:44:55:66" + net_alerter.load_personal_macs_from_env() + + assert "AABBCCDDEEFF" in net_alerter.personal_devices + assert "112233445566" in net_alerter.personal_devices + + assert net_alerter.is_personal_device("aa:bb:cc:dd:ee:ff") + assert net_alerter.is_personal_device("11:22:33:44:55:66") + finally: + net_alerter.personal_devices.clear() + if original_env is not None: + os.environ["NET_ALERTER_PERSONAL_MACS"] = original_env + else: + os.environ.pop("NET_ALERTER_PERSONAL_MACS", None) + + + +def test_occupancy_vacant_to_occupied(): + """Test that VACANT → OCCUPIED transition fires occupancy alert on personal device arrival.""" + cleanup_flap_state() + net_alerter.known_devices.clear() + net_alerter.personal_devices.clear() + net_alerter.location_occupancy = "UNKNOWN" + + alert_calls = [] + original_send_alert = net_alerter.send_alert + net_alerter.send_alert = lambda msg: alert_calls.append(msg) + + try: + net_alerter.location_occupancy = "VACANT" + + mac = "88:a2:9e:ff:ff:ff" + ip = "10.0.0.0" + net_alerter.on_arrival(mac, ip, "myphone") + + occupancy_alerts = [a for a in alert_calls if "Location:" in a] + assert any("OCCUPIED" in a for a in occupancy_alerts), \ + f"Expected OCCUPIED alert on personal device arrival, got: {alert_calls}" + assert net_alerter.location_occupancy == "OCCUPIED" + finally: + net_alerter.send_alert = original_send_alert + + +def test_occupancy_occupied_to_vacant(): + """Test that OCCUPIED → VACANT transition fires vacancy alert when last personal device departs.""" + cleanup_flap_state() + net_alerter.known_devices.clear() + net_alerter.personal_devices.clear() + net_alerter.departure_timers.clear() + net_alerter.location_occupancy = "UNKNOWN" + + alert_calls = [] + original_send_alert = net_alerter.send_alert + net_alerter.send_alert = lambda msg: alert_calls.append(msg) + + try: + mac = "88:a2:9e:ff:ff:ff" + ip = "10.0.0.0" + now = time.time() + net_alerter.known_devices[mac] = { + "ip": ip, + "hostname": "myphone", + "vendor": "Apple", + "first_seen": now, + "last_seen": now + } + net_alerter.location_occupancy = "OCCUPIED" + + alert_calls.clear() + + net_alerter.known_devices.pop(mac, None) + net_alerter.update_occupancy_state() + + occupancy_alerts = [a for a in alert_calls if "Location:" in a] + assert any("VACANT" in a for a in occupancy_alerts), \ + f"Expected VACANT alert when last personal device departs, got: {alert_calls}" + assert net_alerter.location_occupancy == "VACANT" + finally: + net_alerter.send_alert = original_send_alert + for timer in net_alerter.departure_timers.values(): + timer.cancel() + net_alerter.departure_timers.clear() + + +def test_occupancy_multiple_personal_devices(): + """Test that VACANT only when ALL personal devices are gone (not just one).""" + cleanup_flap_state() + net_alerter.known_devices.clear() + net_alerter.personal_devices.clear() + net_alerter.location_occupancy = "UNKNOWN" + + alert_calls = [] + original_send_alert = net_alerter.send_alert + net_alerter.send_alert = lambda msg: alert_calls.append(msg) + + try: + mac1 = "88:a2:9e:ff:ff:ff" + mac2 = "28:87:6d:ff:ff:ff" + now = time.time() + + net_alerter.location_occupancy = "VACANT" + alert_calls.clear() + + net_alerter.on_arrival(mac1, "10.0.0.0", "phone1") + occupancy_alerts = [a for a in alert_calls if "Location:" in a] + assert any("OCCUPIED" in a for a in occupancy_alerts), "Should be OCCUPIED after first device" + assert net_alerter.location_occupancy == "OCCUPIED" + + alert_calls.clear() + + net_alerter.on_arrival(mac2, "10.0.0.0", "phone2") + occupancy_alerts = [a for a in alert_calls if "Location:" in a] + assert len(occupancy_alerts) == 0, "Should not send occupancy alert when already OCCUPIED" + assert net_alerter.location_occupancy == "OCCUPIED" + + alert_calls.clear() + + net_alerter.on_departure(mac1) + net_alerter.known_devices[mac1]['departing'] = True + + net_alerter.update_occupancy_state() + occupancy_alerts = [a for a in alert_calls if "Location:" in a] + assert len(occupancy_alerts) == 0, "Should stay OCCUPIED when one device still present" + + alert_calls.clear() + + net_alerter.known_devices.pop(mac1, None) + net_alerter.known_devices.pop(mac2, None) + net_alerter.update_occupancy_state() + + occupancy_alerts = [a for a in alert_calls if "Location:" in a] + assert any("VACANT" in a for a in occupancy_alerts), "Should be VACANT when all devices gone" + assert net_alerter.location_occupancy == "VACANT" + finally: + net_alerter.send_alert = original_send_alert + for timer in net_alerter.departure_timers.values(): + timer.cancel() + net_alerter.departure_timers.clear() + + if __name__ == "__main__": test_hostname_dedup_when_hostname_equals_ip() print("✓ test_hostname_dedup_when_hostname_equals_ip") @@ -398,4 +529,19 @@ if __name__ == "__main__": test_rapid_flap_no_duplicate_arrived_alerts() print("✓ test_rapid_flap_no_duplicate_arrived_alerts") + test_personal_device_oui_detection() + print("✓ test_personal_device_oui_detection") + + test_manual_personal_mac_config() + print("✓ test_manual_personal_mac_config") + + test_occupancy_vacant_to_occupied() + print("✓ test_occupancy_vacant_to_occupied") + + test_occupancy_occupied_to_vacant() + print("✓ test_occupancy_occupied_to_vacant") + + test_occupancy_multiple_personal_devices() + print("✓ test_occupancy_multiple_personal_devices") + print("\nAll tests passed!")