test_cli.py 7.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276
  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 typing
  18. import unittest.mock
  19. import pytest
  20. import systemctl_mqtt
  21. import systemctl_mqtt._homeassistant
  22. import systemctl_mqtt._utils
  23. # pylint: disable=protected-access
  24. @pytest.mark.parametrize(
  25. (
  26. "argv",
  27. "expected_mqtt_host",
  28. "expected_mqtt_port",
  29. "expected_mqtt_disable_tls",
  30. "expected_username",
  31. "expected_password",
  32. "expected_topic_prefix",
  33. ),
  34. [
  35. (
  36. ["", "--mqtt-host", "mqtt-broker.local"],
  37. "mqtt-broker.local",
  38. 8883,
  39. False,
  40. None,
  41. None,
  42. None,
  43. ),
  44. (
  45. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-disable-tls"],
  46. "mqtt-broker.local",
  47. 1883,
  48. True,
  49. None,
  50. None,
  51. None,
  52. ),
  53. (
  54. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8883"],
  55. "mqtt-broker.local",
  56. 8883,
  57. False,
  58. None,
  59. None,
  60. None,
  61. ),
  62. (
  63. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8884"],
  64. "mqtt-broker.local",
  65. 8884,
  66. False,
  67. None,
  68. None,
  69. None,
  70. ),
  71. (
  72. [
  73. "",
  74. "--mqtt-host",
  75. "mqtt-broker.local",
  76. "--mqtt-port",
  77. "8884",
  78. "--mqtt-disable-tls",
  79. ],
  80. "mqtt-broker.local",
  81. 8884,
  82. True,
  83. None,
  84. None,
  85. None,
  86. ),
  87. (
  88. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-username", "me"],
  89. "mqtt-broker.local",
  90. 8883,
  91. False,
  92. "me",
  93. None,
  94. None,
  95. ),
  96. (
  97. [
  98. "",
  99. "--mqtt-host",
  100. "mqtt-broker.local",
  101. "--mqtt-username",
  102. "me",
  103. "--mqtt-password",
  104. "secret",
  105. ],
  106. "mqtt-broker.local",
  107. 8883,
  108. False,
  109. "me",
  110. "secret",
  111. None,
  112. ),
  113. (
  114. [
  115. "",
  116. "--mqtt-host",
  117. "mqtt-broker.local",
  118. "--mqtt-topic-prefix",
  119. "system/command",
  120. ],
  121. "mqtt-broker.local",
  122. 8883,
  123. False,
  124. None,
  125. None,
  126. "system/command",
  127. ),
  128. ],
  129. )
  130. def test__main(
  131. argv,
  132. expected_mqtt_host,
  133. expected_mqtt_port,
  134. expected_mqtt_disable_tls,
  135. expected_username,
  136. expected_password,
  137. expected_topic_prefix: typing.Optional[str],
  138. ):
  139. # pylint: disable=too-many-arguments
  140. with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
  141. "sys.argv", argv
  142. ), unittest.mock.patch(
  143. "systemctl_mqtt._utils.get_hostname", return_value="hostname"
  144. ):
  145. # pylint: disable=protected-access
  146. systemctl_mqtt._main()
  147. run_mock.assert_called_once_with(
  148. mqtt_host=expected_mqtt_host,
  149. mqtt_port=expected_mqtt_port,
  150. mqtt_disable_tls=expected_mqtt_disable_tls,
  151. mqtt_username=expected_username,
  152. mqtt_password=expected_password,
  153. mqtt_topic_prefix=expected_topic_prefix or "systemctl/hostname",
  154. homeassistant_discovery_prefix="homeassistant",
  155. homeassistant_node_id="hostname",
  156. )
  157. @pytest.mark.parametrize(
  158. ("password_file_content", "expected_password",),
  159. [
  160. ("secret", "secret"),
  161. ("secret space", "secret space"),
  162. ("secret ", "secret "),
  163. (" secret ", " secret "),
  164. ("secret\n", "secret"),
  165. ("secret\n\n", "secret\n"),
  166. ("secret\r\n", "secret"),
  167. ("secret\n\r\n", "secret\n"),
  168. ("你好\n", "你好"),
  169. ],
  170. )
  171. def test__main_password_file(tmpdir, password_file_content, expected_password):
  172. mqtt_password_path = tmpdir.join("mqtt-password")
  173. with mqtt_password_path.open("w") as mqtt_password_file:
  174. mqtt_password_file.write(password_file_content)
  175. with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
  176. "sys.argv",
  177. [
  178. "",
  179. "--mqtt-host",
  180. "localhost",
  181. "--mqtt-username",
  182. "me",
  183. "--mqtt-password-file",
  184. str(mqtt_password_path),
  185. ],
  186. ), unittest.mock.patch(
  187. "systemctl_mqtt._utils.get_hostname", return_value="hostname"
  188. ):
  189. # pylint: disable=protected-access
  190. systemctl_mqtt._main()
  191. run_mock.assert_called_once_with(
  192. mqtt_host="localhost",
  193. mqtt_port=8883,
  194. mqtt_disable_tls=False,
  195. mqtt_username="me",
  196. mqtt_password=expected_password,
  197. mqtt_topic_prefix="systemctl/hostname",
  198. homeassistant_discovery_prefix="homeassistant",
  199. homeassistant_node_id="hostname",
  200. )
  201. def test__main_password_file_collision(capsys):
  202. with unittest.mock.patch(
  203. "sys.argv",
  204. [
  205. "",
  206. "--mqtt-host",
  207. "localhost",
  208. "--mqtt-username",
  209. "me",
  210. "--mqtt-password",
  211. "secret",
  212. "--mqtt-password-file",
  213. "/var/lib/secrets/mqtt/password",
  214. ],
  215. ):
  216. with pytest.raises(SystemExit):
  217. # pylint: disable=protected-access
  218. systemctl_mqtt._main()
  219. out, err = capsys.readouterr()
  220. assert not out
  221. assert (
  222. "argument --mqtt-password-file: not allowed with argument --mqtt-password\n"
  223. in err
  224. )
  225. @pytest.mark.parametrize(
  226. ("args", "discovery_prefix"),
  227. [
  228. ([], "homeassistant"),
  229. (["--homeassistant-discovery-prefix", "home/assistant"], "home/assistant"),
  230. ],
  231. )
  232. def test__main_homeassistant_discovery_prefix(args, discovery_prefix):
  233. with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
  234. "sys.argv", ["", "--mqtt-host", "mqtt-broker.local"] + args
  235. ):
  236. systemctl_mqtt._main()
  237. assert run_mock.call_count == 1
  238. assert run_mock.call_args[1]["homeassistant_discovery_prefix"] == discovery_prefix
  239. @pytest.mark.parametrize(
  240. ("args", "node_id"),
  241. [([], "fallback"), (["--homeassistant-node-id", "raspberrypi"], "raspberrypi"),],
  242. )
  243. def test__main_homeassistant_node_id(args, node_id):
  244. with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
  245. "sys.argv", ["", "--mqtt-host", "mqtt-broker.local"] + args
  246. ), unittest.mock.patch(
  247. "systemctl_mqtt._utils.get_hostname", return_value="fallback",
  248. ):
  249. systemctl_mqtt._main()
  250. assert run_mock.call_count == 1
  251. assert run_mock.call_args[1]["homeassistant_node_id"] == node_id
  252. @pytest.mark.parametrize(
  253. "args", [["--homeassistant-node-id", "no pe"], ["--homeassistant-node-id", ""]],
  254. )
  255. def test__main_homeassistant_node_id_invalid(args):
  256. with unittest.mock.patch(
  257. "sys.argv", ["", "--mqtt-host", "mqtt-broker.local"] + args
  258. ):
  259. with pytest.raises(ValueError):
  260. systemctl_mqtt._main()