yaris-obd/tests/test_app_settings.py

import os
import unittest
from unittest.mock import patch

from modules import app_cellular, app_settings


def valid_payload():
    return {
        "dashcam_upload_enabled": False,
        "trips_upload_enabled": True,
        "video_convert_enabled": True,
        "local_video_retention_days": 14,
        "cellular_enabled": True,
        "cellular_apn": "carrier.apn",
        "ntfy_enabled": True,
        "ntfy_url": "https://ntfy.example/private-topic",
        "ntfy_token": "secret-token",
        "clear_ntfy_url": False,
        "clear_ntfy_token": False,
        "alert_thresholds": {
            "temperature_c": 75.5,
            "disk_percent": 88,
            "tracking_backlog": 500,
            "video_backlog": 8,
            "gps_loss_seconds": 90,
            "repeat_seconds": 7200,
        },
    }


class SettingsUpdateTest(unittest.TestCase):
    def setUp(self):
        self.settings = {}

    def get_setting(self, key, default=None):
        return self.settings.get(key, default)

    def save_settings(self, settings):
        self.settings.update(settings)

    def update(self, payload):
        with (
            patch(
                "modules.app_settings.app_sql.get_setting",
                side_effect=self.get_setting,
            ),
            patch(
                "modules.app_settings.app_sql.set_settings",
                side_effect=self.save_settings,
            ),
        ):
            return app_settings.update_settings(payload)

    def test_valid_settings_are_saved_together(self):
        self.assertEqual(self.update(valid_payload()), {})

        self.assertEqual(self.settings["dashcam_upload_enabled"], "0")
        self.assertEqual(self.settings["local_video_retention_days"], "14")
        self.assertEqual(self.settings["cellular_enabled"], "1")
        self.assertEqual(self.settings["cellular_apn"], "carrier.apn")
        self.assertEqual(self.settings["alerts_enabled"], "1")
        self.assertEqual(
            self.settings["alert_ntfy_url"],
            "https://ntfy.example/private-topic",
        )
        self.assertEqual(self.settings["alert_ntfy_token"], "secret-token")
        self.assertEqual(self.settings["alert_temperature_c"], "75.5")

    def test_invalid_payload_does_not_save_any_settings(self):
        payload = valid_payload()
        payload["cellular_apn"] = "bad apn"
        payload["alert_thresholds"]["repeat_seconds"] = 30

        errors = self.update(payload)

        self.assertIn("cellular_apn", errors)
        self.assertIn("repeat_seconds", errors)
        self.assertEqual(self.settings, {})

    def test_enabled_transports_require_configuration(self):
        payload = valid_payload()
        payload["cellular_apn"] = ""
        payload["ntfy_url"] = ""

        errors = self.update(payload)

        self.assertIn("cellular_apn", errors)
        self.assertIn("ntfy_url", errors)

    def test_blank_secrets_are_preserved_and_explicit_clear_removes_them(self):
        self.settings.update(
            {
                "alert_ntfy_url": "https://ntfy.example/existing",
                "alert_ntfy_token": "existing-token",
            }
        )
        payload = valid_payload()
        payload["ntfy_url"] = ""
        payload["ntfy_token"] = ""
        payload["ntfy_enabled"] = False

        self.assertEqual(self.update(payload), {})
        self.assertEqual(
            self.settings["alert_ntfy_url"],
            "https://ntfy.example/existing",
        )
        self.assertEqual(self.settings["alert_ntfy_token"], "existing-token")

        payload["clear_ntfy_url"] = True
        payload["clear_ntfy_token"] = True
        self.assertEqual(self.update(payload), {})
        self.assertEqual(self.settings["alert_ntfy_url"], "")
        self.assertEqual(self.settings["alert_ntfy_token"], "")

    def test_fractional_integer_field_is_rejected(self):
        payload = valid_payload()
        payload["alert_thresholds"]["video_backlog"] = 1.5

        errors = self.update(payload)

        self.assertIn("video_backlog", errors)


class SettingsReadTest(unittest.TestCase):
    @patch("modules.app_settings.app_video.retention_days", return_value=30)
    @patch(
        "modules.app_settings.app_upload.is_dashcam_upload_enabled",
        return_value=True,
    )
    @patch("modules.app_settings.app_sql.get_setting")
    def test_get_settings_does_not_return_ntfy_secrets(
        self,
        get_setting,
        _upload_enabled,
        _retention_days,
    ):
        stored = {
            "cellular_enabled": "0",
            "cellular_apn": "",
            "alerts_enabled": "1",
            "alert_ntfy_url": "https://ntfy.example/private",
            "alert_ntfy_token": "secret-token",
        }
        get_setting.side_effect = lambda key, default=None: stored.get(key, default)

        settings = app_settings.get_settings()

        self.assertTrue(settings["ntfy_url_configured"])
        self.assertTrue(settings["ntfy_token_configured"])
        self.assertNotIn("ntfy_url", settings)
        self.assertNotIn("ntfy_token", settings)


class CellularSettingsTest(unittest.TestCase):
    @patch.dict(
        os.environ,
        {
            "YARIS_CELLULAR_ENABLED": "0",
            "YARIS_CELLULAR_APN": "legacy.apn",
        },
    )
    @patch("modules.app_cellular.app_sql.get_setting")
    def test_persisted_cellular_settings_apply_without_restart(self, get_setting):
        settings = {"cellular_enabled": "1", "cellular_apn": "live.apn"}
        get_setting.side_effect = lambda key, default=None: settings.get(key, default)

        self.assertTrue(app_cellular.is_configured())
        self.assertEqual(app_cellular.get_apn(), "live.apn")


if __name__ == "__main__":
    unittest.main()