test_cli.py 8.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287
  1. # switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
  2. # compatible with home-assistant.io's MQTT Switch & Cover platform
  3. #
  4. # Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
  5. #
  6. # This program is free software: you can redistribute it and/or modify
  7. # it under the terms of the GNU General Public License as published by
  8. # the Free Software Foundation, either version 3 of the License, or
  9. # any later version.
  10. #
  11. # This program is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14. # GNU General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU General Public License
  17. # along with this program. If not, see <https://www.gnu.org/licenses/>.
  18. import json
  19. import logging
  20. import subprocess
  21. import typing
  22. import unittest.mock
  23. import pytest
  24. import switchbot_mqtt
  25. import switchbot_mqtt._cli
  26. # pylint: disable=protected-access; tests
  27. # pylint: disable=too-many-arguments; these are tests, no API
  28. def test_console_entry_point():
  29. assert subprocess.run(
  30. ["switchbot-mqtt", "--help"], stdout=subprocess.PIPE, check=True
  31. ).stdout.startswith(b"usage: ")
  32. @pytest.mark.parametrize(
  33. (
  34. "argv",
  35. "expected_mqtt_host",
  36. "expected_mqtt_port",
  37. "expected_username",
  38. "expected_password",
  39. "expected_retry_count",
  40. ),
  41. [
  42. (
  43. ["", "--mqtt-host", "mqtt-broker.local"],
  44. "mqtt-broker.local",
  45. 1883,
  46. None,
  47. None,
  48. 3,
  49. ),
  50. (
  51. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8883"],
  52. "mqtt-broker.local",
  53. 8883,
  54. None,
  55. None,
  56. 3,
  57. ),
  58. (
  59. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-username", "me"],
  60. "mqtt-broker.local",
  61. 1883,
  62. "me",
  63. None,
  64. 3,
  65. ),
  66. (
  67. [
  68. "",
  69. "--mqtt-host",
  70. "mqtt-broker.local",
  71. "--mqtt-username",
  72. "me",
  73. "--mqtt-password",
  74. "secret",
  75. "--retries",
  76. "21",
  77. ],
  78. "mqtt-broker.local",
  79. 1883,
  80. "me",
  81. "secret",
  82. 21,
  83. ),
  84. ],
  85. )
  86. def test__main(
  87. argv,
  88. expected_mqtt_host,
  89. expected_mqtt_port,
  90. expected_username,
  91. expected_password,
  92. expected_retry_count,
  93. ):
  94. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  95. "sys.argv", argv
  96. ):
  97. switchbot_mqtt._cli._main()
  98. run_mock.assert_called_once_with(
  99. mqtt_host=expected_mqtt_host,
  100. mqtt_port=expected_mqtt_port,
  101. mqtt_username=expected_username,
  102. mqtt_password=expected_password,
  103. retry_count=expected_retry_count,
  104. device_passwords={},
  105. fetch_device_info=False,
  106. )
  107. @pytest.mark.parametrize(
  108. ("mqtt_password_file_content", "expected_password"),
  109. [
  110. ("secret", "secret"),
  111. ("secret space", "secret space"),
  112. ("secret ", "secret "),
  113. (" secret ", " secret "),
  114. ("secret\n", "secret"),
  115. ("secret\n\n", "secret\n"),
  116. ("secret\r\n", "secret"),
  117. ("secret\n\r\n", "secret\n"),
  118. ("你好\n", "你好"),
  119. ],
  120. )
  121. def test__main_mqtt_password_file(
  122. tmpdir, mqtt_password_file_content, expected_password
  123. ):
  124. mqtt_password_path = tmpdir.join("mqtt-password")
  125. with mqtt_password_path.open("w") as mqtt_password_file:
  126. mqtt_password_file.write(mqtt_password_file_content)
  127. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  128. "sys.argv",
  129. [
  130. "",
  131. "--mqtt-host",
  132. "localhost",
  133. "--mqtt-username",
  134. "me",
  135. "--mqtt-password-file",
  136. str(mqtt_password_path),
  137. ],
  138. ):
  139. switchbot_mqtt._cli._main()
  140. run_mock.assert_called_once_with(
  141. mqtt_host="localhost",
  142. mqtt_port=1883,
  143. mqtt_username="me",
  144. mqtt_password=expected_password,
  145. retry_count=3,
  146. device_passwords={},
  147. fetch_device_info=False,
  148. )
  149. def test__main_mqtt_password_file_collision(capsys):
  150. with unittest.mock.patch(
  151. "sys.argv",
  152. [
  153. "",
  154. "--mqtt-host",
  155. "localhost",
  156. "--mqtt-username",
  157. "me",
  158. "--mqtt-password",
  159. "secret",
  160. "--mqtt-password-file",
  161. "/var/lib/secrets/mqtt/password",
  162. ],
  163. ):
  164. with pytest.raises(SystemExit):
  165. switchbot_mqtt._cli._main()
  166. out, err = capsys.readouterr()
  167. assert not out
  168. assert (
  169. "argument --mqtt-password-file: not allowed with argument --mqtt-password\n"
  170. in err
  171. )
  172. @pytest.mark.parametrize(
  173. "device_passwords",
  174. [
  175. {},
  176. {"11:22:33:44:55:66": "password", "aa:bb:cc:dd:ee:ff": "secret"},
  177. ],
  178. )
  179. def test__main_device_password_file(tmpdir, device_passwords):
  180. device_passwords_path = tmpdir.join("passwords.json")
  181. device_passwords_path.write_text(json.dumps(device_passwords), encoding="utf8")
  182. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  183. "sys.argv",
  184. [
  185. "",
  186. "--mqtt-host",
  187. "localhost",
  188. "--device-password-file",
  189. str(device_passwords_path),
  190. ],
  191. ):
  192. switchbot_mqtt._cli._main()
  193. run_mock.assert_called_once_with(
  194. mqtt_host="localhost",
  195. mqtt_port=1883,
  196. mqtt_username=None,
  197. mqtt_password=None,
  198. retry_count=3,
  199. device_passwords=device_passwords,
  200. fetch_device_info=False,
  201. )
  202. def test__main_fetch_device_info():
  203. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  204. "sys.argv",
  205. [
  206. "",
  207. "--mqtt-host",
  208. "localhost",
  209. ],
  210. ):
  211. switchbot_mqtt._cli._main()
  212. default_kwargs = dict(
  213. mqtt_host="localhost",
  214. mqtt_port=1883,
  215. mqtt_username=None,
  216. mqtt_password=None,
  217. retry_count=3,
  218. device_passwords={},
  219. )
  220. run_mock.assert_called_once_with(fetch_device_info=False, **default_kwargs)
  221. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  222. "sys.argv",
  223. ["", "--mqtt-host", "localhost", "--fetch-device-info"],
  224. ):
  225. switchbot_mqtt._cli._main()
  226. run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
  227. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  228. "sys.argv",
  229. ["", "--mqtt-host", "localhost"],
  230. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": "21"}):
  231. switchbot_mqtt._cli._main()
  232. run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
  233. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  234. "sys.argv",
  235. ["", "--mqtt-host", "localhost"],
  236. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": ""}):
  237. switchbot_mqtt._cli._main()
  238. run_mock.assert_called_once_with(fetch_device_info=False, **default_kwargs)
  239. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  240. "sys.argv",
  241. ["", "--mqtt-host", "localhost"],
  242. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": " "}):
  243. switchbot_mqtt._cli._main()
  244. run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
  245. @pytest.mark.parametrize(
  246. ("additional_argv", "root_log_level", "log_format"),
  247. [
  248. ([], logging.INFO, "%(message)s"),
  249. (
  250. ["--debug"],
  251. logging.DEBUG,
  252. "%(asctime)s:%(levelname)s:%(name)s:%(funcName)s:%(message)s",
  253. ),
  254. ],
  255. )
  256. def test__main_log_config(
  257. additional_argv: typing.List[str], root_log_level: int, log_format: str
  258. ):
  259. with unittest.mock.patch(
  260. "sys.argv", ["", "--mqtt-host", "localhost"] + additional_argv
  261. ), unittest.mock.patch(
  262. "logging.basicConfig"
  263. ) as logging_basic_config_mock, unittest.mock.patch(
  264. "switchbot_mqtt._run"
  265. ):
  266. switchbot_mqtt._cli._main()
  267. logging_basic_config_mock.assert_called_once_with(
  268. level=root_log_level, format=log_format, datefmt="%Y-%m-%dT%H:%M:%S%z"
  269. )