test_cli.py 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  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 typing
  18. import unittest.mock
  19. import pytest
  20. import systemctl_mqtt
  21. @pytest.mark.parametrize(
  22. (
  23. "argv",
  24. "expected_mqtt_host",
  25. "expected_mqtt_port",
  26. "expected_username",
  27. "expected_password",
  28. "expected_topic_prefix",
  29. ),
  30. [
  31. (
  32. ["", "--mqtt-host", "mqtt-broker.local"],
  33. "mqtt-broker.local",
  34. 8883,
  35. None,
  36. None,
  37. None,
  38. ),
  39. (
  40. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8883"],
  41. "mqtt-broker.local",
  42. 8883,
  43. None,
  44. None,
  45. None,
  46. ),
  47. (
  48. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-username", "me"],
  49. "mqtt-broker.local",
  50. 8883,
  51. "me",
  52. None,
  53. None,
  54. ),
  55. (
  56. [
  57. "",
  58. "--mqtt-host",
  59. "mqtt-broker.local",
  60. "--mqtt-username",
  61. "me",
  62. "--mqtt-password",
  63. "secret",
  64. ],
  65. "mqtt-broker.local",
  66. 8883,
  67. "me",
  68. "secret",
  69. None,
  70. ),
  71. (
  72. [
  73. "",
  74. "--mqtt-host",
  75. "mqtt-broker.local",
  76. "--mqtt-topic-prefix",
  77. "system/command",
  78. ],
  79. "mqtt-broker.local",
  80. 8883,
  81. None,
  82. None,
  83. "system/command",
  84. ),
  85. ],
  86. )
  87. def test__main(
  88. argv,
  89. expected_mqtt_host,
  90. expected_mqtt_port,
  91. expected_username,
  92. expected_password,
  93. expected_topic_prefix: typing.Optional[str],
  94. ):
  95. # pylint: disable=too-many-arguments
  96. with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
  97. "sys.argv", argv
  98. ), unittest.mock.patch("systemctl_mqtt._get_hostname", return_value="hostname"):
  99. # pylint: disable=protected-access
  100. systemctl_mqtt._main()
  101. run_mock.assert_called_once_with(
  102. mqtt_host=expected_mqtt_host,
  103. mqtt_port=expected_mqtt_port,
  104. mqtt_username=expected_username,
  105. mqtt_password=expected_password,
  106. mqtt_topic_prefix=expected_topic_prefix or "systemctl/hostname",
  107. )
  108. @pytest.mark.parametrize(
  109. ("password_file_content", "expected_password",),
  110. [
  111. ("secret", "secret"),
  112. ("secret space", "secret space"),
  113. ("secret ", "secret "),
  114. (" secret ", " secret "),
  115. ("secret\n", "secret"),
  116. ("secret\n\n", "secret\n"),
  117. ("secret\r\n", "secret"),
  118. ("secret\n\r\n", "secret\n"),
  119. ("你好\n", "你好"),
  120. ],
  121. )
  122. def test__main_password_file(tmpdir, password_file_content, expected_password):
  123. mqtt_password_path = tmpdir.join("mqtt-password")
  124. with mqtt_password_path.open("w") as mqtt_password_file:
  125. mqtt_password_file.write(password_file_content)
  126. with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
  127. "sys.argv",
  128. [
  129. "",
  130. "--mqtt-host",
  131. "localhost",
  132. "--mqtt-username",
  133. "me",
  134. "--mqtt-password-file",
  135. str(mqtt_password_path),
  136. ],
  137. ), unittest.mock.patch("systemctl_mqtt._get_hostname", return_value="hostname"):
  138. # pylint: disable=protected-access
  139. systemctl_mqtt._main()
  140. run_mock.assert_called_once_with(
  141. mqtt_host="localhost",
  142. mqtt_port=8883,
  143. mqtt_username="me",
  144. mqtt_password=expected_password,
  145. mqtt_topic_prefix="systemctl/hostname",
  146. )
  147. def test__main_password_file_collision(capsys):
  148. with unittest.mock.patch(
  149. "sys.argv",
  150. [
  151. "",
  152. "--mqtt-host",
  153. "localhost",
  154. "--mqtt-username",
  155. "me",
  156. "--mqtt-password",
  157. "secret",
  158. "--mqtt-password-file",
  159. "/var/lib/secrets/mqtt/password",
  160. ],
  161. ):
  162. with pytest.raises(SystemExit):
  163. # pylint: disable=protected-access
  164. systemctl_mqtt._main()
  165. out, err = capsys.readouterr()
  166. assert not out
  167. assert (
  168. "argument --mqtt-password-file: not allowed with argument --mqtt-password\n"
  169. in err
  170. )
  171. @pytest.mark.parametrize("hostname", ["test"])
  172. def test__get_hostname(hostname):
  173. with unittest.mock.patch("socket.gethostname", return_value=hostname):
  174. # pylint: disable=protected-access
  175. assert systemctl_mqtt._get_hostname() == hostname