quoije/yaris-obd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add offline and ntfy alerts

Browse Source 
Co-authored-by: Copilot <[email protected]>
This commit is contained in:
Mathieu Moran
2026-08-09 20:28:42 -04:00
co-authored by Copilot
parent 8b664667e5
commit 96332adab6
12 changed files with 1098 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_app_alerts.py
+307
View File
@@ -0,0 +1,307 @@
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()