import unittest from unittest.mock import patch import requests from modules import app_alerts def condition(active=True): return { "active": active, "value": {"value": 1}, "message": "Test alert", "priority": "default", "tags": ["test"], } class AlertEngineTest(unittest.TestCase): def setUp(self): self.states = {} self.saved = {} self.config = { "enabled": True, "ntfy_url": "http://127.0.0.1/topic", "ntfy_token": "", "alert_repeat_seconds": 3600, "errors": {}, } self.send_calls = [] def sender(self, *args): self.send_calls.append(args) def process(self, active, now): with ( patch( "modules.app_alerts.app_sql.get_alert_states", return_value=self.states, ), patch( "modules.app_alerts.app_sql.save_alert_state", side_effect=lambda key, state: self.saved.update({key: state}), ), ): app_alerts.process_conditions( {"mil": condition(active)}, self.config, now=now, sender=self.sender, ) self.states = {key: value.copy() for key, value in self.saved.items()} def test_sends_once_on_transition_and_repeats_after_cooldown(self): self.process(True, 100) self.process(True, 200) self.process(True, 3800) self.assertEqual(len(self.send_calls), 2) self.assertEqual(self.states["mil"]["last_sent_at"], 3800) def test_sends_recovery_after_active_alert(self): self.process(True, 100) self.process(False, 200) self.assertEqual(len(self.send_calls), 2) self.assertTrue(self.send_calls[1][2].startswith("Resolved:")) self.assertFalse(self.states["mil"]["recovery_pending"]) def test_failed_send_is_retried_after_retry_interval(self): attempts = [] def failing_sender(*args): attempts.append(args) raise requests.ConnectionError("offline") with ( patch( "modules.app_alerts.app_sql.get_alert_states", return_value={}, ), patch( "modules.app_alerts.app_sql.save_alert_state", side_effect=lambda key, state: self.saved.update({key: state}), ), ): app_alerts.process_conditions( {"mil": condition(True)}, self.config, now=100, sender=failing_sender, ) self.states = {"mil": self.saved["mil"].copy()} self.assertEqual( self.states["mil"]["last_error"], "Notification connection failed", ) self.process(True, 200) self.process(True, 401) self.assertEqual(len(attempts), 1) self.assertEqual(len(self.send_calls), 1) self.assertEqual(self.states["mil"]["last_sent_at"], 401) def test_disabled_alerts_track_condition_without_sending(self): self.config["enabled"] = False self.process(True, 100) self.assertEqual(self.send_calls, []) self.assertTrue(self.states["mil"]["active"]) self.assertIsNone(self.states["mil"]["last_sent_at"]) class NtfyTransportTest(unittest.TestCase): @patch("modules.app_alerts.requests.post") def test_sends_ntfy_headers_body_and_token(self, post): response = post.return_value response.raise_for_status.return_value = None app_alerts.send_ntfy( "https://ntfy.example/topic", "secret", "Alert title", "Alert body", "high", ["warning", "car"], ) post.assert_called_once_with( "https://ntfy.example/topic", data=b"Alert body", headers={ "Title": "Alert title", "Priority": "high", "Tags": "warning,car", "Authorization": "Bearer secret", }, timeout=5, ) response.raise_for_status.assert_called_once_with() def test_transport_error_does_not_expose_url(self): response = requests.Response() response.status_code = 401 error = requests.HTTPError( "401 for https://ntfy.example/private-topic", response=response, ) message = app_alerts._transport_error(error) self.assertEqual(message, "Notification server returned HTTP 401") self.assertNotIn("private-topic", message) class GpsLossTest(unittest.TestCase): def setUp(self): app_alerts._moving_started_at = None def test_waits_for_full_timeout_before_first_fix(self): self.assertFalse( app_alerts.gps_loss_while_moving(20, None, 60, now=100) ) self.assertFalse( app_alerts.gps_loss_while_moving(20, None, 60, now=159) ) self.assertTrue( app_alerts.gps_loss_while_moving(20, None, 60, now=160) ) def test_stopping_resets_moving_timer(self): app_alerts.gps_loss_while_moving(20, None, 60, now=100) self.assertFalse( app_alerts.gps_loss_while_moving(0, None, 60, now=170) ) self.assertFalse( app_alerts.gps_loss_while_moving(20, None, 60, now=171) ) class PowerConditionTest(unittest.TestCase): @patch("modules.app_alerts.app_sql.set_setting") @patch("modules.app_alerts.app_sql.get_setting") def test_new_historical_flag_stays_active_during_grace( self, get_setting, set_setting ): get_setting.side_effect = lambda key, default=None: { "alert_throttling_seen": "[]", "alert_power_event_until": "0", }.get(key, default) power = app_alerts._power_condition( { "current": [], "occurred": ["under_voltage"], "raw": "0x10000", }, now=100, ) self.assertTrue(power["active"]) self.assertEqual(power["newly_occurred"], ["under_voltage"]) self.assertEqual(set_setting.call_count, 2) def test_cleared_firmware_flag_can_alert_again(self): settings = { "alert_throttling_seen": '["under_voltage"]', "alert_power_event_until": "0", } def get_setting(key, default=None): return settings.get(key, default) def set_setting(key, value): settings[key] = value with ( patch( "modules.app_alerts.app_sql.get_setting", side_effect=get_setting, ), patch( "modules.app_alerts.app_sql.set_setting", side_effect=set_setting, ), ): cleared = app_alerts._power_condition( {"current": [], "occurred": [], "raw": "0x0"}, now=100, ) recurred = app_alerts._power_condition( { "current": [], "occurred": ["under_voltage"], "raw": "0x10000", }, now=200, ) self.assertFalse(cleared["active"]) self.assertTrue(recurred["active"]) self.assertEqual(recurred["newly_occurred"], ["under_voltage"]) @patch("modules.app_alerts.app_sql.set_setting") @patch("modules.app_alerts.app_sql.get_setting") def test_unavailable_read_preserves_seen_flags( self, get_setting, set_setting ): get_setting.side_effect = lambda key, default=None: { "alert_throttling_seen": '["under_voltage"]', "alert_power_event_until": "0", }.get(key, default) power = app_alerts._power_condition(None, now=100) self.assertFalse(power["active"]) set_setting.assert_not_called() class DashcamUsageTest(unittest.TestCase): def setUp(self): app_alerts._last_dashcam_check_at = None app_alerts._last_dashcam_usage = {"status": "not_checked"} @patch("modules.app_alerts.app_dash.dashcam_storage_usage") def test_caches_dashcam_usage_for_five_minutes(self, storage_usage): storage_usage.return_value = { "status": "ok", "used_percent": 75, } first = app_alerts._dashcam_usage(now=100) cached = app_alerts._dashcam_usage(now=200) refreshed = app_alerts._dashcam_usage(now=401) self.assertEqual(first, cached) self.assertEqual(refreshed["used_percent"], 75) self.assertEqual(storage_usage.call_count, 2) class DiagnosticFreshnessTest(unittest.TestCase): def setUp(self): self.original_mil = app_alerts.app_obd.mil_on self.original_mil_updated = app_alerts.app_obd.mil_updated_at self.original_dtc = app_alerts.app_obd.dtc_codes self.original_dtc_updated = app_alerts.app_obd.dtc_updated_at def tearDown(self): app_alerts.app_obd.mil_on = self.original_mil app_alerts.app_obd.mil_updated_at = self.original_mil_updated app_alerts.app_obd.dtc_codes = self.original_dtc app_alerts.app_obd.dtc_updated_at = self.original_dtc_updated @patch("modules.app_obd.time.monotonic", return_value=100) def test_stale_mil_and_dtc_values_are_suppressed(self, _monotonic): app_alerts.app_obd.mil_on = True app_alerts.app_obd.mil_updated_at = 50 app_alerts.app_obd.dtc_codes = [{"code": "P0001"}] app_alerts.app_obd.dtc_updated_at = 50 self.assertIsNone(app_alerts.app_obd.get_fresh_mil()) self.assertEqual(app_alerts.app_obd.get_fresh_dtc_codes(), []) if __name__ == "__main__": unittest.main()