refactor: add param 'delay' to _dbus.schedule_shutdown

systemctl_mqtt/__init__.py

@@ -225,6 +225,7 @@ _MQTT_TOPIC_SUFFIX_ACTION_MAPPING = {
             # pylint: disable=protected-access
             systemctl_mqtt._dbus.schedule_shutdown,
             action="poweroff",
+            delay=datetime.timedelta(seconds=4),
         ),
     ),
 }

systemctl_mqtt/_dbus.py

@@ -22,8 +22,6 @@ import dbus
 
 _LOGGER = logging.getLogger(__name__)
 
-_SHUTDOWN_DELAY = datetime.timedelta(seconds=4)
-
 
 def get_login_manager() -> dbus.proxies.Interface:
     # https://dbus.freedesktop.org/doc/dbus-python/tutorial.html
@@ -61,10 +59,10 @@ def _log_shutdown_inhibitors(login_manager: dbus.proxies.Interface) -> None:
         _LOGGER.debug("no shutdown inhibitor locks found")
 
 
-def schedule_shutdown(action: str) -> None:
+def schedule_shutdown(action: str, delay: datetime.timedelta) -> None:
     # https://github.com/systemd/systemd/blob/v237/src/systemctl/systemctl.c#L8553
     assert action in ["poweroff", "reboot"], action
-    shutdown_datetime = datetime.datetime.now() + _SHUTDOWN_DELAY
+    shutdown_datetime = datetime.datetime.now() + delay
     # datetime.datetime.isoformat(timespec=) not available in python3.5
     # https://github.com/python/cpython/blob/v3.5.9/Lib/datetime.py#L1552
     _LOGGER.info(

tests/test_dbus.py

@@ -98,24 +98,23 @@ def test__log_shutdown_inhibitors_fail(caplog):
 
 
 @pytest.mark.parametrize("action", ["poweroff", "reboot"])
-def test__schedule_shutdown(action):
+@pytest.mark.parametrize("delay", [datetime.timedelta(0), datetime.timedelta(hours=1)])
+def test__schedule_shutdown(action, delay):
     login_manager_mock = unittest.mock.MagicMock()
     with unittest.mock.patch(
         "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock,
     ):
-        systemctl_mqtt._dbus.schedule_shutdown(action=action)
+        systemctl_mqtt._dbus.schedule_shutdown(action=action, delay=delay)
     assert login_manager_mock.ScheduleShutdown.call_count == 1
     schedule_args, schedule_kwargs = login_manager_mock.ScheduleShutdown.call_args
     assert len(schedule_args) == 2
     assert schedule_args[0] == action
     assert isinstance(schedule_args[1], dbus.UInt64)
     shutdown_datetime = datetime.datetime.fromtimestamp(
-        schedule_args[1] / 10 ** 6, tz=_UTC,
-    )
-    delay = shutdown_datetime - datetime.datetime.now(tz=_UTC)
-    assert delay.total_seconds() == pytest.approx(
-        systemctl_mqtt._dbus._SHUTDOWN_DELAY.total_seconds(), abs=0.1,
+        schedule_args[1] / 10 ** 6, tz=_UTC
     )
+    actual_delay = shutdown_datetime - datetime.datetime.now(tz=_UTC)
+    assert actual_delay.total_seconds() == pytest.approx(delay.total_seconds(), abs=0.1)
     assert not schedule_kwargs
 
 
@@ -138,7 +137,9 @@ def test__schedule_shutdown_fail(caplog, action, exception_message, log_message)
     with unittest.mock.patch(
         "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock,
     ), caplog.at_level(logging.DEBUG):
-        systemctl_mqtt._dbus.schedule_shutdown(action=action)
+        systemctl_mqtt._dbus.schedule_shutdown(
+            action=action, delay=datetime.timedelta(seconds=21)
+        )
     assert login_manager_mock.ScheduleShutdown.call_count == 1
     assert len(caplog.records) == 3
     assert caplog.records[0].levelno == logging.INFO