test update

tests/test_alarmclock.py

@@ -0,0 +1,144 @@
+from contextlib import nullcontext
+import datetime
+import pickle
+import tempfile
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import Mock, patch
+
+sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'),
+                str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')]
+
+from alarmclock import AlarmClock
+
+
+class Harness(AlarmClock):
+    def __init__(self):
+        self.calendar = ['previous calendar']
+        self.calendar_retry_handle = None
+        self.calendar_retry_attempt = 0
+        self.dataset = {}
+        self.load_cal_from_ics = Mock(return_value=False)
+        self.run_in = Mock(side_effect=lambda callback, delay: f'timer-{delay}')
+        self.timer_running = Mock(return_value=True)
+        self.cancel_timer = Mock()
+        self.log_info = Mock()
+        self.log_warning = Mock()
+        self.log_error = Mock()
+        self.log_group = Mock(side_effect=lambda name: nullcontext())
+        self.update_workday_status = Mock()
+        self.update_wakeup_time = Mock()
+
+
+class CalendarRefreshTests(unittest.TestCase):
+    def test_last_success_survives_dataset_save_and_restore(self):
+        app = Harness()
+        app.calendar = []
+        app.load_cal_from_ics.return_value = True
+        app.refresh_cal()
+        last_success = app.calendar_last_success
+        self.assertIsNotNone(last_success)
+        with tempfile.TemporaryDirectory() as folder:
+            dataset_path = Path(folder) / 'alarm_clock.dataset'
+            app.get_dataset_name = Mock(return_value=str(dataset_path))
+            app.terminate()
+            restored = Harness()
+            with dataset_path.open('rb') as stream:
+                restored.dataset = pickle.load(stream)
+        self.assertEqual(restored.calendar_last_success, last_success)
+        restored._log_calendar_failure('request timed out')
+        restored.log_info.assert_called_once()
+        restored.log_warning.assert_not_called()
+        restored.log_error.assert_not_called()
+
+    def test_dataset_without_timestamp_has_no_known_success(self):
+        app = Harness()
+        app.dataset = {'other_data': 42}
+        self.assertIsNone(app.calendar_last_success)
+        app._log_calendar_failure('request timed out')
+        app.log_error.assert_called_once()
+        self.assertEqual(app.dataset, {'other_data': 42})
+
+    def test_failure_preserves_calendar_and_schedules_two_retries(self):
+        app = Harness()
+
+        self.assertFalse(app.refresh_cal())
+        self.assertEqual(app.calendar, ['previous calendar'])
+        app.run_in.assert_called_once_with(app.on_calendar_retry, 120)
+
+        app.on_calendar_retry({})
+        self.assertEqual(app.run_in.call_args_list[-1].args, (app.on_calendar_retry, 300))
+
+        app.on_calendar_retry({})
+        self.assertEqual(app.run_in.call_count, 2)
+
+    def test_failure_level_depends_on_last_success(self):
+        now = datetime.datetime(2026, 11, 4, 12, tzinfo=datetime.timezone.utc)
+        for hours, level in ((0, 'info'), (23.999, 'info'), (24, 'warning'),
+                             (47.999, 'warning'), (48, 'error'), (72, 'error'),
+                             (None, 'error')):
+            with self.subTest(hours=hours):
+                app = Harness()
+                last_success = None if hours is None else now - datetime.timedelta(hours=hours)
+                app.calendar_last_success = last_success
+                with patch('alarmclock.datetime.datetime') as clock:
+                    clock.now.return_value = now
+                    app._log_calendar_failure('request timed out')
+                getattr(app, f'log_{level}').assert_called_once()
+                for other in {'info', 'warning', 'error'} - {level}:
+                    getattr(app, f'log_{other}').assert_not_called()
+                self.assertEqual(app.calendar_last_success, last_success)
+                message = getattr(app, f'log_{level}').call_args.args[0]
+                self.assertIn('request timed out', message)
+                if last_success:
+                    self.assertIn(last_success.isoformat(), message)
+
+    def test_success_updates_timestamp_even_when_calendar_is_unchanged(self):
+        app = Harness()
+        app.calendar = []
+        app.calendar_last_success = datetime.datetime(2026, 11, 1, tzinfo=datetime.timezone.utc)
+        app.calendar_retry_attempt = 2
+        app.calendar_retry_handle = 'pending-timer'
+        app.load_cal_from_ics.return_value = True
+        now = datetime.datetime(2026, 11, 4, tzinfo=datetime.timezone.utc)
+        with patch('alarmclock.datetime.datetime') as clock:
+            clock.now.return_value = now
+            self.assertFalse(app.refresh_cal())
+        app.cancel_timer.assert_called_once_with('pending-timer')
+        self.assertEqual(app.calendar_last_success, now)
+        self.assertEqual(app.calendar_retry_attempt, 0)
+
+    @patch('alarmclock.get')
+    def test_timeout_uses_freshness_level_and_preserves_calendar(self, get):
+        import requests
+        app = Harness()
+        app.calendar_last_success = datetime.datetime.now(datetime.timezone.utc)
+        last_success = app.calendar_last_success
+        app.load_cal_from_ics = lambda calendar: AlarmClock.load_cal_from_ics(app, calendar)
+        get.side_effect = requests.exceptions.ReadTimeout()
+        self.assertFalse(app.refresh_cal())
+        app.log_warning.assert_not_called()
+        app.log_error.assert_not_called()
+        self.assertTrue(any('ReadTimeout' in call.args[0] for call in app.log_info.call_args_list))
+        self.assertEqual(app.calendar_last_success, last_success)
+        self.assertEqual(app.calendar, ['previous calendar'])
+        app.run_in.assert_called_once_with(app.on_calendar_retry, 120)
+
+    @patch('alarmclock.icalendar.Calendar.from_ical')
+    @patch('alarmclock.get')
+    def test_http_request_uses_short_timeouts_and_validates_status(self, get, from_ical):
+        response = Mock(content=b'BEGIN:VCALENDAR\r\nEND:VCALENDAR\r\n')
+        get.return_value = response
+        from_ical.return_value.walk.return_value = []
+        app = Harness()
+
+        self.assertTrue(AlarmClock.load_cal_from_ics(app, []))
+
+        self.assertEqual(get.call_args.kwargs['timeout'], (3, 5))
+        response.raise_for_status.assert_called_once_with()
+        from_ical.assert_called_once_with(response.content)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_informationscollector.py

@@ -1,7 +1,9 @@
 from pathlib import Path
 import sys
+import io
+import json
 import unittest
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'),
                 str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')]
@@ -41,5 +43,51 @@ class WarningCollectorTests(unittest.TestCase):
         )
 
 
+class AppDaemonDockerVersionTests(unittest.TestCase):
+    def setUp(self):
+        self.app = object.__new__(InformationsCollector)
+        self.app.log_warning = Mock()
+        self.app.dataset = {'appdaemon_latest_official_version': '4.5.14'}
+        self.app.appdaemon_version_sensor = Mock()
+
+    @staticmethod
+    def response(tags, next_url=None):
+        return io.StringIO(json.dumps({'results': tags, 'next': next_url}))
+
+    def test_selects_highest_stable_published_tag_across_pages(self):
+        pages = [
+            self.response([{'name': name, 'images': [{}]} for name in
+                           ['latest', 'dev', '4.5.14rc1', '4.5.9']], 'https://hub.docker.com/page2'),
+            self.response([{'name': '4.5.13', 'images': [{}]},
+                           {'name': '4.5.14', 'images': []},
+                           {'name': '4.5.12', 'images': [{}]}]),
+        ]
+        with patch('informationscollector.urlopen', side_effect=pages) as fetch:
+            self.assertEqual(self.app._get_latest_official_appdaemon_version(), '4.5.13')
+        self.assertEqual(fetch.call_count, 2)
+        fetch.assert_any_call(self.app.APPDAEMON_DOCKER_TAGS_URL, timeout=10)
+        self.app.log_warning.assert_not_called()
+
+    def test_empty_stable_results_are_reported(self):
+        with patch('informationscollector.urlopen', return_value=self.response([])):
+            self.assertIsNone(self.app._get_latest_official_appdaemon_version())
+        self.app.log_warning.assert_called_once()
+
+    def test_sensor_uses_docker_cache_without_reusing_pypi_cache(self):
+        with patch.object(InformationsCollector, '_get_installed_appdaemon_version', return_value='4.5.10'):
+            with patch('informationscollector.urlopen', side_effect=OSError('offline')):
+                self.app.update_appdaemon_version_sensor()
+            self.app.appdaemon_version_sensor.set_state.assert_called_with(
+                state='4.5.10', attributes={'latest_official_version': None, 'update_available': None})
+            with patch('informationscollector.urlopen', return_value=self.response([
+                    {'name': '4.5.13', 'images': [{}]}])):
+                self.app.update_appdaemon_version_sensor()
+            self.assertEqual(self.app.dataset['appdaemon_latest_docker_version'], '4.5.13')
+            with patch('informationscollector.urlopen', side_effect=OSError('offline')):
+                self.app.update_appdaemon_version_sensor()
+            self.app.appdaemon_version_sensor.set_state.assert_called_with(
+                state='4.5.10', attributes={'latest_official_version': '4.5.13', 'update_available': True})
+
+
 if __name__ == '__main__':
     unittest.main()

tests/test_notificationsmanager_logging.py

@@ -0,0 +1,78 @@
+"""Notification logging must avoid stack inspection without changing delivery."""
+import logging
+from pathlib import Path
+import sys
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock, call, patch
+
+sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'),
+                str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')]
+from appdaemon.adapi import ADAPI
+from notificationsmanager import NotificationsManager, Notification, SmartCondition
+
+
+class NotificationLoggingTests(unittest.TestCase):
+    def setUp(self):
+        self.api = ADAPI.__new__(ADAPI)
+        self.api.AD = SimpleNamespace(config=SimpleNamespace(ascii_encode=False))
+        self.api.logger = Mock()
+        self.sinks = {'main_log': Mock(), 'error_log': Mock()}
+        self.api.get_user_log = Mock(side_effect=self.sinks.__getitem__)
+        self.app = NotificationsManager.__new__(NotificationsManager)
+        self.app.get_ad_api = Mock(return_value=self.api)
+        self.app.args = {}
+        self.app.call_service = Mock()
+        self.app.run_in = Mock()
+        self.app.stop_app = Mock()
+        self.app._config_model = SimpleNamespace(name='notifications_manager')
+        self.app.initialize()
+        self.config = {
+            'message': {'title': 'Alarme', 'content': 'Activée', 'timeout': 10,
+                        'data': {'tag': 'alarm'}},
+            'recipients': {'phones': {'services': ['notify/one', 'notify/two']}}
+        }
+        self.notification = Notification(self.app, {'message': {}, 'recipients': {}}, 'alarm')
+        self.notification.config = self.config
+
+    def test_send_and_clear_keep_delivery_without_inspecting_stack(self):
+        with patch('appdaemon.adapi.inspect.stack', side_effect=AssertionError('stack inspection')):
+            self.notification.send_notification('phones')
+            self.notification.clear_notification('phones', 'alarm')
+        self.app.call_service.assert_has_calls([
+            call('notify/one', title='Alarme', message='Activée', data={'tag': 'alarm'}),
+            call('notify/two', title='Alarme', message='Activée', data={'tag': 'alarm'}),
+            call('notify/one', message='clear_notification', data={'tag': 'alarm'}),
+            call('notify/two', message='clear_notification', data={'tag': 'alarm'}),
+        ])
+        self.app.run_in.assert_called_once_with(self.notification.on_timeout_expired, 10,
+                                                tag='alarm', recipient='phones')
+        self.assertEqual(self.api.logger.log.call_count, 4)
+        self.api.get_user_log.assert_not_called()
+
+    def test_errors_use_app_logger_and_error_log_and_stop(self):
+        with patch('appdaemon.adapi.inspect.stack', side_effect=AssertionError('stack inspection')):
+            self.app.log_error('Échec', stop_app=True)
+        for sink in (self.api.logger, self.sinks['error_log']):
+            self.assertEqual(sink.log.call_args.args, (logging.ERROR, 'Échec'))
+        self.sinks['main_log'].log.assert_not_called()
+        self.app.stop_app.assert_called_once_with('notifications_manager')
+
+    def test_smartcondition_fallback_uses_initialized_interface(self):
+        self.config['recipients']['phones']['conditions'] = True
+        self.app.args = {'notifications': {'alarm': self.config}}
+        real_class = SmartCondition.Evaluator
+        def factory(api, *args, **kwargs):
+            condition = real_class.__new__(real_class)
+            condition._Evaluator__logger_interface = None
+            condition._Evaluator__appdaemon_api = api
+            condition._Evaluator__log_info('Condition évaluée')
+            return condition
+        with patch('appdaemon.adapi.inspect.stack', side_effect=AssertionError('stack inspection')):
+            with patch.object(SmartCondition, 'Evaluator', side_effect=factory):
+                self.app.initialize()
+        self.assertEqual(self.api.logger.log.call_args.args, (logging.INFO, 'Condition évaluée'))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_vertex2.py

@@ -20,6 +20,7 @@ class Harness(Vertex2):
         self._status_requested = threading.Event()
         self._last_logged_status = None
         self._input_5v_states = {}
+        self._cancel_input_5v_refresh()
         self._schedule_input_5v_refresh = Mock()
         self._connection_lock = threading.Lock()
         self._interrupt_connection = None
@@ -197,6 +198,54 @@ class PowerTests(unittest.TestCase):
         app.log_warning.assert_not_called()
 
 
+class StatusTimeoutDiagnosticsTests(unittest.TestCase):
+    def test_timeout_reports_missing_fields_and_partial_response(self):
+        for silent in (True, False):
+            with self.subTest(silent=silent):
+                app = Harness()
+                app.video_tx_signal_sensor = Mock()
+                app.video_tx_signal_sensors = {}
+                app.allm_sensor = Mock()
+                app.allm_sensors = {}
+                app._reset_no_signal_spam = Mock()
+                app.log_warning.side_effect = lambda message: app._interrupt_stop.set()
+                connection = Mock()
+                fields = ('rx0', 'rx1', 'tx0', 'tx1', 'tx0sink', 'tx1sink',
+                          'aud0', 'aud1', 'audout', 'spd0')
+                response = ('\n'.join(f'{field.upper()}: no signal' for field in fields)
+                            + '\ninsel 0 0\nint\nSPD1: incom').encode('ascii')
+                clock = [100.0]
+                calls = [0]
+
+                def recv(size):
+                    calls[0] += 1
+                    if calls[0] == 1:
+                        return response
+                    clock[0] = 106.0
+                    if silent:
+                        raise socket.timeout()
+                    return b'plete'
+
+                connection.recv.side_effect = recv
+                context = Mock(__enter__=Mock(return_value=connection),
+                               __exit__=Mock(return_value=False))
+                with patch('vertex2.socket.create_connection', return_value=context), \
+                        patch('vertex2.time.monotonic', side_effect=lambda: clock[0]):
+                    app._interrupt_listener(app._interrupt_stop)
+                app.log_warning.assert_called_once()
+                message = app.log_warning.call_args.args[0]
+                self.assertIn("missing=['spd1']", message)
+                self.assertIn('elapsed=6.00s, limit=5s', message)
+                self.assertIn('interrupts=1', message)
+                self.assertIn(f'bytes={len(response) + (0 if silent else 5)}', message)
+                self.assertIn("'insel 0 0'", message)
+                partial = 'incom' if silent else 'incomplete'
+                self.assertIn(f"partial_line='SPD1: {partial}'", message)
+                self.assertIn('reconfigure=True', message)
+                app.run_in.assert_not_called()
+                app.unreachable_sensor.set_state.assert_called_with(state='on')
+
+
 class ReconfigurationTests(unittest.TestCase):
     def test_reconnect_reapplies_current_plus5_and_selections(self):
         app = Harness()
@@ -360,10 +409,11 @@ class StatusChangeTests(unittest.TestCase):
             app._refresh_input_5v_sensors.reset_mock()
             app._apply_vertex2_status(dict(self.status, **metadata))
             app._refresh_input_5v_sensors.assert_not_called()
-            app.run_in.assert_called_once_with(app._refresh_input_5v_sensors, 0.1)
+            app.run_in.assert_called_once_with(app._on_input_5v_refresh, 1,
+                                                generation=app._input_5v_refresh_generation)
             callback = app.run_in.call_args.args[0]
-            callback({})
+            callback(app.run_in.call_args.kwargs)
-            app._refresh_input_5v_sensors.assert_called_once_with({})
+            app._refresh_input_5v_sensors.assert_called_once_with()
 
     def test_identical_status_is_silent_but_still_applied(self):
         app = self.app
@@ -668,6 +718,62 @@ class InputSelectTests(unittest.TestCase):
 
 
 class Input5VTests(unittest.TestCase):
+    def test_bursts_share_one_read_and_retry_then_recover(self):
+        from requests.exceptions import ConnectTimeout
+        app = self.app
+        del app._schedule_input_5v_refresh
+        for _ in range(10):
+            app._schedule_input_5v_refresh()
+        self.assertEqual(app.run_in.call_count, 1)
+        first = app.run_in.call_args
+        with patch('vertex2.get', side_effect=ConnectTimeout('test')):
+            first.args[0](first.kwargs)
+        app.log_warning.assert_not_called()
+        self.assertEqual(app.run_in.call_count, 2)
+        retry = app.run_in.call_args
+        self.assertEqual(retry.args[1], 5)
+        for _ in range(10):
+            app._schedule_input_5v_refresh()
+        self.assertEqual(app.run_in.call_count, 2)
+        with patch('vertex2.get') as get:
+            get.return_value.json.return_value = {'rx0in5v': '2'}
+            retry.args[0](retry.kwargs)
+            get.assert_called_once()
+        self.assertEqual(app._input_5v_states[0], 'on')
+        self.assertEqual(app._input_5v_failures, 0)
+        self.assertIsNone(app._input_5v_refresh_handle)
+        app.log_warning.assert_not_called()
+        self.assertTrue(any('restored after' in c.args[0] for c in app.log_info.call_args_list))
+
+    def test_failed_retry_warns_once_without_an_endless_retry_loop(self):
+        from requests.exceptions import ConnectTimeout
+        app = self.app
+        del app._schedule_input_5v_refresh
+        with patch('vertex2.get', side_effect=ConnectTimeout('test')):
+            app._refresh_input_5v_sensors()
+            retry = app.run_in.call_args
+            retry.args[0](retry.kwargs)
+            app._refresh_input_5v_sensors()
+        app.log_warning.assert_called_once()
+        self.assertIn('attempt=2', app.log_warning.call_args.args[0])
+        self.assertEqual(app.run_in.call_count, 1)
+
+    def test_power_off_cancels_retry_and_discards_stale_callback(self):
+        from requests.exceptions import ConnectTimeout
+        app = self.app
+        del app._schedule_input_5v_refresh
+        with patch('vertex2.get', side_effect=ConnectTimeout('test')):
+            app._refresh_input_5v_sensors()
+        retry = app.run_in.call_args
+        handle = app._input_5v_refresh_handle
+        app._suspend_vertex2()
+        app.cancel_timer.assert_any_call(handle, silent=True)
+        # Even if power returns before a cancelled callback arrives, ignore it.
+        with patch('vertex2.get') as get:
+            retry.args[0](retry.kwargs)
+            get.assert_not_called()
+        self.assertEqual(app._input_5v_failures, 0)
+
     def setUp(self):
         self.app = app = SelectHarness()
         app.input_names = {'AppleTV': 0, 'Xbox': 1}