Cobra
fa0e2491cf
- Reject dash/colon MAC-format device names (e.g. 45-48-39-7F-73-C7)
from open mode — were slipping past is_generic_name()
- Add exponential backoff (1s, 2s) on Matrix 429/5xx, max 3 attempts —
startup burst was rate-limiting and dropping alerts
- Commit active scanner flag (scanning_mode=active) that was in working
tree but not in prior commit c0b91e9 despite the commit message
- Add 3 tests: MAC format filter, 429 retry success, 429 give-up
Fixes DevTrack #518
260 lines
10 KiB
Python
Executable File
260 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Unit tests for ble_alerter.py.
|
|
Mock bleak entirely — no real BLE hardware required.
|
|
"""
|
|
|
|
import asyncio
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch, AsyncMock, call
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Mock bleak before import
|
|
sys.modules['bleak'] = MagicMock()
|
|
|
|
# Now import the daemon (this will use the mocked bleak)
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location(
|
|
"ble_alerter",
|
|
Path(__file__).parent / "ble_alerter.py"
|
|
)
|
|
ble_alerter = importlib.util.module_from_spec(spec)
|
|
sys.modules['ble_alerter'] = ble_alerter
|
|
spec.loader.exec_module(ble_alerter)
|
|
|
|
|
|
class TestIsGenericName(unittest.TestCase):
|
|
"""Test generic name filtering."""
|
|
|
|
def test_empty_name(self):
|
|
"""Empty/None names are generic."""
|
|
self.assertTrue(ble_alerter.is_generic_name(""))
|
|
self.assertTrue(ble_alerter.is_generic_name(None))
|
|
|
|
def test_too_short(self):
|
|
"""Names < 3 chars are generic."""
|
|
self.assertTrue(ble_alerter.is_generic_name("a"))
|
|
self.assertTrue(ble_alerter.is_generic_name("ab"))
|
|
# Names with 3+ chars but non-hex are OK
|
|
self.assertFalse(ble_alerter.is_generic_name("iOS"))
|
|
self.assertFalse(ble_alerter.is_generic_name("Dog"))
|
|
|
|
def test_all_hex(self):
|
|
"""Pure hex is generic (looks like MAC fragment)."""
|
|
self.assertTrue(ble_alerter.is_generic_name("ABCDEF"))
|
|
self.assertTrue(ble_alerter.is_generic_name("123456"))
|
|
self.assertTrue(ble_alerter.is_generic_name("abc")) # a, b, c are hex digits
|
|
# iPhone contains 'h' which is hex, but 'o' and 'n' are not
|
|
self.assertFalse(ble_alerter.is_generic_name("iPhone"))
|
|
|
|
def test_generic_patterns(self):
|
|
"""Common patterns are generic."""
|
|
self.assertTrue(ble_alerter.is_generic_name("LE-Device"))
|
|
self.assertTrue(ble_alerter.is_generic_name("BLE-Sensor"))
|
|
self.assertTrue(ble_alerter.is_generic_name("Unknown Device"))
|
|
self.assertFalse(ble_alerter.is_generic_name("John's iPhone"))
|
|
|
|
def test_mac_format_names(self):
|
|
"""MAC-format names (dash or colon separated) are generic."""
|
|
self.assertTrue(ble_alerter.is_generic_name("45-48-39-7F-73-C7"))
|
|
self.assertTrue(ble_alerter.is_generic_name("45:48:39:7F:73:C7"))
|
|
self.assertTrue(ble_alerter.is_generic_name("aa:bb:cc:dd:ee:ff"))
|
|
# Partial MAC (not 6 groups) is NOT caught by regex — falls through to other checks
|
|
self.assertTrue(ble_alerter.is_generic_name("AABBCC")) # caught by all-hex check
|
|
|
|
def test_real_device_names(self):
|
|
"""Real device names are not generic."""
|
|
self.assertFalse(ble_alerter.is_generic_name("John's iPhone"))
|
|
self.assertFalse(ble_alerter.is_generic_name("Kelly's AirPods Pro"))
|
|
self.assertFalse(ble_alerter.is_generic_name("Pixel Watch"))
|
|
|
|
|
|
class TestFmtDuration(unittest.TestCase):
|
|
"""Test duration formatting."""
|
|
|
|
def test_seconds_only(self):
|
|
"""Format seconds."""
|
|
self.assertEqual(ble_alerter.fmt_duration(30), "30s")
|
|
self.assertEqual(ble_alerter.fmt_duration(59), "59s")
|
|
|
|
def test_minutes_only(self):
|
|
"""Format minutes."""
|
|
self.assertEqual(ble_alerter.fmt_duration(60), "1m")
|
|
self.assertEqual(ble_alerter.fmt_duration(120), "2m")
|
|
self.assertEqual(ble_alerter.fmt_duration(90), "1m 30s")
|
|
|
|
def test_hours_and_minutes(self):
|
|
"""Format hours + minutes."""
|
|
self.assertEqual(ble_alerter.fmt_duration(3600), "1h")
|
|
self.assertEqual(ble_alerter.fmt_duration(3900), "1h 5m")
|
|
self.assertEqual(ble_alerter.fmt_duration(7200), "2h")
|
|
|
|
|
|
class TestSendAlert(unittest.TestCase):
|
|
"""Test Matrix alert sending."""
|
|
|
|
def setUp(self):
|
|
"""Reset config before each test."""
|
|
ble_alerter.MATRIX_ACCESS_TOKEN = ""
|
|
ble_alerter.MATRIX_HOMESERVER = "https://m.test.com"
|
|
ble_alerter.MATRIX_ROOM_ID = "!abc123:m.test.com"
|
|
|
|
def test_no_token_skips_alert(self):
|
|
"""No token = no alert."""
|
|
ble_alerter.MATRIX_ACCESS_TOKEN = ""
|
|
with patch('ble_alerter.logging') as mock_log:
|
|
ble_alerter.send_alert("test")
|
|
mock_log.warning.assert_called()
|
|
|
|
@patch('ble_alerter.urllib.request.urlopen')
|
|
def test_alert_sent_to_matrix(self, mock_urlopen):
|
|
"""Alert is sent to Matrix API."""
|
|
ble_alerter.MATRIX_ACCESS_TOKEN = "syt_test"
|
|
ble_alerter.MATRIX_ROOM_ID = "!abc:m.test.com"
|
|
|
|
mock_urlopen.return_value.__enter__ = MagicMock(return_value=MagicMock())
|
|
mock_urlopen.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
ble_alerter.send_alert("[BLE] ARRIVED: test")
|
|
|
|
mock_urlopen.assert_called_once()
|
|
args, kwargs = mock_urlopen.call_args
|
|
self.assertIn("_matrix/client/v3/rooms", args[0].full_url)
|
|
|
|
@patch('ble_alerter.time.sleep')
|
|
@patch('ble_alerter.urllib.request.urlopen')
|
|
def test_send_alert_retries_on_429(self, mock_urlopen, mock_sleep):
|
|
"""429 response triggers retry with backoff, succeeds on second attempt."""
|
|
import urllib.error
|
|
ble_alerter.MATRIX_ACCESS_TOKEN = "syt_test"
|
|
ble_alerter.MATRIX_ROOM_ID = "!abc:m.test.com"
|
|
|
|
# First call raises 429, second succeeds
|
|
mock_urlopen.side_effect = [
|
|
urllib.error.HTTPError(None, 429, "Too Many Requests", {}, None),
|
|
MagicMock(__enter__=MagicMock(return_value=MagicMock()), __exit__=MagicMock(return_value=False)),
|
|
]
|
|
|
|
ble_alerter.send_alert("[BLE] test")
|
|
|
|
self.assertEqual(mock_urlopen.call_count, 2)
|
|
mock_sleep.assert_called_once_with(1) # 2^0 = 1s backoff
|
|
|
|
@patch('ble_alerter.time.sleep')
|
|
@patch('ble_alerter.urllib.request.urlopen')
|
|
def test_send_alert_gives_up_after_3_attempts(self, mock_urlopen, mock_sleep):
|
|
"""Persistent 429 gives up after 3 attempts."""
|
|
import urllib.error
|
|
ble_alerter.MATRIX_ACCESS_TOKEN = "syt_test"
|
|
ble_alerter.MATRIX_ROOM_ID = "!abc:m.test.com"
|
|
|
|
mock_urlopen.side_effect = urllib.error.HTTPError(None, 429, "Too Many Requests", {}, None)
|
|
|
|
ble_alerter.send_alert("[BLE] test")
|
|
|
|
self.assertEqual(mock_urlopen.call_count, 3)
|
|
self.assertEqual(mock_sleep.call_count, 2) # sleeps after attempt 0 and 1
|
|
|
|
|
|
class TestArrivalDeparture(unittest.IsolatedAsyncioTestCase):
|
|
"""Test arrival/departure detection."""
|
|
|
|
async def asyncSetUp(self):
|
|
"""Reset state before each test."""
|
|
ble_alerter.tracked = {}
|
|
ble_alerter.MATRIX_ACCESS_TOKEN = "syt_test"
|
|
ble_alerter.MODE = "open"
|
|
|
|
async def test_first_arrival_creates_entry(self):
|
|
"""First advertisement triggers arrival."""
|
|
with patch('ble_alerter.send_alert') as mock_send:
|
|
await ble_alerter.on_arrival("iPhone", "AA:BB:CC:DD:EE:FF", -65)
|
|
|
|
self.assertEqual(len(ble_alerter.tracked), 1)
|
|
self.assertIn("iPhone", ble_alerter.tracked)
|
|
mock_send.assert_called_once()
|
|
self.assertIn("ARRIVED", mock_send.call_args[0][0])
|
|
|
|
async def test_same_device_updates_timestamp(self):
|
|
"""Re-advertisement updates last_seen without new alert."""
|
|
with patch('ble_alerter.send_alert') as mock_send:
|
|
await ble_alerter.on_arrival("iPhone", "AA:BB:CC:DD:EE:FF", -65)
|
|
first_sent = mock_send.call_count
|
|
|
|
# Simulate re-advertisement
|
|
await ble_alerter.on_arrival("iPhone", "AA:BB:CC:DD:EE:FF", -63)
|
|
|
|
# Should still be 1 alert (no new arrival alert)
|
|
self.assertEqual(mock_send.call_count, first_sent)
|
|
self.assertEqual(len(ble_alerter.tracked), 1)
|
|
|
|
async def test_departure_alert(self):
|
|
"""Departure triggers alert with duration."""
|
|
with patch('ble_alerter.send_alert') as mock_send:
|
|
await ble_alerter.on_arrival("iPhone", "AA:BB:CC:DD:EE:FF", -65)
|
|
await ble_alerter.on_departure("iPhone")
|
|
|
|
# Should be 2 alerts: arrival + departure
|
|
self.assertEqual(mock_send.call_count, 2)
|
|
departure_msg = mock_send.call_args_list[1][0][0]
|
|
self.assertIn("DEPARTED", departure_msg)
|
|
self.assertIn("present", departure_msg)
|
|
|
|
|
|
class TestModeDetection(unittest.IsolatedAsyncioTestCase):
|
|
"""Test open vs known mode detection."""
|
|
|
|
async def asyncSetUp(self):
|
|
"""Reset state before each test."""
|
|
ble_alerter.tracked = {}
|
|
ble_alerter.MATRIX_ACCESS_TOKEN = "syt_test"
|
|
|
|
async def test_open_mode_accepts_named_devices(self):
|
|
"""Open mode: any good name is accepted."""
|
|
ble_alerter.MODE = "open"
|
|
with patch('ble_alerter.send_alert'):
|
|
await ble_alerter.on_arrival("Phone", "AA:BB:CC:DD:EE:FF", -65)
|
|
self.assertEqual(len(ble_alerter.tracked), 1)
|
|
|
|
async def test_open_mode_filters_generic(self):
|
|
"""Open mode: generic names are skipped (handled in scan callback)."""
|
|
ble_alerter.MODE = "open"
|
|
# The filtering happens in scan_loop callback, not on_arrival
|
|
# This test just verifies on_arrival stores the data
|
|
with patch('ble_alerter.send_alert'):
|
|
await ble_alerter.on_arrival("MyDevice", "AA:BB:CC:DD:EE:FF", -65)
|
|
self.assertEqual(len(ble_alerter.tracked), 1)
|
|
|
|
|
|
class TestLoadEnv(unittest.TestCase):
|
|
"""Test environment loading."""
|
|
|
|
def test_load_env_sets_globals(self):
|
|
"""load_env() populates global config."""
|
|
# Create temporary .env file
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.env', delete=False) as f:
|
|
f.write("MODE=known\n")
|
|
f.write("KNOWN_DEVICES=iPhone,AirPods\n")
|
|
f.write("DEPARTURE_TIMEOUT=120\n")
|
|
f.write("RSSI_MIN=-80\n")
|
|
env_path = f.name
|
|
|
|
try:
|
|
with patch('ble_alerter.Path') as mock_path:
|
|
mock_path.return_value.exists.return_value = True
|
|
mock_path.return_value.read_text.return_value = open(env_path).read()
|
|
|
|
ble_alerter.load_env()
|
|
|
|
# Verify globals were set (note: these are module-level, so check indirectly)
|
|
# This is a basic sanity check; full verification requires inspection
|
|
finally:
|
|
import os
|
|
os.unlink(env_path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|