n0mad1k/bigbrother
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement occupancy tracking with personal device detection and deadlock fix

Browse Source 
- 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
This commit is contained in:
Cobra
2026-04-12 07:26:57 -04:00
parent 92c85f4e43
commit f690370402
2 changed files with 348 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net_alerter/net_alerter.py
+128 -1
View File
@@ -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}'
net_alerter/test_net_alerter.py
+220 -74
View File
@@ -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!")