add command-line option `--monitor-system-unit [unit_name]` enabling reports of systemd system units' ActiveState

implements https://github.com/fphammerle/systemctl-mqtt/issues/56

CHANGELOG.md

@@ -10,11 +10,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   (https://github.com/fphammerle/systemctl-mqtt/issues/97)
 - birth & last will message on topic `systemctl/[hostname]/status`
   ("online" / "offline", https://github.com/fphammerle/systemctl-mqtt/issues/38)
+- command-line option `--monitor-system-unit [unit_name]` enables reports on
+  topic `systemctl/[hostname]/unit/system/[unit_name]/active-state`
+  (https://github.com/fphammerle/systemctl-mqtt/issues/56)
 - automatic discovery in home assistant:
   - availability status
   - entity `button.[hostname]_logind_lock_all_sessions`
   - entity `button.[hostname]_logind_poweroff`
   - entity `button.[hostname]_logind_suspend`
+  - entity `sensor.[hostname]_unit_system_[unit_name]_active_state`
+    for each command-line parameter `--monitor-system-unit [unit_name]`
 - command-line option `--log-level {debug,info,warning,error,critical}`
 - declare compatibility with `python3.11`, `python3.12` & `python3.13`
 

README.md

@@ -51,6 +51,15 @@ $ mosquitto_pub -h MQTT_BROKER -t systemctl/hostname/lock-all-sessions -n
 $ mosquitto_pub -h MQTT_BROKER -t systemctl/hostname/suspend -n
 ```
 
+### Monitor `ActiveState` of System Units
+
+```
+$ mosquitto_pub --monitor-system-unit foo.service \
+    --monitor-system-unit bar.service …
+```
+enables reports on topic
+`systemctl/[hostname]/unit/system/[unit_name]/active-state`.
+
 ## Home Assistant 🏡
 
 ### Automatic Discovery

systemctl_mqtt/__init__.py

@@ -36,6 +36,7 @@ import jeepney.bus_messages
 import jeepney.io.asyncio
 
 import systemctl_mqtt._dbus.login_manager
+import systemctl_mqtt._dbus.service_manager
 import systemctl_mqtt._homeassistant
 import systemctl_mqtt._mqtt
 
@@ -54,13 +55,15 @@ _LOGGER = logging.getLogger(__name__)
 
 
 class _State:
-    def __init__(
+    # pylint: disable=too-many-instance-attributes
+    def __init__(  # pylint: disable=too-many-arguments
         self,
         *,
         mqtt_topic_prefix: str,
         homeassistant_discovery_prefix: str,
         homeassistant_discovery_object_id: str,
         poweroff_delay: datetime.timedelta,
+        monitored_system_unit_names: typing.List[str],
     ) -> None:
         self._mqtt_topic_prefix = mqtt_topic_prefix
         self._homeassistant_discovery_prefix = homeassistant_discovery_prefix
@@ -71,6 +74,7 @@ class _State:
         self._shutdown_lock: typing.Optional[jeepney.fds.FileDescriptor] = None
         self._shutdown_lock_mutex = threading.Lock()
         self.poweroff_delay = poweroff_delay
+        self._monitored_system_unit_names = monitored_system_unit_names
 
     @property
     def mqtt_topic_prefix(self) -> str:
@@ -84,6 +88,13 @@ class _State:
         # https://github.com/fphammerle/switchbot-mqtt/blob/v3.3.1/switchbot_mqtt/__init__.py#L30
         return self._mqtt_topic_prefix + "/status"
 
+    def get_system_unit_active_state_mqtt_topic(self, *, unit_name: str) -> str:
+        return self._mqtt_topic_prefix + "/unit/system/" + unit_name + "/active-state"
+
+    @property
+    def monitored_system_unit_names(self) -> typing.List[str]:
+        return self._monitored_system_unit_names
+
     @property
     def shutdown_lock_acquired(self) -> bool:
         return self._shutdown_lock is not None
@@ -201,6 +212,16 @@ class _State:
                 "platform": "button",
                 "command_topic": self.mqtt_topic_prefix + "/" + mqtt_topic_suffix,
             }
+        for unit_name in self._monitored_system_unit_names:
+            config["components"]["unit/system/" + unit_name + "/active-state"] = {  # type: ignore
+                "unique_id": f"{unique_id_prefix}-unit-system-{unit_name}-active-state",
+                "object_id": f"{hostname}_unit_system_{unit_name}_active_state",
+                "name": f"{unit_name} active state",
+                "platform": "sensor",
+                "state_topic": self.get_system_unit_active_state_mqtt_topic(
+                    unit_name=unit_name
+                ),
+            }
         _LOGGER.debug("publishing home assistant config on %s", discovery_topic)
         await mqtt_client.publish(
             topic=discovery_topic, payload=json.dumps(config), retain=False
@@ -263,27 +284,124 @@ async def _mqtt_message_loop(*, state: _State, mqtt_client: aiomqtt.Client) -> N
             action_by_topic[message.topic.value].trigger(state=state)
 
 
+async def _dbus_signal_loop_preparing_for_shutdown(
+    *,
+    state: _State,
+    mqtt_client: aiomqtt.Client,
+    dbus_router: jeepney.io.asyncio.DBusRouter,
+    bus_proxy: jeepney.io.asyncio.Proxy,
+) -> None:
+    preparing_for_shutdown_match_rule = (
+        # pylint: disable=protected-access
+        systemctl_mqtt._dbus.login_manager.get_login_manager_signal_match_rule(
+            "PrepareForShutdown"
+        )
+    )
+    assert await bus_proxy.AddMatch(preparing_for_shutdown_match_rule) == ()
+    with dbus_router.filter(preparing_for_shutdown_match_rule) as queue:
+        while True:
+            message: jeepney.low_level.Message = await queue.get()
+            (preparing_for_shutdown,) = message.body
+            await state.preparing_for_shutdown_handler(
+                active=preparing_for_shutdown, mqtt_client=mqtt_client
+            )
+            queue.task_done()
+
+
+async def _get_unit_path(
+    *, service_manager: jeepney.io.asyncio.Proxy, unit_name: str
+) -> str:
+    (path,) = await service_manager.GetUnit(name=unit_name)
+    return path
+
+
+async def _dbus_signal_loop_unit(  # pylint: disable=too-many-arguments
+    *,
+    state: _State,
+    mqtt_client: aiomqtt.Client,
+    dbus_router: jeepney.io.asyncio.DBusRouter,
+    bus_proxy: jeepney.io.asyncio.Proxy,
+    unit_name: str,
+    unit_path: str,
+) -> None:
+    unit_proxy = jeepney.io.asyncio.Proxy(
+        # pylint: disable=protected-access
+        msggen=systemctl_mqtt._dbus.service_manager.Unit(object_path=unit_path),
+        router=dbus_router,
+    )
+    unit_properties_changed_match_rule = jeepney.MatchRule(
+        type="signal",
+        interface="org.freedesktop.DBus.Properties",
+        member="PropertiesChanged",
+        path=unit_path,
+    )
+    assert (await bus_proxy.AddMatch(unit_properties_changed_match_rule)) == ()
+    # > Table 1. Unit ACTIVE states …
+    # > active	Started, bound, plugged in, …
+    # > inactive	Stopped, unbound, unplugged, …
+    # > failed	… process returned error code on exit, crashed, an operation
+    # .         timed out, or after too many restarts).
+    # > activating	Changing from inactive to active.
+    # > deactivating	Changing from active to inactive.
+    # > maintenance	Unit is inactive and … maintenance … in progress.
+    # > reloading	Unit is active and it is reloading its configuration.
+    # > refreshing	Unit is active and a new mount is being activated in its
+    # .             namespace.
+    # https://web.archive.org/web/20250101121304/https://www.freedesktop.org/software/systemd/man/latest/org.freedesktop.systemd1.html
+    active_state_topic = state.get_system_unit_active_state_mqtt_topic(
+        unit_name=unit_name
+    )
+    ((_, last_active_state),) = await unit_proxy.Get(property_name="ActiveState")
+    await mqtt_client.publish(topic=active_state_topic, payload=last_active_state)
+    with dbus_router.filter(unit_properties_changed_match_rule) as queue:
+        while True:
+            await queue.get()
+            ((_, current_active_state),) = await unit_proxy.Get(
+                property_name="ActiveState"
+            )
+            if current_active_state != last_active_state:
+                await mqtt_client.publish(
+                    topic=active_state_topic, payload=current_active_state
+                )
+                last_active_state = current_active_state
+            queue.task_done()
+
+
 async def _dbus_signal_loop(*, state: _State, mqtt_client: aiomqtt.Client) -> None:
     async with jeepney.io.asyncio.open_dbus_router(bus="SYSTEM") as router:
         # router: jeepney.io.asyncio.DBusRouter
         bus_proxy = jeepney.io.asyncio.Proxy(
             msggen=jeepney.bus_messages.message_bus, router=router
         )
-        preparing_for_shutdown_match_rule = (
+        system_service_manager = jeepney.io.asyncio.Proxy(
             # pylint: disable=protected-access
-            systemctl_mqtt._dbus.login_manager.get_login_manager_signal_match_rule(
-                "PrepareForShutdown"
-            )
+            msggen=systemctl_mqtt._dbus.service_manager.ServiceManager(),
+            router=router,
         )
-        assert await bus_proxy.AddMatch(preparing_for_shutdown_match_rule) == ()
-        with router.filter(preparing_for_shutdown_match_rule) as queue:
-            while True:
-                message: jeepney.low_level.Message = await queue.get()
-                (preparing_for_shutdown,) = message.body
-                await state.preparing_for_shutdown_handler(
-                    active=preparing_for_shutdown, mqtt_client=mqtt_client
+        await asyncio.gather(
+            *[
+                _dbus_signal_loop_preparing_for_shutdown(
+                    state=state,
+                    mqtt_client=mqtt_client,
+                    dbus_router=router,
+                    bus_proxy=bus_proxy,
                 )
-                queue.task_done()
+            ]
+            + [
+                _dbus_signal_loop_unit(
+                    state=state,
+                    mqtt_client=mqtt_client,
+                    dbus_router=router,
+                    bus_proxy=bus_proxy,
+                    unit_name=unit_name,
+                    unit_path=await _get_unit_path(
+                        service_manager=system_service_manager, unit_name=unit_name
+                    ),
+                )
+                for unit_name in state.monitored_system_unit_names
+            ],
+            return_exceptions=False,
+        )
 
 
 async def _run(  # pylint: disable=too-many-arguments
@@ -296,6 +414,7 @@ async def _run(  # pylint: disable=too-many-arguments
     homeassistant_discovery_prefix: str,
     homeassistant_discovery_object_id: str,
     poweroff_delay: datetime.timedelta,
+    monitored_system_unit_names: typing.List[str],
     mqtt_disable_tls: bool = False,
 ) -> None:
     state = _State(
@@ -303,6 +422,7 @@ async def _run(  # pylint: disable=too-many-arguments
         homeassistant_discovery_prefix=homeassistant_discovery_prefix,
         homeassistant_discovery_object_id=homeassistant_discovery_object_id,
         poweroff_delay=poweroff_delay,
+        monitored_system_unit_names=monitored_system_unit_names,
     )
     _LOGGER.info(
         "connecting to MQTT broker %s:%d (TLS %s)",
@@ -409,6 +529,14 @@ def _main() -> None:
     argparser.add_argument(
         "--poweroff-delay-seconds", type=float, default=4.0, help="default: %(default)s"
     )
+    argparser.add_argument(
+        "--monitor-system-unit",
+        type=str,
+        metavar="UNIT_NAME",
+        dest="monitored_system_unit_names",
+        action="append",
+        help="e.g. --monitor-system-unit ssh.service --monitor-system-unit custom.service",
+    )
     args = argparser.parse_args()
     logging.root.setLevel(_ARGUMENT_LOG_LEVEL_MAPPING[args.log_level])
     if args.mqtt_port:
@@ -449,5 +577,6 @@ def _main() -> None:
             homeassistant_discovery_prefix=args.homeassistant_discovery_prefix,
             homeassistant_discovery_object_id=args.homeassistant_discovery_object_id,
             poweroff_delay=datetime.timedelta(seconds=args.poweroff_delay_seconds),
+            monitored_system_unit_names=args.monitored_system_unit_names or [],
         )
     )

systemctl_mqtt/_dbus/__init__.py

@@ -0,0 +1,44 @@
+# systemctl-mqtt - MQTT client triggering & reporting shutdown on systemd-based systems
+#
+# Copyright (C) 2024 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 abc
+
+import jeepney
+
+
+class Properties(jeepney.MessageGenerator):
+    """
+    https://dbus.freedesktop.org/doc/dbus-specification.html#standard-interfaces-properties
+    """
+
+    # pylint: disable=too-few-public-methods
+
+    interface = "org.freedesktop.DBus.Properties"  # overwritten
+
+    # pylint: disable=invalid-name
+
+    def Get(self, property_name: str) -> jeepney.low_level.Message:
+        return jeepney.new_method_call(
+            remote_obj=jeepney.DBusAddress(
+                object_path=self.object_path,
+                bus_name=self.bus_name,
+                interface="org.freedesktop.DBus.Properties",
+            ),
+            method="Get",
+            signature="ss",
+            body=(self.interface, property_name),
+        )

systemctl_mqtt/_dbus/login_manager.py

@@ -21,6 +21,8 @@ import logging
 import jeepney
 import jeepney.io.blocking
 
+import systemctl_mqtt._dbus
+
 _LOGGER = logging.getLogger(__name__)
 
 _LOGIN_MANAGER_OBJECT_PATH = "/org/freedesktop/login1"
@@ -36,7 +38,7 @@ def get_login_manager_signal_match_rule(member: str) -> jeepney.MatchRule:
     )
 
 
-class LoginManager(jeepney.MessageGenerator):
+class LoginManager(systemctl_mqtt._dbus.Properties):  # pylint: disable=protected-access
     """
     https://freedesktop.org/wiki/Software/systemd/logind/
 
@@ -88,18 +90,6 @@ class LoginManager(jeepney.MessageGenerator):
             body=(what, who, why, mode),
         )
 
-    def Get(self, property_name: str) -> jeepney.low_level.Message:
-        return jeepney.new_method_call(
-            remote_obj=jeepney.DBusAddress(
-                object_path=self.object_path,
-                bus_name=self.bus_name,
-                interface="org.freedesktop.DBus.Properties",
-            ),
-            method="Get",
-            signature="ss",
-            body=(self.interface, property_name),
-        )
-
 
 def get_login_manager_proxy() -> jeepney.io.blocking.Proxy:
     # https://jeepney.readthedocs.io/en/latest/integrate.html

systemctl_mqtt/_dbus/service_manager.py

@@ -0,0 +1,57 @@
+# systemctl-mqtt - MQTT client triggering & reporting shutdown on systemd-based systems
+#
+# Copyright (C) 2024 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 jeepney
+
+import systemctl_mqtt._dbus
+
+
+class ServiceManager(jeepney.MessageGenerator):
+    """
+    https://www.freedesktop.org/software/systemd/man/latest/org.freedesktop.systemd1.html
+    """
+
+    # pylint: disable=too-few-public-methods
+
+    interface = "org.freedesktop.systemd1.Manager"
+
+    def __init__(self):
+        super().__init__(
+            object_path="/org/freedesktop/systemd1", bus_name="org.freedesktop.systemd1"
+        )
+
+    # pylint: disable=invalid-name
+
+    def GetUnit(self, name: str) -> jeepney.low_level.Message:
+        return jeepney.new_method_call(
+            remote_obj=self, method="GetUnit", signature="s", body=(name,)
+        )
+
+
+class Unit(systemctl_mqtt._dbus.Properties):  # pylint: disable=protected-access
+    """
+    https://www.freedesktop.org/software/systemd/man/latest/org.freedesktop.systemd1.html#Unit%20Objects
+    """
+
+    # pylint: disable=too-few-public-methods
+
+    interface = "org.freedesktop.systemd1.Unit"
+
+    def __init__(self, *, object_path: str):
+        super().__init__(object_path=object_path, bus_name="org.freedesktop.systemd1")
+
+    # pylint: disable=invalid-name

tests/dbus/message-generators/test_service_manager.py

@@ -0,0 +1,57 @@
+# systemctl-mqtt - MQTT client triggering & reporting shutdown on systemd-based systems
+#
+# Copyright (C) 2024 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 unittest.mock
+
+import pytest
+import jeepney.io.asyncio
+import jeepney.low_level
+
+import systemctl_mqtt
+
+# pylint: disable=protected-access
+
+
+@pytest.mark.asyncio
+async def test__get_unit_path() -> None:
+    router_mock = unittest.mock.AsyncMock()
+    reply_mock = unittest.mock.MagicMock()
+    expected_path = "/org/freedesktop/systemd1/unit/ssh_2eservice"
+    reply_mock.body = (expected_path,)
+    router_mock.send_and_get_reply.return_value = reply_mock
+    service_manager = jeepney.io.asyncio.Proxy(
+        msggen=systemctl_mqtt._dbus.service_manager.ServiceManager(),
+        router=router_mock,
+    )
+    assert (
+        await systemctl_mqtt._get_unit_path(
+            service_manager=service_manager, unit_name="ssh.service"
+        )
+        == expected_path
+    )
+    router_mock.send_and_get_reply.assert_awaited_once()
+    (msg,), send_kwargs = router_mock.send_and_get_reply.await_args
+    assert isinstance(msg, jeepney.low_level.Message)
+    assert msg.header.fields == {
+        jeepney.low_level.HeaderFields.path: "/org/freedesktop/systemd1",
+        jeepney.low_level.HeaderFields.destination: "org.freedesktop.systemd1",
+        jeepney.low_level.HeaderFields.interface: "org.freedesktop.systemd1.Manager",
+        jeepney.low_level.HeaderFields.member: "GetUnit",
+        jeepney.low_level.HeaderFields.signature: "s",
+    }
+    assert msg.body == ("ssh.service",)
+    assert not send_kwargs

tests/test_action.py

@@ -24,6 +24,7 @@ def test_poweroff_trigger(delay):
                 homeassistant_discovery_prefix="homeassistant",
                 homeassistant_discovery_object_id="node",
                 poweroff_delay=delay,
+                monitored_system_unit_names=[],
             )
         )
     schedule_shutdown_mock.assert_called_once_with(action="poweroff", delay=delay)
@@ -45,6 +46,7 @@ def test_mqtt_topic_suffix_action_mapping_poweroff(topic_suffix, expected_action
                 homeassistant_discovery_prefix="homeassistant",
                 homeassistant_discovery_object_id="node",
                 poweroff_delay=datetime.timedelta(),
+                monitored_system_unit_names=[],
             )
         )
     login_manager_mock.ScheduleShutdown.assert_called_once()

tests/test_cli.py

@@ -182,6 +182,7 @@ def test__main(
         homeassistant_discovery_prefix="homeassistant",
         homeassistant_discovery_object_id="systemctl-mqtt-hostname",
         poweroff_delay=datetime.timedelta(seconds=4),
+        monitored_system_unit_names=[],
     )
 
 
@@ -229,6 +230,7 @@ def test__main_password_file(tmpdir, password_file_content, expected_password):
         homeassistant_discovery_prefix="homeassistant",
         homeassistant_discovery_object_id="systemctl-mqtt-hostname",
         poweroff_delay=datetime.timedelta(seconds=4),
+        monitored_system_unit_names=[],
     )
 
 

tests/test_dbus.py

@@ -181,24 +181,38 @@ def test_lock_all_sessions(caplog):
     assert caplog.records[0].message == "instruct all sessions to activate screen locks"
 
 
+async def _get_unit_path_mock(  # pylint: disable=unused-argument
+    *, service_manager: jeepney.io.asyncio.Proxy, unit_name: str
+) -> str:
+    return "/org/freedesktop/systemd1/unit/" + unit_name
+
+
 @pytest.mark.asyncio
-async def test__dbus_signal_loop():
+@pytest.mark.parametrize(
+    "monitored_system_unit_names", [[], ["foo.service", "bar.service"]]
+)
+async def test__dbus_signal_loop(monitored_system_unit_names: typing.List[str]) -> None:
     # pylint: disable=too-many-locals,too-many-arguments
     state_mock = unittest.mock.AsyncMock()
     with unittest.mock.patch(
         "jeepney.io.asyncio.open_dbus_router",
-    ) as open_dbus_router_mock:
+    ) as open_dbus_router_mock, unittest.mock.patch(
+        "systemctl_mqtt._get_unit_path", _get_unit_path_mock
+    ), unittest.mock.patch(
+        "systemctl_mqtt._dbus_signal_loop_unit"
+    ) as dbus_signal_loop_unit_mock:
         async with open_dbus_router_mock() as dbus_router_mock:
             pass
         add_match_reply = unittest.mock.Mock()
         add_match_reply.body = ()
         dbus_router_mock.send_and_get_reply.return_value = add_match_reply
-        msg_queue = asyncio.Queue()
+        msg_queue: asyncio.Queue[jeepney.low_level.Message] = asyncio.Queue()
         await msg_queue.put(jeepney.low_level.Message(header=None, body=(False,)))
         await msg_queue.put(jeepney.low_level.Message(header=None, body=(True,)))
         await msg_queue.put(jeepney.low_level.Message(header=None, body=(False,)))
         dbus_router_mock.filter = unittest.mock.MagicMock()
         dbus_router_mock.filter.return_value.__enter__.return_value = msg_queue
+        state_mock.monitored_system_unit_names = monitored_system_unit_names
         # asyncio.TaskGroup added in python3.11
         loop_task = asyncio.create_task(
             systemctl_mqtt._dbus_signal_loop(
@@ -230,3 +244,87 @@ async def test__dbus_signal_loop():
     assert [
         c[1]["active"] for c in state_mock.preparing_for_shutdown_handler.call_args_list
     ] == [False, True, False]
+    assert not any(args for args, _ in dbus_signal_loop_unit_mock.await_args_list)
+    dbus_signal_loop_unit_kwargs = [
+        kwargs for _, kwargs in dbus_signal_loop_unit_mock.await_args_list
+    ]
+    assert [(a["unit_name"], a["unit_path"]) for a in dbus_signal_loop_unit_kwargs] == [
+        (n, f"/org/freedesktop/systemd1/unit/{n}") for n in monitored_system_unit_names
+    ]
+
+
+def _mock_get_active_state_reply(state: str) -> unittest.mock.MagicMock:
+    reply_mock = unittest.mock.MagicMock()
+    reply_mock.body = (("s", state),)
+    return reply_mock
+
+
+@pytest.mark.asyncio
+async def test__dbus_signal_loop_unit() -> None:
+    state = systemctl_mqtt._State(
+        mqtt_topic_prefix="prefix",
+        homeassistant_discovery_prefix="unused",
+        homeassistant_discovery_object_id="unused",
+        poweroff_delay=datetime.timedelta(),
+        monitored_system_unit_names=[],
+    )
+    mqtt_client_mock = unittest.mock.AsyncMock()
+    dbus_router_mock = unittest.mock.AsyncMock()
+    bus_proxy_mock = unittest.mock.AsyncMock()
+    bus_proxy_mock.AddMatch.return_value = ()
+    get_active_state_reply_mock = unittest.mock.MagicMock()
+    get_active_state_reply_mock.body = (("s", "active"),)
+    states = [
+        "active",
+        "deactivating",
+        "inactive",
+        "inactive",
+        "activating",
+        "active",
+        "active",
+        "active",
+        "inactive",
+    ]
+    dbus_router_mock.send_and_get_reply.side_effect = [
+        _mock_get_active_state_reply(s) for s in states
+    ]
+    msg_queue: asyncio.Queue[jeepney.low_level.Message] = asyncio.Queue()
+    for _ in range(len(states) - 1):
+        await msg_queue.put(jeepney.low_level.Message(header=None, body=()))
+    dbus_router_mock.filter = unittest.mock.MagicMock()
+    dbus_router_mock.filter.return_value.__enter__.return_value = msg_queue
+    loop_task = asyncio.create_task(
+        systemctl_mqtt._dbus_signal_loop_unit(
+            state=state,
+            mqtt_client=mqtt_client_mock,
+            dbus_router=dbus_router_mock,
+            bus_proxy=bus_proxy_mock,
+            unit_name="foo.service",
+            unit_path="/org/freedesktop/systemd1/unit/whatever.service",
+        )
+    )
+
+    async def _abort_after_msg_queue():
+        await msg_queue.join()
+        loop_task.cancel()
+
+    with pytest.raises(asyncio.exceptions.CancelledError):
+        await asyncio.gather(*(loop_task, _abort_after_msg_queue()))
+    bus_proxy_mock.AddMatch.assert_awaited_once()
+    ((match_rule,), add_match_kwargs) = bus_proxy_mock.AddMatch.await_args
+    assert match_rule.header_fields["interface"] == "org.freedesktop.DBus.Properties"
+    assert match_rule.header_fields["member"] == "PropertiesChanged"
+    assert not add_match_kwargs
+    assert mqtt_client_mock.publish.await_args_list == [
+        unittest.mock.call(
+            topic="prefix/unit/system/foo.service/active-state", payload=s
+        )
+        for s in [  # consecutive duplicates filtered
+            "active",
+            "deactivating",
+            "inactive",
+            "activating",
+            "active",
+            "inactive",
+        ]
+    ]

tests/test_mqtt.py

@@ -66,6 +66,7 @@ async def test__run(
             homeassistant_discovery_prefix=homeassistant_discovery_prefix,
             homeassistant_discovery_object_id=homeassistant_discovery_object_id,
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=[],
         )
     assert caplog.records[0].levelno == logging.INFO
     assert caplog.records[0].message == (
@@ -169,6 +170,7 @@ async def test__run_tls(caplog, mqtt_host, mqtt_port, mqtt_disable_tls):
             homeassistant_discovery_prefix="homeassistant",
             homeassistant_discovery_object_id="host",
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=[],
         )
     mqtt_client_class_mock.assert_called_once()
     _, mqtt_client_init_kwargs = mqtt_client_class_mock.call_args
@@ -204,6 +206,7 @@ async def test__run_tls_default():
             homeassistant_discovery_prefix="homeassistant",
             homeassistant_discovery_object_id="host",
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=[],
         )
     mqtt_client_class_mock.assert_called_once()
     # enabled by default
@@ -236,6 +239,7 @@ async def test__run_authentication(
             homeassistant_discovery_prefix="discovery-prefix",
             homeassistant_discovery_object_id="node-id",
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=[],
         )
     mqtt_client_class_mock.assert_called_once()
     _, mqtt_client_init_kwargs = mqtt_client_class_mock.call_args
@@ -267,6 +271,7 @@ async def test__run_authentication_missing_username(
                 homeassistant_discovery_prefix="discovery-prefix",
                 homeassistant_discovery_object_id="node-id",
                 poweroff_delay=datetime.timedelta(),
+                monitored_system_unit_names=[],
             )
     dbus_signal_loop_mock.assert_not_called()
 
@@ -295,6 +300,7 @@ async def test__run_sigint(mqtt_topic_prefix: str):
                 homeassistant_discovery_prefix="homeassistant",
                 homeassistant_discovery_object_id="host",
                 poweroff_delay=datetime.timedelta(),
+                monitored_system_unit_names=[],
             )
     async with mqtt_client_class_mock() as mqtt_client_mock:
         pass
@@ -327,6 +333,7 @@ async def test__mqtt_message_loop_trigger_poweroff(
         homeassistant_discovery_prefix="homeassistant",
         homeassistant_discovery_object_id="whatever",
         poweroff_delay=datetime.timedelta(seconds=21),
+        monitored_system_unit_names=[],
     )
     mqtt_client_mock = unittest.mock.AsyncMock()
     mqtt_client_mock.messages.__aiter__.return_value = [
@@ -374,6 +381,7 @@ async def test__mqtt_message_loop_retained(
         homeassistant_discovery_prefix="homeassistant",
         homeassistant_discovery_object_id="whatever",
         poweroff_delay=datetime.timedelta(seconds=21),
+        monitored_system_unit_names=[],
     )
     mqtt_client_mock = unittest.mock.AsyncMock()
     mqtt_client_mock.messages.__aiter__.return_value = [
@@ -402,3 +410,21 @@ async def test__mqtt_message_loop_retained(
             "ignoring retained message on topic 'systemctl/host/poweroff'",
         ),
     ]
+
+
+@pytest.mark.parametrize("mqtt_topic_prefix", ["systemctl/host", "systemd/raspberrypi"])
+@pytest.mark.parametrize("unit_name", ["foo.service", "bar.service"])
+def test_state_get_system_unit_active_state_mqtt_topic(
+    mqtt_topic_prefix: str, unit_name: str
+) -> None:
+    state = systemctl_mqtt._State(
+        mqtt_topic_prefix=mqtt_topic_prefix,
+        homeassistant_discovery_prefix="homeassistant",
+        homeassistant_discovery_object_id="whatever",
+        poweroff_delay=datetime.timedelta(seconds=21),
+        monitored_system_unit_names=[],
+    )
+    assert (
+        state.get_system_unit_active_state_mqtt_topic(unit_name=unit_name)
+        == f"{mqtt_topic_prefix}/unit/system/{unit_name}/active-state"
+    )

tests/test_state_dbus.py

@@ -40,6 +40,7 @@ def test_shutdown_lock():
             homeassistant_discovery_prefix=None,
             homeassistant_discovery_object_id=None,
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=[],
         )
         get_login_manager_mock.return_value.Inhibit.return_value = (lock_fd,)
         state.acquire_shutdown_lock()
@@ -66,6 +67,7 @@ async def test_preparing_for_shutdown_handler(active: bool) -> None:
             homeassistant_discovery_prefix="pre/fix",
             homeassistant_discovery_object_id="obj",
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=[],
         )
     mqtt_client_mock = unittest.mock.MagicMock()
     with unittest.mock.patch.object(
@@ -101,6 +103,7 @@ async def test_publish_preparing_for_shutdown(active: bool) -> None:
             homeassistant_discovery_prefix="pre/fix",
             homeassistant_discovery_object_id="obj",
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=[],
         )
     assert state._login_manager == login_manager_mock
     mqtt_client_mock = unittest.mock.AsyncMock()
@@ -133,6 +136,7 @@ async def test_publish_preparing_for_shutdown_get_fail(caplog):
             homeassistant_discovery_prefix=None,
             homeassistant_discovery_object_id=None,
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=[],
         )
     mqtt_client_mock = unittest.mock.MagicMock()
     await state.publish_preparing_for_shutdown(mqtt_client=None)
@@ -150,16 +154,25 @@ async def test_publish_preparing_for_shutdown_get_fail(caplog):
 @pytest.mark.parametrize("discovery_prefix", ["homeassistant", "home/assistant"])
 @pytest.mark.parametrize("object_id", ["raspberrypi", "debian21"])
 @pytest.mark.parametrize("hostname", ["hostname", "host-name"])
+@pytest.mark.parametrize(
+    "monitored_system_unit_names", [[], ["foo.service", "bar.service"]]
+)
 async def test_publish_homeassistant_device_config(
-    topic_prefix, discovery_prefix, object_id, hostname
-):
+    topic_prefix: str,
+    discovery_prefix: str,
+    object_id: str,
+    hostname: str,
+    monitored_system_unit_names: typing.List[str],
+) -> None:
     with unittest.mock.patch("jeepney.io.blocking.open_dbus_connection"):
         state = systemctl_mqtt._State(
             mqtt_topic_prefix=topic_prefix,
             homeassistant_discovery_prefix=discovery_prefix,
             homeassistant_discovery_object_id=object_id,
             poweroff_delay=datetime.timedelta(),
+            monitored_system_unit_names=monitored_system_unit_names,
         )
+    assert state.monitored_system_unit_names == monitored_system_unit_names
     mqtt_client = unittest.mock.AsyncMock()
     with unittest.mock.patch(
         "systemctl_mqtt._utils.get_hostname", return_value=hostname
@@ -212,5 +225,15 @@ async def test_publish_homeassistant_device_config(
                 "platform": "button",
                 "command_topic": f"{topic_prefix}/suspend",
             },
+        }
+        | {
+            f"unit/system/{n}/active-state": {
+                "unique_id": f"systemctl-mqtt-{hostname}-unit-system-{n}-active-state",
+                "object_id": f"{hostname}_unit_system_{n}_active_state",
+                "name": f"{n} active state",
+                "platform": "sensor",
+                "state_topic": f"{topic_prefix}/unit/system/{n}/active-state",
+            }
+            for n in monitored_system_unit_names
         },
     }