yaris-obd/tests/test_app_alerts.py

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()

    @patch("modules.app_alerts.app_cellular.http_post")
    @patch(
        "modules.app_alerts.app_cellular.is_configured",
        return_value=True,
    )
    @patch("modules.app_alerts.requests.post")
    def test_cellular_is_primary_when_configured(
        self,
        post,
        _is_configured,
        cellular_post,
    ):
        cellular_post.return_value = (200, b"ok")

        app_alerts.send_ntfy(
            "https://ntfy.example/topic",
            "secret",
            "Alert title",
            "Alert body",
            "high",
            ["warning"],
        )

        cellular_post.assert_called_once_with(
            "https://ntfy.example/topic",
            b"Alert body",
            {
                "Title": "Alert title",
                "Priority": "high",
                "Tags": "warning",
                "Authorization": "Bearer secret",
            },
            content_type="text/plain; charset=utf-8",
        )
        post.assert_not_called()

    @patch("modules.app_alerts.app_cellular.http_post")
    @patch(
        "modules.app_alerts.app_cellular.is_configured",
        return_value=True,
    )
    @patch("modules.app_alerts.requests.post")
    def test_cellular_http_error_falls_back_to_wifi(
        self,
        post,
        _is_configured,
        cellular_post,
    ):
        cellular_post.return_value = (503, b"error")
        post.return_value.raise_for_status.return_value = None

        app_alerts.send_ntfy(
            "https://ntfy.example/topic",
            "",
            "Alert title",
            "Alert body",
        )

        cellular_post.assert_called_once()
        post.assert_called_once()

    @patch("modules.app_alerts.app_cellular.http_post")
    @patch(
        "modules.app_alerts.app_cellular.is_configured",
        return_value=True,
    )
    @patch("modules.app_alerts.requests.post")
    def test_wifi_error_is_reported_after_cellular_failure(
        self,
        post,
        _is_configured,
        cellular_post,
    ):
        cellular_post.side_effect = app_alerts.app_cellular.CellularError(
            "modem offline"
        )
        post.side_effect = requests.ReadTimeout("response timed out")

        with self.assertRaises(requests.ReadTimeout):
            app_alerts.send_ntfy(
                "https://ntfy.example/topic",
                "",
                "Alert title",
                "Alert body",
            )

        cellular_post.assert_called_once()

    @patch("modules.app_alerts.app_cellular.http_post")
    @patch(
        "modules.app_alerts.app_cellular.is_configured",
        return_value=True,
    )
    @patch("modules.app_alerts.requests.post")
    def test_both_transport_failures_do_not_expose_private_url(
        self,
        post,
        _is_configured,
        cellular_post,
    ):
        post.side_effect = requests.ConnectionError("offline")
        cellular_post.return_value = (503, b"error")

        with self.assertRaises(requests.ConnectionError) as raised:
            app_alerts.send_ntfy(
                "https://ntfy.example/private-topic",
                "",
                "Alert title",
                "Alert body",
            )

        self.assertEqual(
            app_alerts._transport_error(raised.exception),
            "Notification connection failed",
        )

    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()