145 lines
6.3 KiB
Python
145 lines
6.3 KiB
Python
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()
|