diff --git a/tests/test_alarmclock.py b/tests/test_alarmclock.py new file mode 100644 index 0000000..d66a448 --- /dev/null +++ b/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() diff --git a/tests/test_informationscollector.py b/tests/test_informationscollector.py index 4bf118d..a3557fe 100644 --- a/tests/test_informationscollector.py +++ b/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() diff --git a/tests/test_notificationsmanager_logging.py b/tests/test_notificationsmanager_logging.py new file mode 100644 index 0000000..3c26e5f --- /dev/null +++ b/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() diff --git a/tests/test_vertex2.py b/tests/test_vertex2.py index eabc3ad..a76b4d5 100644 --- a/tests/test_vertex2.py +++ b/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({}) - app._refresh_input_5v_sensors.assert_called_once_with({}) + callback(app.run_in.call_args.kwargs) + 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}