test_cli.py 8.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280
  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 typing
  21. import unittest.mock
  22. import pytest
  23. import switchbot_mqtt
  24. import switchbot_mqtt._cli
  25. # pylint: disable=protected-access; tests
  26. # pylint: disable=too-many-arguments; these are tests, no API
  27. @pytest.mark.parametrize(
  28. (
  29. "argv",
  30. "expected_mqtt_host",
  31. "expected_mqtt_port",
  32. "expected_username",
  33. "expected_password",
  34. "expected_retry_count",
  35. ),
  36. [
  37. (
  38. ["", "--mqtt-host", "mqtt-broker.local"],
  39. "mqtt-broker.local",
  40. 1883,
  41. None,
  42. None,
  43. 3,
  44. ),
  45. (
  46. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8883"],
  47. "mqtt-broker.local",
  48. 8883,
  49. None,
  50. None,
  51. 3,
  52. ),
  53. (
  54. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-username", "me"],
  55. "mqtt-broker.local",
  56. 1883,
  57. "me",
  58. None,
  59. 3,
  60. ),
  61. (
  62. [
  63. "",
  64. "--mqtt-host",
  65. "mqtt-broker.local",
  66. "--mqtt-username",
  67. "me",
  68. "--mqtt-password",
  69. "secret",
  70. "--retries",
  71. "21",
  72. ],
  73. "mqtt-broker.local",
  74. 1883,
  75. "me",
  76. "secret",
  77. 21,
  78. ),
  79. ],
  80. )
  81. def test__main(
  82. argv,
  83. expected_mqtt_host,
  84. expected_mqtt_port,
  85. expected_username,
  86. expected_password,
  87. expected_retry_count,
  88. ):
  89. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  90. "sys.argv", argv
  91. ):
  92. switchbot_mqtt._cli._main()
  93. run_mock.assert_called_once_with(
  94. mqtt_host=expected_mqtt_host,
  95. mqtt_port=expected_mqtt_port,
  96. mqtt_username=expected_username,
  97. mqtt_password=expected_password,
  98. retry_count=expected_retry_count,
  99. device_passwords={},
  100. fetch_device_info=False,
  101. )
  102. @pytest.mark.parametrize(
  103. ("mqtt_password_file_content", "expected_password"),
  104. [
  105. ("secret", "secret"),
  106. ("secret space", "secret space"),
  107. ("secret ", "secret "),
  108. (" secret ", " secret "),
  109. ("secret\n", "secret"),
  110. ("secret\n\n", "secret\n"),
  111. ("secret\r\n", "secret"),
  112. ("secret\n\r\n", "secret\n"),
  113. ("你好\n", "你好"),
  114. ],
  115. )
  116. def test__main_mqtt_password_file(
  117. tmpdir, mqtt_password_file_content, expected_password
  118. ):
  119. mqtt_password_path = tmpdir.join("mqtt-password")
  120. with mqtt_password_path.open("w") as mqtt_password_file:
  121. mqtt_password_file.write(mqtt_password_file_content)
  122. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  123. "sys.argv",
  124. [
  125. "",
  126. "--mqtt-host",
  127. "localhost",
  128. "--mqtt-username",
  129. "me",
  130. "--mqtt-password-file",
  131. str(mqtt_password_path),
  132. ],
  133. ):
  134. switchbot_mqtt._cli._main()
  135. run_mock.assert_called_once_with(
  136. mqtt_host="localhost",
  137. mqtt_port=1883,
  138. mqtt_username="me",
  139. mqtt_password=expected_password,
  140. retry_count=3,
  141. device_passwords={},
  142. fetch_device_info=False,
  143. )
  144. def test__main_mqtt_password_file_collision(capsys):
  145. with unittest.mock.patch(
  146. "sys.argv",
  147. [
  148. "",
  149. "--mqtt-host",
  150. "localhost",
  151. "--mqtt-username",
  152. "me",
  153. "--mqtt-password",
  154. "secret",
  155. "--mqtt-password-file",
  156. "/var/lib/secrets/mqtt/password",
  157. ],
  158. ):
  159. with pytest.raises(SystemExit):
  160. switchbot_mqtt._cli._main()
  161. out, err = capsys.readouterr()
  162. assert not out
  163. assert (
  164. "argument --mqtt-password-file: not allowed with argument --mqtt-password\n"
  165. in err
  166. )
  167. @pytest.mark.parametrize(
  168. "device_passwords",
  169. [
  170. {},
  171. {"11:22:33:44:55:66": "password", "aa:bb:cc:dd:ee:ff": "secret"},
  172. ],
  173. )
  174. def test__main_device_password_file(tmpdir, device_passwords):
  175. device_passwords_path = tmpdir.join("passwords.json")
  176. device_passwords_path.write_text(json.dumps(device_passwords), encoding="utf8")
  177. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  178. "sys.argv",
  179. [
  180. "",
  181. "--mqtt-host",
  182. "localhost",
  183. "--device-password-file",
  184. str(device_passwords_path),
  185. ],
  186. ):
  187. switchbot_mqtt._cli._main()
  188. run_mock.assert_called_once_with(
  189. mqtt_host="localhost",
  190. mqtt_port=1883,
  191. mqtt_username=None,
  192. mqtt_password=None,
  193. retry_count=3,
  194. device_passwords=device_passwords,
  195. fetch_device_info=False,
  196. )
  197. def test__main_fetch_device_info():
  198. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  199. "sys.argv",
  200. [
  201. "",
  202. "--mqtt-host",
  203. "localhost",
  204. ],
  205. ):
  206. switchbot_mqtt._cli._main()
  207. default_kwargs = dict(
  208. mqtt_host="localhost",
  209. mqtt_port=1883,
  210. mqtt_username=None,
  211. mqtt_password=None,
  212. retry_count=3,
  213. device_passwords={},
  214. )
  215. run_mock.assert_called_once_with(fetch_device_info=False, **default_kwargs)
  216. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  217. "sys.argv",
  218. ["", "--mqtt-host", "localhost", "--fetch-device-info"],
  219. ):
  220. switchbot_mqtt._cli._main()
  221. run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
  222. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  223. "sys.argv",
  224. ["", "--mqtt-host", "localhost"],
  225. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": "21"}):
  226. switchbot_mqtt._cli._main()
  227. run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
  228. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  229. "sys.argv",
  230. ["", "--mqtt-host", "localhost"],
  231. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": ""}):
  232. switchbot_mqtt._cli._main()
  233. run_mock.assert_called_once_with(fetch_device_info=False, **default_kwargs)
  234. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  235. "sys.argv",
  236. ["", "--mqtt-host", "localhost"],
  237. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": " "}):
  238. switchbot_mqtt._cli._main()
  239. run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
  240. @pytest.mark.parametrize(
  241. ("additional_argv", "root_log_level", "log_format"),
  242. [
  243. ([], logging.INFO, "%(message)s"),
  244. (
  245. ["--debug"],
  246. logging.DEBUG,
  247. "%(asctime)s:%(levelname)s:%(name)s:%(funcName)s:%(message)s",
  248. ),
  249. ],
  250. )
  251. def test__main_log_config(
  252. additional_argv: typing.List[str], root_log_level: int, log_format: str
  253. ):
  254. with unittest.mock.patch(
  255. "sys.argv", ["", "--mqtt-host", "localhost"] + additional_argv
  256. ), unittest.mock.patch(
  257. "logging.basicConfig"
  258. ) as logging_basic_config_mock, unittest.mock.patch(
  259. "switchbot_mqtt._run"
  260. ):
  261. switchbot_mqtt._cli._main()
  262. logging_basic_config_mock.assert_called_once_with(
  263. level=root_log_level, format=log_format, datefmt="%Y-%m-%dT%H:%M:%S%z"
  264. )