From 485b124c116a4e33e67c825c121fed0255cec4b4 Mon Sep 17 00:00:00 2001 From: Cobra Date: Fri, 10 Apr 2026 06:31:41 -0400 Subject: [PATCH] Add unit tests for BLE alerter (mocked bleak, async scenarios) --- net_alerter/test_ble_alerter.py | 213 ++++++++++++++++++++++++++++++++ 1 file changed, 213 insertions(+) create mode 100755 net_alerter/test_ble_alerter.py diff --git a/net_alerter/test_ble_alerter.py b/net_alerter/test_ble_alerter.py new file mode 100755 index 0000000..ecfaf9b --- /dev/null +++ b/net_alerter/test_ble_alerter.py @@ -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()