Add unit tests for BLE alerter (mocked bleak, async scenarios)
This commit is contained in:
Cobra
Executable
+213
@@ -0,0 +1,213 @@
|
||||
#!/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"))
|
||||
self.assertFalse(ble_alerter.is_generic_name("abc"))
|
||||
|
||||
def test_all_hex(self):
|
||||
"""Pure hex is generic."""
|
||||
self.assertTrue(ble_alerter.is_generic_name("ABCDEF"))
|
||||
self.assertTrue(ble_alerter.is_generic_name("123456"))
|
||||
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_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)
|
||||
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user