1
0

test_dbus.py 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105
  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. @pytest.mark.parametrize("action", ["poweroff", "reboot"])
  32. def test__schedule_shutdown(action):
  33. login_manager_mock = unittest.mock.MagicMock()
  34. with unittest.mock.patch(
  35. "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
  36. ):
  37. systemctl_mqtt._schedule_shutdown(action=action)
  38. login_manager_mock.ScheduleShutdown.assert_called_once()
  39. schedule_args, schedule_kwargs = login_manager_mock.ScheduleShutdown.call_args
  40. assert len(schedule_args) == 2
  41. assert schedule_args[0] == action
  42. shutdown_datetime = datetime.datetime.fromtimestamp(
  43. schedule_args[1] / 10 ** 6, tz=_UTC,
  44. )
  45. delay = shutdown_datetime - datetime.datetime.now(tz=_UTC)
  46. assert delay.total_seconds() == pytest.approx(
  47. datetime.timedelta(seconds=4).total_seconds(), abs=0.1,
  48. )
  49. assert not schedule_kwargs
  50. @pytest.mark.parametrize("action", ["poweroff"])
  51. @pytest.mark.parametrize(
  52. ("exception_message", "log_message"),
  53. [
  54. ("test message", "test message"),
  55. (
  56. "Interactive authentication required.",
  57. "unauthorized; missing polkit authorization rules?",
  58. ),
  59. ],
  60. )
  61. def test__schedule_shutdown_fail(caplog, action, exception_message, log_message):
  62. login_manager_mock = unittest.mock.MagicMock()
  63. login_manager_mock.ScheduleShutdown.side_effect = dbus.DBusException(
  64. exception_message
  65. )
  66. with unittest.mock.patch(
  67. "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
  68. ), caplog.at_level(logging.DEBUG):
  69. systemctl_mqtt._schedule_shutdown(action=action)
  70. login_manager_mock.ScheduleShutdown.assert_called_once()
  71. assert len(caplog.records) == 2
  72. assert caplog.records[0].levelno == logging.INFO
  73. assert caplog.records[0].message.startswith("scheduling {} for ".format(action))
  74. assert caplog.records[1].levelno == logging.ERROR
  75. assert caplog.records[1].message == "failed to schedule {}: {}".format(
  76. action, log_message
  77. )
  78. @pytest.mark.parametrize(
  79. ("topic_suffix", "expected_action_arg"), [("poweroff", "poweroff")]
  80. )
  81. def test_mqtt_topic_suffix_action_mapping(topic_suffix, expected_action_arg):
  82. action = systemctl_mqtt._MQTT_TOPIC_SUFFIX_ACTION_MAPPING[topic_suffix]
  83. login_manager_mock = unittest.mock.MagicMock()
  84. with unittest.mock.patch(
  85. "systemctl_mqtt._get_login_manager", return_value=login_manager_mock
  86. ):
  87. action()
  88. login_manager_mock.ScheduleShutdown.assert_called_once()
  89. schedule_args, schedule_kwargs = login_manager_mock.ScheduleShutdown.call_args
  90. assert len(schedule_args) == 2
  91. assert schedule_args[0] == expected_action_arg
  92. assert not schedule_kwargs