diff --git a/emmental/alarm.py b/emmental/alarm.py new file mode 100644 index 0000000..63ae2ac --- /dev/null +++ b/emmental/alarm.py @@ -0,0 +1,48 @@ +# Copyright 2023 (c) Anna Schumaker. +"""Functions for configuring a callback at a specific time.""" +import datetime +import math +from gi.repository import GLib + +_GSOURCE_MAPPING = dict() +_NEXT_ALARM_ID = 1 + + +def _calc_seconds(time: datetime.time) -> int: + """Calculate the number of seconds until the given time.""" + now = datetime.datetime.now() + then = datetime.datetime.combine(now.date(), time) + + if now >= then: + then += datetime.timedelta(days=1) + + return math.ceil((then - now).total_seconds()) + + +def __set_alarm(time: datetime.time, func: callable, alarm_id: int) -> None: + gsrcid = GLib.timeout_add_seconds(_calc_seconds(time), _do_alarm, + time, func, alarm_id) + _GSOURCE_MAPPING[alarm_id] = gsrcid + return alarm_id + + +def _do_alarm(time: datetime.time, func: callable, alarm_id: int) -> bool: + """Run an alarm callback.""" + func() + __set_alarm(time, func, alarm_id) + return GLib.SOURCE_REMOVE + + +def set_alarm(time: datetime.time, func: callable) -> int: + """Register a callback to be called at a specific time.""" + global _NEXT_ALARM_ID + + res = __set_alarm(time, func, _NEXT_ALARM_ID) + _NEXT_ALARM_ID += 1 + return res + + +def cancel_alarm(alarm_id: int) -> None: + """Cancel an alarm.""" + GLib.source_remove(_GSOURCE_MAPPING[alarm_id]) + del _GSOURCE_MAPPING[alarm_id] diff --git a/tests/test_alarm.py b/tests/test_alarm.py new file mode 100644 index 0000000..b7d1ba0 --- /dev/null +++ b/tests/test_alarm.py @@ -0,0 +1,72 @@ +# Copyright 2023 (c) Anna Schumaker. +"""Test our functions for callbacks at a specific time.""" +import datetime +import unittest.mock +import emmental.alarm +from gi.repository import GLib + + +class TestAlarm(unittest.TestCase): + """Test case for callbacks at a specific time.""" + + def setUp(self): + """Set up common variables.""" + emmental.alarm._GSOURCE_MAPPING.clear() + emmental.alarm._NEXT_ALARM_ID = 1 + self.midnight = datetime.time(hour=0, minute=0, second=0) + + def test_state(self): + """Test our global state.""" + self.assertDictEqual(emmental.alarm._GSOURCE_MAPPING, {}) + self.assertEqual(emmental.alarm._NEXT_ALARM_ID, 1) + + def test_calc_seconds(self): + """Test calculating the seconds until the next alarm.""" + now = datetime.datetime.now() + time = (now + datetime.timedelta(minutes=2)).time() + self.assertEqual(emmental.alarm._calc_seconds(time), 120) + + time = (now - datetime.timedelta(minutes=2)).time() + self.assertEqual(emmental.alarm._calc_seconds(time), 86280) + + @unittest.mock.patch("gi.repository.GLib.timeout_add_seconds") + def test_set_alarm(self, mock_timeout_add: unittest.mock.Mock): + """Test setting an alarm.""" + callback = unittest.mock.Mock() + seconds = emmental.alarm._calc_seconds(self.midnight) + mock_timeout_add.return_value = 42 + + srcid = emmental.alarm.set_alarm(self.midnight, callback) + mock_timeout_add.assert_called_with(seconds, emmental.alarm._do_alarm, + self.midnight, callback, 1) + self.assertEqual(srcid, 1) + self.assertEqual(emmental.alarm._NEXT_ALARM_ID, 2) + self.assertEqual(emmental.alarm._GSOURCE_MAPPING[1], 42) + + @unittest.mock.patch("gi.repository.GLib.source_remove") + @unittest.mock.patch("gi.repository.GLib.timeout_add_seconds") + def test_cancel_alarm(self, mock_timeout_add: unittest.mock.Mock, + mock_source_remove: unittest.mock.Mock): + """Test cancelling an alarm.""" + callback = unittest.mock.Mock() + mock_timeout_add.return_value = 42 + srcid = emmental.alarm.set_alarm(self.midnight, callback) + + emmental.alarm.cancel_alarm(srcid) + mock_source_remove.assert_called_with(42) + self.assertNotIn(srcid, emmental.alarm._GSOURCE_MAPPING) + + @unittest.mock.patch("gi.repository.GLib.timeout_add_seconds") + def test_do_alarm(self, mock_timeout_add: unittest.mock.Mock): + """Test triggering an alarm.""" + callback = unittest.mock.Mock() + seconds = emmental.alarm._calc_seconds(self.midnight) + emmental.alarm._GSOURCE_MAPPING[1] = 2 + mock_timeout_add.return_value = 42 + + self.assertEqual(emmental.alarm._do_alarm(self.midnight, callback, 1), + GLib.SOURCE_REMOVE) + callback.assert_called() + mock_timeout_add.assert_called_with(seconds, emmental.alarm._do_alarm, + self.midnight, callback, 1) + self.assertEqual(emmental.alarm._GSOURCE_MAPPING[1], 42)