refactor: move some constants & methods to new systemctl_mqtt/_dbus.py

systemctl_mqtt/__init__.py

@@ -28,94 +28,18 @@ import typing
 
 import dbus
 import dbus.mainloop.glib
-import dbus.types
 
 # black keeps inserting a blank line above
 # https://pygobject.readthedocs.io/en/latest/getting_started.html#ubuntu-logo-ubuntu-debian-logo-debian
 import gi.repository.GLib  # pylint-import-requirements: imports=PyGObject
 import paho.mqtt.client
 
+import systemctl_mqtt._dbus
 import systemctl_mqtt._homeassistant
 import systemctl_mqtt._mqtt
 
 _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
-    bus = dbus.SystemBus()
-    proxy = bus.get_object(
-        bus_name="org.freedesktop.login1", object_path="/org/freedesktop/login1"
-    )  # type: dbus.proxies.ProxyObject
-    # https://freedesktop.org/wiki/Software/systemd/logind/
-    return dbus.Interface(object=proxy, dbus_interface="org.freedesktop.login1.Manager")
-
-
-def _log_shutdown_inhibitors(login_manager: dbus.proxies.Interface) -> None:
-    if _LOGGER.getEffectiveLevel() > logging.DEBUG:
-        return
-    found_inhibitor = False
-    try:
-        # https://www.freedesktop.org/wiki/Software/systemd/inhibit/
-        for what, who, why, mode, uid, pid in login_manager.ListInhibitors():
-            if "shutdown" in what:
-                found_inhibitor = True
-                _LOGGER.debug(
-                    "detected shutdown inhibitor %s (pid=%u, uid=%u, mode=%s): %s",
-                    who,
-                    pid,
-                    uid,
-                    mode,
-                    why,
-                )
-    except dbus.DBusException as exc:
-        _LOGGER.warning(
-            "failed to fetch shutdown inhibitors: %s", exc.get_dbus_message()
-        )
-        return
-    if not found_inhibitor:
-        _LOGGER.debug("no shutdown inhibitor locks found")
-
-
-def _schedule_shutdown(action: str) -> 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
-    # datetime.datetime.isoformat(timespec=) not available in python3.5
-    # https://github.com/python/cpython/blob/v3.5.9/Lib/datetime.py#L1552
-    _LOGGER.info(
-        "scheduling %s for %s", action, shutdown_datetime.strftime("%Y-%m-%d %H:%M:%S"),
-    )
-    # https://dbus.freedesktop.org/doc/dbus-python/tutorial.html?highlight=signature#basic-types
-    shutdown_epoch_usec = dbus.UInt64(shutdown_datetime.timestamp() * 10 ** 6)
-    login_manager = _get_login_manager()
-    try:
-        # $ gdbus introspect --system --dest org.freedesktop.login1 \
-        #       --object-path /org/freedesktop/login1 | grep -A 1 ScheduleShutdown
-        # ScheduleShutdown(in  s arg_0,
-        #                  in  t arg_1);
-        # $ gdbus call --system --dest org.freedesktop.login1 \
-        #       --object-path /org/freedesktop/login1 \
-        #       --method org.freedesktop.login1.Manager.ScheduleShutdown \
-        #       poweroff "$(date --date=10min +%s)000000"
-        # $ dbus-send --type=method_call --print-reply --system --dest=org.freedesktop.login1 \
-        #       /org/freedesktop/login1 \
-        #       org.freedesktop.login1.Manager.ScheduleShutdown \
-        #       string:poweroff "uint64:$(date --date=10min +%s)000000"
-        login_manager.ScheduleShutdown(action, shutdown_epoch_usec)
-    except dbus.DBusException as exc:
-        exc_msg = exc.get_dbus_message()
-        if "authentication required" in exc_msg.lower():
-            _LOGGER.error(
-                "failed to schedule %s: unauthorized; missing polkit authorization rules?",
-                action,
-            )
-        else:
-            _LOGGER.error("failed to schedule %s: %s", action, exc_msg)
-    _log_shutdown_inhibitors(login_manager)
-
 
 class _State:
     def __init__(
@@ -127,7 +51,9 @@ class _State:
         self._mqtt_topic_prefix = mqtt_topic_prefix
         self._homeassistant_discovery_prefix = homeassistant_discovery_prefix
         self._homeassistant_node_id = homeassistant_node_id
-        self._login_manager = _get_login_manager()  # type: dbus.proxies.Interface
+        self._login_manager = (
+            systemctl_mqtt._dbus.get_login_manager()
+        )  # type: dbus.proxies.Interface
         self._shutdown_lock = None  # type: typing.Optional[dbus.types.UnixFd]
         self._shutdown_lock_mutex = threading.Lock()
 
@@ -287,7 +213,12 @@ class _MQTTAction:
 
 _MQTT_TOPIC_SUFFIX_ACTION_MAPPING = {
     "poweroff": _MQTTAction(
-        name="poweroff", action=functools.partial(_schedule_shutdown, action="poweroff")
+        name="poweroff",
+        action=functools.partial(
+            # pylint: disable=protected-access
+            systemctl_mqtt._dbus.schedule_shutdown,
+            action="poweroff",
+        ),
     ),
 }
 

systemctl_mqtt/_dbus.py

@@ -0,0 +1,99 @@
+# systemctl-mqtt - MQTT client triggering & reporting shutdown on systemd-based systems
+#
+# Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import datetime
+import logging
+
+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
+    bus = dbus.SystemBus()
+    proxy = bus.get_object(
+        bus_name="org.freedesktop.login1", object_path="/org/freedesktop/login1"
+    )  # type: dbus.proxies.ProxyObject
+    # https://freedesktop.org/wiki/Software/systemd/logind/
+    return dbus.Interface(object=proxy, dbus_interface="org.freedesktop.login1.Manager")
+
+
+def _log_shutdown_inhibitors(login_manager: dbus.proxies.Interface) -> None:
+    if _LOGGER.getEffectiveLevel() > logging.DEBUG:
+        return
+    found_inhibitor = False
+    try:
+        # https://www.freedesktop.org/wiki/Software/systemd/inhibit/
+        for what, who, why, mode, uid, pid in login_manager.ListInhibitors():
+            if "shutdown" in what:
+                found_inhibitor = True
+                _LOGGER.debug(
+                    "detected shutdown inhibitor %s (pid=%u, uid=%u, mode=%s): %s",
+                    who,
+                    pid,
+                    uid,
+                    mode,
+                    why,
+                )
+    except dbus.DBusException as exc:
+        _LOGGER.warning(
+            "failed to fetch shutdown inhibitors: %s", exc.get_dbus_message()
+        )
+        return
+    if not found_inhibitor:
+        _LOGGER.debug("no shutdown inhibitor locks found")
+
+
+def schedule_shutdown(action: str) -> 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
+    # datetime.datetime.isoformat(timespec=) not available in python3.5
+    # https://github.com/python/cpython/blob/v3.5.9/Lib/datetime.py#L1552
+    _LOGGER.info(
+        "scheduling %s for %s", action, shutdown_datetime.strftime("%Y-%m-%d %H:%M:%S"),
+    )
+    # https://dbus.freedesktop.org/doc/dbus-python/tutorial.html?highlight=signature#basic-types
+    shutdown_epoch_usec = dbus.UInt64(shutdown_datetime.timestamp() * 10 ** 6)
+    login_manager = get_login_manager()
+    try:
+        # $ gdbus introspect --system --dest org.freedesktop.login1 \
+        #       --object-path /org/freedesktop/login1 | grep -A 1 ScheduleShutdown
+        # ScheduleShutdown(in  s arg_0,
+        #                  in  t arg_1);
+        # $ gdbus call --system --dest org.freedesktop.login1 \
+        #       --object-path /org/freedesktop/login1 \
+        #       --method org.freedesktop.login1.Manager.ScheduleShutdown \
+        #       poweroff "$(date --date=10min +%s)000000"
+        # $ dbus-send --type=method_call --print-reply --system --dest=org.freedesktop.login1 \
+        #       /org/freedesktop/login1 \
+        #       org.freedesktop.login1.Manager.ScheduleShutdown \
+        #       string:poweroff "uint64:$(date --date=10min +%s)000000"
+        login_manager.ScheduleShutdown(action, shutdown_epoch_usec)
+    except dbus.DBusException as exc:
+        exc_msg = exc.get_dbus_message()
+        if "authentication required" in exc_msg.lower():
+            _LOGGER.error(
+                "failed to schedule %s: unauthorized; missing polkit authorization rules?",
+                action,
+            )
+        else:
+            _LOGGER.error("failed to schedule %s: %s", action, exc_msg)
+    _log_shutdown_inhibitors(login_manager)

tests/test_dbus.py

@@ -22,15 +22,15 @@ import unittest.mock
 import dbus
 import pytest
 
-import systemctl_mqtt
+import systemctl_mqtt._dbus
 
 _UTC = datetime.timezone(offset=datetime.timedelta(seconds=0))
 
 # pylint: disable=protected-access
 
 
-def test__get_login_manager():
-    login_manager = systemctl_mqtt._get_login_manager()
+def test_get_login_manager():
+    login_manager = systemctl_mqtt._dbus.get_login_manager()
     assert isinstance(login_manager, dbus.proxies.Interface)
     assert login_manager.dbus_interface == "org.freedesktop.login1.Manager"
     # https://freedesktop.org/wiki/Software/systemd/logind/
@@ -67,7 +67,7 @@ def test__log_shutdown_inhibitors_some(caplog):
         signature=dbus.Signature("(ssssuu)"),
     )
     with caplog.at_level(logging.DEBUG):
-        systemctl_mqtt._log_shutdown_inhibitors(login_manager)
+        systemctl_mqtt._dbus._log_shutdown_inhibitors(login_manager)
     assert len(caplog.records) == 2
     assert caplog.records[0].levelno == logging.DEBUG
     assert (
@@ -81,7 +81,7 @@ def test__log_shutdown_inhibitors_none(caplog):
     login_manager = unittest.mock.MagicMock()
     login_manager.ListInhibitors.return_value = dbus.Array([])
     with caplog.at_level(logging.DEBUG):
-        systemctl_mqtt._log_shutdown_inhibitors(login_manager)
+        systemctl_mqtt._dbus._log_shutdown_inhibitors(login_manager)
     assert len(caplog.records) == 1
     assert caplog.records[0].levelno == logging.DEBUG
     assert caplog.records[0].message == "no shutdown inhibitor locks found"
@@ -91,7 +91,7 @@ def test__log_shutdown_inhibitors_fail(caplog):
     login_manager = unittest.mock.MagicMock()
     login_manager.ListInhibitors.side_effect = dbus.DBusException("mocked")
     with caplog.at_level(logging.DEBUG):
-        systemctl_mqtt._log_shutdown_inhibitors(login_manager)
+        systemctl_mqtt._dbus._log_shutdown_inhibitors(login_manager)
     assert len(caplog.records) == 1
     assert caplog.records[0].levelno == logging.WARNING
     assert caplog.records[0].message == "failed to fetch shutdown inhibitors: mocked"
@@ -101,9 +101,9 @@ def test__log_shutdown_inhibitors_fail(caplog):
 def test__schedule_shutdown(action):
     login_manager_mock = unittest.mock.MagicMock()
     with unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
+        "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock,
     ):
-        systemctl_mqtt._schedule_shutdown(action=action)
+        systemctl_mqtt._dbus.schedule_shutdown(action=action)
     assert login_manager_mock.ScheduleShutdown.call_count == 1
     schedule_args, schedule_kwargs = login_manager_mock.ScheduleShutdown.call_args
     assert len(schedule_args) == 2
@@ -114,7 +114,7 @@ def test__schedule_shutdown(action):
     )
     delay = shutdown_datetime - datetime.datetime.now(tz=_UTC)
     assert delay.total_seconds() == pytest.approx(
-        systemctl_mqtt._SHUTDOWN_DELAY.total_seconds(), abs=0.1,
+        systemctl_mqtt._dbus._SHUTDOWN_DELAY.total_seconds(), abs=0.1,
     )
     assert not schedule_kwargs
 
@@ -136,9 +136,9 @@ def test__schedule_shutdown_fail(caplog, action, exception_message, log_message)
         exception_message
     )
     with unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
+        "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock,
     ), caplog.at_level(logging.DEBUG):
-        systemctl_mqtt._schedule_shutdown(action=action)
+        systemctl_mqtt._dbus.schedule_shutdown(action=action)
     assert login_manager_mock.ScheduleShutdown.call_count == 1
     assert len(caplog.records) == 3
     assert caplog.records[0].levelno == logging.INFO
@@ -157,7 +157,7 @@ def test_mqtt_topic_suffix_action_mapping(topic_suffix, expected_action_arg):
     mqtt_action = systemctl_mqtt._MQTT_TOPIC_SUFFIX_ACTION_MAPPING[topic_suffix]
     login_manager_mock = unittest.mock.MagicMock()
     with unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
+        "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock,
     ):
         mqtt_action.action()
     assert login_manager_mock.ScheduleShutdown.call_count == 1

tests/test_mqtt.py

@@ -54,7 +54,7 @@ def test__run(
     ) as mqtt_loop_forever_mock, unittest.mock.patch(
         "gi.repository.GLib.MainLoop.run"
     ) as glib_loop_mock, unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager"
+        "systemctl_mqtt._dbus.get_login_manager"
     ) as get_login_manager_mock:
         ssl_wrap_socket_mock.return_value.send = len
         get_login_manager_mock.return_value.Get.return_value = dbus.Boolean(False)
@@ -160,7 +160,7 @@ def test__run_authentication(
     ) as mqtt_loop_forever_mock, unittest.mock.patch(
         "gi.repository.GLib.MainLoop.run"
     ), unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager"
+        "systemctl_mqtt._dbus.get_login_manager"
     ):
         ssl_wrap_socket_mock.return_value.send = len
         systemctl_mqtt._run(
@@ -191,7 +191,7 @@ def _initialize_mqtt_client(
     ) as mqtt_loop_forever_mock, unittest.mock.patch(
         "gi.repository.GLib.MainLoop.run"
     ), unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager"
+        "systemctl_mqtt._dbus.get_login_manager"
     ) as get_login_manager_mock:
         ssl_wrap_socket_mock.return_value.send = len
         get_login_manager_mock.return_value.Get.return_value = dbus.Boolean(False)
@@ -241,7 +241,7 @@ def test__client_handle_message(caplog, mqtt_host, mqtt_port, mqtt_topic_prefix)
 @pytest.mark.parametrize("mqtt_password", ["secret"])
 def test__run_authentication_missing_username(mqtt_host, mqtt_port, mqtt_password):
     with unittest.mock.patch("paho.mqtt.client.Client"), unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager"
+        "systemctl_mqtt._dbus.get_login_manager"
     ):
         with pytest.raises(ValueError):
             systemctl_mqtt._run(

tests/test_state_dbus.py

@@ -29,7 +29,7 @@ import systemctl_mqtt
 
 def test_shutdown_lock():
     lock_fd = unittest.mock.MagicMock()
-    with unittest.mock.patch("systemctl_mqtt._get_login_manager"):
+    with unittest.mock.patch("systemctl_mqtt._dbus.get_login_manager"):
         state = systemctl_mqtt._State(
             mqtt_topic_prefix="any",
             homeassistant_discovery_prefix=None,
@@ -50,7 +50,7 @@ def test_shutdown_lock():
 
 @pytest.mark.parametrize("active", [True, False])
 def test_prepare_for_shutdown_handler(caplog, active):
-    with unittest.mock.patch("systemctl_mqtt._get_login_manager"):
+    with unittest.mock.patch("systemctl_mqtt._dbus.get_login_manager"):
         state = systemctl_mqtt._State(
             mqtt_topic_prefix="any",
             homeassistant_discovery_prefix=None,
@@ -92,7 +92,7 @@ def test_publish_preparing_for_shutdown(active):
     login_manager_mock = unittest.mock.MagicMock()
     login_manager_mock.Get.return_value = dbus.Boolean(active)
     with unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
+        "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock
     ):
         state = systemctl_mqtt._State(
             mqtt_topic_prefix="any",
@@ -118,7 +118,7 @@ def test_publish_preparing_for_shutdown_get_fail(caplog):
     login_manager_mock = unittest.mock.MagicMock()
     login_manager_mock.Get.side_effect = dbus.DBusException("mocked")
     with unittest.mock.patch(
-        "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
+        "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock
     ):
         state = systemctl_mqtt._State(
             mqtt_topic_prefix="any",