migrate from dbus-python to pure-python jeepney

fixes #39

.github/workflows/python.yml

@@ -23,11 +23,6 @@ jobs:
       with:
         python-version: ${{ matrix.python-version }}
     - run: pip install --upgrade pipenv==2024.1.0
-    - run: sudo apt-get update
-    # TODO exclude dbus-python & PyGObject from pipenv install
-    - run: sudo apt-get install --yes --no-install-recommends
-        libdbus-1-dev
-        libgirepository1.0-dev
     - run: pipenv install --python "$PYTHON_VERSION" --deploy --dev
       env:
         PYTHON_VERSION: ${{ matrix.python-version }}
@@ -57,10 +52,6 @@ jobs:
     # > [...]/coverage/inorout.py:507: CoverageWarning:
     # . Module  was never imported. (module-not-imported)
     - run: pip install --upgrade pipenv==2023.6.18
-    - run: sudo apt-get update
-    - run: sudo apt-get install --yes --no-install-recommends
-        libdbus-1-dev
-        libgirepository1.0-dev
     # by default pipenv picks the latest version in PATH
     - run: pipenv install --python "$PYTHON_VERSION" --deploy --dev
       env:

CHANGELOG.md

@@ -16,7 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   to pure-python [jeepney](https://gitlab.com/takluyver/jeepney)
   (removes indirect dependency on libdbus, glib,
   [PyGObject](https://gitlab.gnome.org/GNOME/pygobject) and
-  [pycairo](https://github.com/pygobject/pycairo))
+  [pycairo](https://github.com/pygobject/pycairo),
+  fixes https://github.com/fphammerle/systemctl-mqtt/issues/39)
 - automatic discovery in home assistant:
   - replace component-based (topic:
     `<discovery_prefix>/binary_sensor/<node_id>/preparing-for-shutdown/config`)

systemctl_mqtt/__init__.py

@@ -78,14 +78,15 @@ def acquire_shutdown_lock(self) -> None:
                 why="Report shutdown via MQTT",
                 mode="delay",
             )
+            assert isinstance(
+                self._shutdown_lock, jeepney.fds.FileDescriptor
+            ), self._shutdown_lock
             _LOGGER.debug("acquired shutdown inhibitor lock")
 
     def release_shutdown_lock(self) -> None:
         with self._shutdown_lock_mutex:
             if self._shutdown_lock:
-                # TODO
-                # https://dbus.freedesktop.org/doc/dbus-python/dbus.types.html#dbus.types.UnixFd.take
-                os.close(self._shutdown_lock.take())
+                self._shutdown_lock.close()
                 _LOGGER.debug("released shutdown inhibitor lock")
                 self._shutdown_lock = None
 
@@ -276,7 +277,7 @@ def _mqtt_on_connect(
         )
 
 
-def _run(
+def _run(  # pylint: disable=too-many-arguments
     *,
     mqtt_host: str,
     mqtt_port: int,
@@ -288,13 +289,14 @@ def _run(
     poweroff_delay: datetime.timedelta,
     mqtt_disable_tls: bool = False,
 ) -> None:
+    # pylint: disable=too-many-locals; will be split up when switching to async mqtt
     dbus_connection = jeepney.io.blocking.open_dbus_connection(bus="SYSTEM")
     bus_proxy = jeepney.io.blocking.Proxy(
         msggen=jeepney.bus_messages.message_bus, connection=dbus_connection
     )
     preparing_for_shutdown_match_rule = (
         # pylint: disable=protected-access
-        systemctl_mqtt._dbus.get_login_manager_signal_match_rule("SessionNew")
+        systemctl_mqtt._dbus.get_login_manager_signal_match_rule("PrepareForShutdown")
     )
     assert bus_proxy.AddMatch(preparing_for_shutdown_match_rule) == ()
     state = _State(
@@ -327,9 +329,9 @@ def _run(
     try:
         with dbus_connection.filter(preparing_for_shutdown_match_rule) as queue:
             while True:
-                print(dbus_connection.recv_until_filtered(queue).body)
+                (preparing_for_sleep,) = dbus_connection.recv_until_filtered(queue).body
                 state.preparing_for_shutdown_handler(
-                    active=True, mqtt_client=mqtt_client  # TODO
+                    active=preparing_for_sleep, mqtt_client=mqtt_client
                 )
     finally:
         # blocks until loop_forever stops

systemctl_mqtt/_dbus.py

@@ -60,6 +60,9 @@ def ListInhibitors(self) -> jeepney.low_level.Message:
     def LockSessions(self) -> jeepney.low_level.Message:
         return jeepney.new_method_call(remote_obj=self, method="LockSessions")
 
+    def CanPowerOff(self) -> jeepney.low_level.Message:
+        return jeepney.new_method_call(remote_obj=self, method="CanPowerOff")
+
     def ScheduleShutdown(
         self, *, action: str, time: datetime.datetime
     ) -> jeepney.low_level.Message:

tests/dbus/message-generators/test_login_manager.py

@@ -0,0 +1,98 @@
+# systemctl-mqtt - MQTT client triggering & reporting shutdown on systemd-based systems
+#
+# Copyright (C) 2024 Fabian Peter Hammerle <[email protected]>
+#
+# 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 contextlib
+import datetime
+import typing
+import unittest.mock
+
+import pytest
+from jeepney.low_level import HeaderFields, Message
+
+import systemctl_mqtt._dbus
+
+# pylint: disable=protected-access
+
+
+@contextlib.contextmanager
+def mock_open_dbus_connection() -> typing.Iterator[unittest.mock.MagicMock]:
+    with unittest.mock.patch("jeepney.io.blocking.open_dbus_connection") as mock:
+        yield mock.return_value
+
+
+@pytest.mark.parametrize(
+    ("member", "signature", "kwargs", "body"),
+    [
+        ("ListInhibitors", None, {}, ()),
+        ("LockSessions", None, {}, ()),
+        ("CanPowerOff", None, {}, ()),
+        (
+            "ScheduleShutdown",
+            "st",
+            {
+                "action": "poweroff",
+                "time": datetime.datetime(
+                    1970, 1, 1, 0, 0, tzinfo=datetime.timezone.utc
+                ),
+            },
+            ("poweroff", 0),
+        ),
+        (
+            "Inhibit",
+            "ssss",
+            {"what": "poweroff", "who": "me", "why": "fixing bugs", "mode": "block"},
+            ("poweroff", "me", "fixing bugs", "block"),
+        ),
+    ],
+)
+def test_method(
+    member: str,
+    signature: typing.Optional[str],
+    kwargs: typing.Dict[str, typing.Any],
+    body: typing.Tuple[typing.Any],
+) -> None:
+    with mock_open_dbus_connection() as dbus_connection_mock:
+        proxy = systemctl_mqtt._dbus.get_login_manager_proxy()
+    getattr(proxy, member)(**kwargs)
+    dbus_connection_mock.send_and_get_reply.assert_called_once()
+    message: Message = dbus_connection_mock.send_and_get_reply.call_args[0][0]
+    if signature:
+        assert message.header.fields.pop(HeaderFields.signature) == signature
+    assert message.header.fields == {
+        HeaderFields.path: "/org/freedesktop/login1",
+        HeaderFields.destination: "org.freedesktop.login1",
+        HeaderFields.interface: "org.freedesktop.login1.Manager",
+        HeaderFields.member: member,
+    }
+    assert message.body == body
+
+
+@pytest.mark.parametrize("property_name", ["HandlePowerKey", "Docked"])
+def test_get(property_name: str) -> None:
+    with mock_open_dbus_connection() as dbus_connection_mock:
+        proxy = systemctl_mqtt._dbus.get_login_manager_proxy()
+    proxy.Get(property_name=property_name)
+    dbus_connection_mock.send_and_get_reply.assert_called_once()
+    message: Message = dbus_connection_mock.send_and_get_reply.call_args[0][0]
+    assert message.header.fields == {
+        HeaderFields.path: "/org/freedesktop/login1",
+        HeaderFields.destination: "org.freedesktop.login1",
+        HeaderFields.interface: "org.freedesktop.DBus.Properties",
+        HeaderFields.member: "Get",
+        HeaderFields.signature: "ss",
+    }
+    assert message.body == ("org.freedesktop.login1.Manager", property_name)

tests/test_action.py

@@ -14,6 +14,8 @@
 def test_poweroff_trigger(delay):
     action = systemctl_mqtt._MQTTActionSchedulePoweroff()
     with unittest.mock.patch(
+        "systemctl_mqtt._dbus.get_login_manager_proxy"
+    ), unittest.mock.patch(
         "systemctl_mqtt._dbus.schedule_shutdown"
     ) as schedule_shutdown_mock:
         action.trigger(
@@ -34,7 +36,7 @@ def test_mqtt_topic_suffix_action_mapping_poweroff(topic_suffix, expected_action
     mqtt_action = systemctl_mqtt._MQTT_TOPIC_SUFFIX_ACTION_MAPPING[topic_suffix]
     login_manager_mock = unittest.mock.MagicMock()
     with unittest.mock.patch(
-        "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock
+        "systemctl_mqtt._dbus.get_login_manager_proxy", return_value=login_manager_mock
     ):
         mqtt_action.trigger(
             state=systemctl_mqtt._State(
@@ -46,16 +48,19 @@ def test_mqtt_topic_suffix_action_mapping_poweroff(topic_suffix, expected_action
         )
     login_manager_mock.ScheduleShutdown.assert_called_once()
     schedule_args, schedule_kwargs = login_manager_mock.ScheduleShutdown.call_args
-    assert len(schedule_args) == 2
-    assert schedule_args[0] == expected_action_arg
+    assert not schedule_args
+    assert schedule_kwargs.pop("action") == expected_action_arg
+    assert abs(
+        datetime.datetime.now() - schedule_kwargs.pop("time")
+    ) < datetime.timedelta(seconds=2)
     assert not schedule_kwargs
 
 
 def test_mqtt_topic_suffix_action_mapping_lock():
     mqtt_action = systemctl_mqtt._MQTT_TOPIC_SUFFIX_ACTION_MAPPING["lock-all-sessions"]
     login_manager_mock = unittest.mock.MagicMock()
     with unittest.mock.patch(
-        "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock
+        "systemctl_mqtt._dbus.get_login_manager_proxy", return_value=login_manager_mock
     ):
         mqtt_action.trigger(state="dummy")
     login_manager_mock.LockSessions.assert_called_once_with()