test_cli.py 8.6 KB

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