1
0

test_state_dbus.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181
  1. # systemctl-mqtt - MQTT client triggering & reporting shutdown on systemd-based systems
  2. #
  3. # Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
  4. #
  5. # This program is free software: you can redistribute it and/or modify
  6. # it under the terms of the GNU General Public License as published by
  7. # the Free Software Foundation, either version 3 of the License, or
  8. # any later version.
  9. #
  10. # This program is distributed in the hope that it will be useful,
  11. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. # GNU General Public License for more details.
  14. #
  15. # You should have received a copy of the GNU General Public License
  16. # along with this program. If not, see <https://www.gnu.org/licenses/>.
  17. import datetime
  18. import json
  19. import logging
  20. import unittest.mock
  21. import dbus.types
  22. import pytest
  23. import systemctl_mqtt
  24. # pylint: disable=protected-access
  25. def test_shutdown_lock():
  26. lock_fd = unittest.mock.MagicMock()
  27. with unittest.mock.patch("systemctl_mqtt._dbus.get_login_manager"):
  28. state = systemctl_mqtt._State(
  29. mqtt_topic_prefix="any",
  30. homeassistant_discovery_prefix=None,
  31. homeassistant_node_id=None,
  32. poweroff_delay=datetime.timedelta(),
  33. )
  34. state._login_manager.Inhibit.return_value = lock_fd
  35. state.acquire_shutdown_lock()
  36. state._login_manager.Inhibit.assert_called_once_with(
  37. "shutdown", "systemctl-mqtt", "Report shutdown via MQTT", "delay"
  38. )
  39. assert state._shutdown_lock == lock_fd
  40. # https://dbus.freedesktop.org/doc/dbus-python/dbus.types.html#dbus.types.UnixFd.take
  41. lock_fd.take.return_value = "fdnum"
  42. with unittest.mock.patch("os.close") as close_mock:
  43. state.release_shutdown_lock()
  44. close_mock.assert_called_once_with("fdnum")
  45. @pytest.mark.parametrize("active", [True, False])
  46. def test_prepare_for_shutdown_handler(caplog, active):
  47. with unittest.mock.patch("systemctl_mqtt._dbus.get_login_manager"):
  48. state = systemctl_mqtt._State(
  49. mqtt_topic_prefix="any",
  50. homeassistant_discovery_prefix=None,
  51. homeassistant_node_id=None,
  52. poweroff_delay=datetime.timedelta(),
  53. )
  54. mqtt_client_mock = unittest.mock.MagicMock()
  55. state.register_prepare_for_shutdown_handler(mqtt_client=mqtt_client_mock)
  56. # pylint: disable=no-member,comparison-with-callable
  57. connect_to_signal_kwargs = state._login_manager.connect_to_signal.call_args[1]
  58. assert connect_to_signal_kwargs["signal_name"] == "PrepareForShutdown"
  59. handler_function = connect_to_signal_kwargs["handler_function"]
  60. assert handler_function.func == state._prepare_for_shutdown_handler
  61. with unittest.mock.patch.object(
  62. state, "acquire_shutdown_lock"
  63. ) as acquire_lock_mock, unittest.mock.patch.object(
  64. state, "release_shutdown_lock"
  65. ) as release_lock_mock:
  66. handler_function(dbus.types.Boolean(active))
  67. if active:
  68. acquire_lock_mock.assert_not_called()
  69. release_lock_mock.assert_called_once_with()
  70. else:
  71. acquire_lock_mock.assert_called_once_with()
  72. release_lock_mock.assert_not_called()
  73. mqtt_client_mock.publish.assert_called_once_with(
  74. topic="any/preparing-for-shutdown",
  75. payload="true" if active else "false",
  76. retain=True,
  77. )
  78. assert len(caplog.records) == 1
  79. assert caplog.records[0].levelno == logging.ERROR
  80. assert caplog.records[0].message.startswith(
  81. "failed to publish on any/preparing-for-shutdown"
  82. )
  83. @pytest.mark.parametrize("active", [True, False])
  84. def test_publish_preparing_for_shutdown(active):
  85. login_manager_mock = unittest.mock.MagicMock()
  86. login_manager_mock.Get.return_value = dbus.Boolean(active)
  87. with unittest.mock.patch(
  88. "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock
  89. ):
  90. state = systemctl_mqtt._State(
  91. mqtt_topic_prefix="any",
  92. homeassistant_discovery_prefix=None,
  93. homeassistant_node_id=None,
  94. poweroff_delay=datetime.timedelta(),
  95. )
  96. assert state._login_manager == login_manager_mock
  97. mqtt_client_mock = unittest.mock.MagicMock()
  98. state.publish_preparing_for_shutdown(mqtt_client=mqtt_client_mock)
  99. login_manager_mock.Get.assert_called_once_with(
  100. "org.freedesktop.login1.Manager",
  101. "PreparingForShutdown",
  102. dbus_interface="org.freedesktop.DBus.Properties",
  103. )
  104. mqtt_client_mock.publish.assert_called_once_with(
  105. topic="any/preparing-for-shutdown",
  106. payload="true" if active else "false",
  107. retain=True,
  108. )
  109. def test_publish_preparing_for_shutdown_get_fail(caplog):
  110. login_manager_mock = unittest.mock.MagicMock()
  111. login_manager_mock.Get.side_effect = dbus.DBusException("mocked")
  112. with unittest.mock.patch(
  113. "systemctl_mqtt._dbus.get_login_manager", return_value=login_manager_mock
  114. ):
  115. state = systemctl_mqtt._State(
  116. mqtt_topic_prefix="any",
  117. homeassistant_discovery_prefix=None,
  118. homeassistant_node_id=None,
  119. poweroff_delay=datetime.timedelta(),
  120. )
  121. mqtt_client_mock = unittest.mock.MagicMock()
  122. state.publish_preparing_for_shutdown(mqtt_client=None)
  123. mqtt_client_mock.publish.assert_not_called()
  124. assert len(caplog.records) == 1
  125. assert caplog.records[0].levelno == logging.ERROR
  126. assert (
  127. caplog.records[0].message
  128. == "failed to read logind's PreparingForShutdown property: mocked"
  129. )
  130. @pytest.mark.parametrize("topic_prefix", ["systemctl/hostname", "hostname/systemctl"])
  131. @pytest.mark.parametrize("discovery_prefix", ["homeassistant", "home/assistant"])
  132. @pytest.mark.parametrize("node_id", ["node", "node-id"])
  133. @pytest.mark.parametrize("hostname", ["hostname", "host-name"])
  134. def test_publish_preparing_for_shutdown_homeassistant_config(
  135. topic_prefix, discovery_prefix, node_id, hostname
  136. ):
  137. state = systemctl_mqtt._State(
  138. mqtt_topic_prefix=topic_prefix,
  139. homeassistant_discovery_prefix=discovery_prefix,
  140. homeassistant_node_id=node_id,
  141. poweroff_delay=datetime.timedelta(),
  142. )
  143. mqtt_client = unittest.mock.MagicMock()
  144. with unittest.mock.patch(
  145. "systemctl_mqtt._utils.get_hostname", return_value=hostname
  146. ):
  147. state.publish_preparing_for_shutdown_homeassistant_config(
  148. mqtt_client=mqtt_client
  149. )
  150. assert mqtt_client.publish.call_count == 1
  151. publish_args, publish_kwargs = mqtt_client.publish.call_args
  152. assert not publish_args
  153. assert publish_kwargs["retain"]
  154. assert (
  155. publish_kwargs["topic"]
  156. == discovery_prefix
  157. + "/binary_sensor/"
  158. + node_id
  159. + "/preparing-for-shutdown/config"
  160. )
  161. assert json.loads(publish_kwargs["payload"]) == {
  162. "unique_id": "systemctl-mqtt/" + node_id + "/logind/preparing-for-shutdown",
  163. "state_topic": topic_prefix + "/preparing-for-shutdown",
  164. "payload_on": "true",
  165. "payload_off": "false",
  166. "name": node_id + " preparing for shutdown",
  167. }