test_dbus.py 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. # systemctl-mqtt - MQTT client triggering 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 logging
  19. import unittest.mock
  20. import dbus
  21. import pytest
  22. import systemctl_mqtt
  23. _UTC = datetime.timezone(offset=datetime.timedelta(seconds=0))
  24. # pylint: disable=protected-access
  25. def test__get_login_manager():
  26. login_manager = systemctl_mqtt._get_login_manager()
  27. assert isinstance(login_manager, dbus.proxies.Interface)
  28. assert login_manager.dbus_interface == "org.freedesktop.login1.Manager"
  29. # https://freedesktop.org/wiki/Software/systemd/logind/
  30. assert isinstance(login_manager.CanPowerOff(), dbus.String)
  31. def test__log_shutdown_inhibitors_some(caplog):
  32. login_manager = unittest.mock.MagicMock()
  33. login_manager.ListInhibitors.return_value = dbus.Array(
  34. [
  35. dbus.Struct(
  36. (
  37. dbus.String("shutdown:sleep"),
  38. dbus.String("Developer"),
  39. dbus.String("Haven't pushed my commits yet"),
  40. dbus.String("delay"),
  41. dbus.UInt32(1000),
  42. dbus.UInt32(1234),
  43. ),
  44. signature=None,
  45. ),
  46. dbus.Struct(
  47. (
  48. dbus.String("shutdown"),
  49. dbus.String("Editor"),
  50. dbus.String(""),
  51. dbus.String("Unsafed files open"),
  52. dbus.UInt32(0),
  53. dbus.UInt32(42),
  54. ),
  55. signature=None,
  56. ),
  57. ],
  58. signature=dbus.Signature("(ssssuu)"),
  59. )
  60. with caplog.at_level(logging.DEBUG):
  61. systemctl_mqtt._log_shutdown_inhibitors(login_manager)
  62. assert len(caplog.records) == 2
  63. assert caplog.records[0].levelno == logging.DEBUG
  64. assert (
  65. caplog.records[0].message
  66. == "detected shutdown inhibitor Developer (pid=1234, uid=1000, mode=delay): "
  67. + "Haven't pushed my commits yet"
  68. )
  69. def test__log_shutdown_inhibitors_none(caplog):
  70. login_manager = unittest.mock.MagicMock()
  71. login_manager.ListInhibitors.return_value = dbus.Array([])
  72. with caplog.at_level(logging.DEBUG):
  73. systemctl_mqtt._log_shutdown_inhibitors(login_manager)
  74. assert len(caplog.records) == 1
  75. assert caplog.records[0].levelno == logging.DEBUG
  76. assert caplog.records[0].message == "no shutdown inhibitor locks found"
  77. def test__log_shutdown_inhibitors_fail(caplog):
  78. login_manager = unittest.mock.MagicMock()
  79. login_manager.ListInhibitors.side_effect = dbus.DBusException("mocked")
  80. with caplog.at_level(logging.DEBUG):
  81. systemctl_mqtt._log_shutdown_inhibitors(login_manager)
  82. assert len(caplog.records) == 1
  83. assert caplog.records[0].levelno == logging.WARNING
  84. assert caplog.records[0].message == "failed to fetch shutdown inhibitors: mocked"
  85. @pytest.mark.parametrize("action", ["poweroff", "reboot"])
  86. def test__schedule_shutdown(action):
  87. login_manager_mock = unittest.mock.MagicMock()
  88. with unittest.mock.patch(
  89. "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
  90. ):
  91. systemctl_mqtt._schedule_shutdown(action=action)
  92. assert login_manager_mock.ScheduleShutdown.call_count == 1
  93. schedule_args, schedule_kwargs = login_manager_mock.ScheduleShutdown.call_args
  94. assert len(schedule_args) == 2
  95. assert schedule_args[0] == action
  96. assert isinstance(schedule_args[1], dbus.UInt64)
  97. shutdown_datetime = datetime.datetime.fromtimestamp(
  98. schedule_args[1] / 10 ** 6, tz=_UTC,
  99. )
  100. delay = shutdown_datetime - datetime.datetime.now(tz=_UTC)
  101. assert delay.total_seconds() == pytest.approx(
  102. systemctl_mqtt._SHUTDOWN_DELAY.total_seconds(), abs=0.1,
  103. )
  104. assert not schedule_kwargs
  105. @pytest.mark.parametrize("action", ["poweroff"])
  106. @pytest.mark.parametrize(
  107. ("exception_message", "log_message"),
  108. [
  109. ("test message", "test message"),
  110. (
  111. "Interactive authentication required.",
  112. "unauthorized; missing polkit authorization rules?",
  113. ),
  114. ],
  115. )
  116. def test__schedule_shutdown_fail(caplog, action, exception_message, log_message):
  117. login_manager_mock = unittest.mock.MagicMock()
  118. login_manager_mock.ScheduleShutdown.side_effect = dbus.DBusException(
  119. exception_message
  120. )
  121. with unittest.mock.patch(
  122. "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
  123. ), caplog.at_level(logging.DEBUG):
  124. systemctl_mqtt._schedule_shutdown(action=action)
  125. assert login_manager_mock.ScheduleShutdown.call_count == 1
  126. assert len(caplog.records) == 3
  127. assert caplog.records[0].levelno == logging.INFO
  128. assert caplog.records[0].message.startswith("scheduling {} for ".format(action))
  129. assert caplog.records[1].levelno == logging.ERROR
  130. assert caplog.records[1].message == "failed to schedule {}: {}".format(
  131. action, log_message
  132. )
  133. assert "inhibitor" in caplog.records[2].message
  134. @pytest.mark.parametrize(
  135. ("topic_suffix", "expected_action_arg"), [("poweroff", "poweroff")]
  136. )
  137. def test_mqtt_topic_suffix_action_mapping(topic_suffix, expected_action_arg):
  138. mqtt_action = systemctl_mqtt._MQTT_TOPIC_SUFFIX_ACTION_MAPPING[topic_suffix]
  139. login_manager_mock = unittest.mock.MagicMock()
  140. with unittest.mock.patch(
  141. "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
  142. ):
  143. mqtt_action.action()
  144. assert login_manager_mock.ScheduleShutdown.call_count == 1
  145. schedule_args, schedule_kwargs = login_manager_mock.ScheduleShutdown.call_args
  146. assert len(schedule_args) == 2
  147. assert schedule_args[0] == expected_action_arg
  148. assert not schedule_kwargs