1
0

test_state_dbus.py 3.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576
  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 logging
  18. import unittest.mock
  19. import dbus.types
  20. import pytest
  21. import systemctl_mqtt
  22. # pylint: disable=protected-access
  23. def test_shutdown_lock():
  24. lock_fd = unittest.mock.MagicMock()
  25. with unittest.mock.patch("systemctl_mqtt._get_login_manager"):
  26. state = systemctl_mqtt._State(mqtt_topic_prefix="any")
  27. state._login_manager.Inhibit.return_value = lock_fd
  28. state.acquire_shutdown_lock()
  29. state._login_manager.Inhibit.assert_called_once_with(
  30. "shutdown", "systemctl-mqtt", "Report shutdown via MQTT", "delay",
  31. )
  32. assert state._shutdown_lock == lock_fd
  33. # https://dbus.freedesktop.org/doc/dbus-python/dbus.types.html#dbus.types.UnixFd.take
  34. lock_fd.take.return_value = "fdnum"
  35. with unittest.mock.patch("os.close") as close_mock:
  36. state.release_shutdown_lock()
  37. close_mock.assert_called_once_with("fdnum")
  38. @pytest.mark.parametrize("active", [True, False])
  39. def test_prepare_for_shutdown_handler(caplog, active):
  40. with unittest.mock.patch("systemctl_mqtt._get_login_manager"):
  41. state = systemctl_mqtt._State(mqtt_topic_prefix="any")
  42. mqtt_client_mock = unittest.mock.MagicMock()
  43. state.register_prepare_for_shutdown_handler(mqtt_client=mqtt_client_mock)
  44. # pylint: disable=no-member,comparison-with-callable
  45. connect_to_signal_kwargs = state._login_manager.connect_to_signal.call_args[1]
  46. assert connect_to_signal_kwargs["signal_name"] == "PrepareForShutdown"
  47. handler_function = connect_to_signal_kwargs["handler_function"]
  48. assert handler_function.func == state._prepare_for_shutdown_handler
  49. with unittest.mock.patch.object(
  50. state, "acquire_shutdown_lock"
  51. ) as acquire_lock_mock, unittest.mock.patch.object(
  52. state, "release_shutdown_lock"
  53. ) as release_lock_mock:
  54. handler_function(dbus.types.Boolean(active))
  55. if active:
  56. acquire_lock_mock.assert_not_called()
  57. release_lock_mock.assert_called_once_with()
  58. else:
  59. acquire_lock_mock.assert_called_once_with()
  60. release_lock_mock.assert_not_called()
  61. mqtt_client_mock.publish.assert_called_once_with(
  62. topic="any/preparing-for-shutdown", payload="true" if active else "false",
  63. )
  64. assert len(caplog.records) == 1
  65. assert caplog.records[0].levelno == logging.ERROR
  66. assert caplog.records[0].message.startswith(
  67. "failed to publish on any/preparing-for-shutdown"
  68. )