1
0

test_cli.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360
  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 pathlib
  21. import subprocess
  22. import typing
  23. import unittest.mock
  24. import _pytest.capture # pylint: disable=import-private-name; typing
  25. import pytest
  26. import switchbot_mqtt
  27. import switchbot_mqtt._cli
  28. # pylint: disable=protected-access; tests
  29. # pylint: disable=too-many-arguments; these are tests, no API
  30. def test_console_entry_point() -> None:
  31. assert subprocess.run(
  32. ["switchbot-mqtt", "--help"], stdout=subprocess.PIPE, check=True
  33. ).stdout.startswith(b"usage: ")
  34. @pytest.mark.parametrize(
  35. (
  36. "argv",
  37. "expected_mqtt_host",
  38. "expected_mqtt_port",
  39. "expected_username",
  40. "expected_password",
  41. "expected_retry_count",
  42. ),
  43. [
  44. (
  45. ["", "--mqtt-host", "mqtt-broker.local"],
  46. "mqtt-broker.local",
  47. 8883,
  48. None,
  49. None,
  50. 3,
  51. ),
  52. (
  53. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8884"],
  54. "mqtt-broker.local",
  55. 8884,
  56. None,
  57. None,
  58. 3,
  59. ),
  60. (
  61. ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-username", "me"],
  62. "mqtt-broker.local",
  63. 8883,
  64. "me",
  65. None,
  66. 3,
  67. ),
  68. (
  69. [
  70. "",
  71. "--mqtt-host",
  72. "mqtt-broker.local",
  73. "--mqtt-username",
  74. "me",
  75. "--mqtt-password",
  76. "secret",
  77. "--retries",
  78. "21",
  79. ],
  80. "mqtt-broker.local",
  81. 8883,
  82. "me",
  83. "secret",
  84. 21,
  85. ),
  86. ],
  87. )
  88. def test__main(
  89. argv: typing.List[str],
  90. expected_mqtt_host: str,
  91. expected_mqtt_port: int,
  92. expected_username: str,
  93. expected_password: str,
  94. expected_retry_count: int,
  95. ) -> None:
  96. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  97. "sys.argv", argv
  98. ):
  99. switchbot_mqtt._cli._main()
  100. run_mock.assert_called_once_with(
  101. mqtt_host=expected_mqtt_host,
  102. mqtt_port=expected_mqtt_port,
  103. mqtt_disable_tls=False,
  104. mqtt_username=expected_username,
  105. mqtt_password=expected_password,
  106. mqtt_topic_prefix="homeassistant/",
  107. retry_count=expected_retry_count,
  108. device_passwords={},
  109. fetch_device_info=False,
  110. )
  111. @pytest.mark.parametrize(
  112. ("mqtt_password_file_content", "expected_password"),
  113. [
  114. ("secret", "secret"),
  115. ("secret space", "secret space"),
  116. ("secret ", "secret "),
  117. (" secret ", " secret "),
  118. ("secret\n", "secret"),
  119. ("secret\n\n", "secret\n"),
  120. ("secret\r\n", "secret"),
  121. ("secret\n\r\n", "secret\n"),
  122. ("你好\n", "你好"),
  123. ],
  124. )
  125. def test__main_mqtt_password_file(
  126. tmp_path: pathlib.Path, mqtt_password_file_content: str, expected_password: str
  127. ) -> None:
  128. mqtt_password_path = tmp_path.joinpath("mqtt-password")
  129. mqtt_password_path.write_text(mqtt_password_file_content, encoding="utf8")
  130. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  131. "sys.argv",
  132. [
  133. "",
  134. "--mqtt-host",
  135. "localhost",
  136. "--mqtt-username",
  137. "me",
  138. "--mqtt-password-file",
  139. str(mqtt_password_path),
  140. ],
  141. ):
  142. switchbot_mqtt._cli._main()
  143. run_mock.assert_called_once_with(
  144. mqtt_host="localhost",
  145. mqtt_port=8883,
  146. mqtt_disable_tls=False,
  147. mqtt_username="me",
  148. mqtt_password=expected_password,
  149. mqtt_topic_prefix="homeassistant/",
  150. retry_count=3,
  151. device_passwords={},
  152. fetch_device_info=False,
  153. )
  154. def test__main_mqtt_password_file_collision(
  155. capsys: _pytest.capture.CaptureFixture,
  156. ) -> None:
  157. with unittest.mock.patch(
  158. "sys.argv",
  159. [
  160. "",
  161. "--mqtt-host",
  162. "localhost",
  163. "--mqtt-username",
  164. "me",
  165. "--mqtt-password",
  166. "secret",
  167. "--mqtt-password-file",
  168. "/var/lib/secrets/mqtt/password",
  169. ],
  170. ):
  171. with pytest.raises(SystemExit):
  172. switchbot_mqtt._cli._main()
  173. out, err = capsys.readouterr()
  174. assert not out
  175. assert (
  176. "argument --mqtt-password-file: not allowed with argument --mqtt-password\n"
  177. in err
  178. )
  179. @pytest.mark.parametrize(
  180. "device_passwords",
  181. [
  182. {},
  183. {"11:22:33:44:55:66": "password", "aa:bb:cc:dd:ee:ff": "secret"},
  184. ],
  185. )
  186. def test__main_device_password_file(
  187. tmp_path: pathlib.Path, device_passwords: typing.Dict[str, str]
  188. ) -> None:
  189. device_passwords_path = tmp_path.joinpath("passwords.json")
  190. device_passwords_path.write_text(json.dumps(device_passwords), encoding="utf8")
  191. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  192. "sys.argv",
  193. [
  194. "",
  195. "--mqtt-host",
  196. "localhost",
  197. "--device-password-file",
  198. str(device_passwords_path),
  199. ],
  200. ):
  201. switchbot_mqtt._cli._main()
  202. run_mock.assert_called_once_with(
  203. mqtt_host="localhost",
  204. mqtt_port=8883,
  205. mqtt_disable_tls=False,
  206. mqtt_username=None,
  207. mqtt_password=None,
  208. mqtt_topic_prefix="homeassistant/",
  209. retry_count=3,
  210. device_passwords=device_passwords,
  211. fetch_device_info=False,
  212. )
  213. _RUN_DEFAULT_KWARGS: typing.Dict[str, typing.Any] = {
  214. "mqtt_host": "localhost",
  215. "mqtt_port": 8883,
  216. "mqtt_disable_tls": False,
  217. "mqtt_username": None,
  218. "mqtt_password": None,
  219. "mqtt_topic_prefix": "homeassistant/",
  220. "retry_count": 3,
  221. "device_passwords": {},
  222. "fetch_device_info": False,
  223. }
  224. def test__main_mqtt_disable_tls() -> None:
  225. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  226. "sys.argv", ["", "--mqtt-host", "mqtt.local", "--mqtt-disable-tls"]
  227. ):
  228. switchbot_mqtt._cli._main()
  229. run_mock.assert_called_once_with(
  230. **{
  231. **_RUN_DEFAULT_KWARGS,
  232. "mqtt_host": "mqtt.local",
  233. "mqtt_disable_tls": True,
  234. "mqtt_port": 1883,
  235. }
  236. )
  237. def test__main_mqtt_disable_tls_overwrite_port() -> None:
  238. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  239. "sys.argv",
  240. ["", "--mqtt-host", "mqtt.local", "--mqtt-port", "1884", "--mqtt-disable-tls"],
  241. ):
  242. switchbot_mqtt._cli._main()
  243. run_mock.assert_called_once_with(
  244. **{
  245. **_RUN_DEFAULT_KWARGS,
  246. "mqtt_host": "mqtt.local",
  247. "mqtt_disable_tls": True,
  248. "mqtt_port": 1884,
  249. }
  250. )
  251. @pytest.mark.parametrize(
  252. ("additional_argv", "expected_topic_prefix"),
  253. [([], "homeassistant/"), (["--mqtt-topic-prefix", ""], "")],
  254. )
  255. def test__main_mqtt_topic_prefix(
  256. additional_argv: typing.List[str], expected_topic_prefix: str
  257. ) -> None:
  258. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  259. "sys.argv", ["", "--mqtt-host", "localhost"] + additional_argv
  260. ):
  261. switchbot_mqtt._cli._main()
  262. run_mock.assert_called_once_with(
  263. **{**_RUN_DEFAULT_KWARGS, "mqtt_topic_prefix": expected_topic_prefix}
  264. )
  265. def test__main_fetch_device_info() -> None:
  266. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  267. "sys.argv",
  268. [
  269. "",
  270. "--mqtt-host",
  271. "localhost",
  272. ],
  273. ):
  274. switchbot_mqtt._cli._main()
  275. run_mock.assert_called_once_with(
  276. **{**_RUN_DEFAULT_KWARGS, "fetch_device_info": False}
  277. )
  278. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  279. "sys.argv",
  280. ["", "--mqtt-host", "localhost", "--fetch-device-info"],
  281. ):
  282. switchbot_mqtt._cli._main()
  283. run_mock.assert_called_once_with(
  284. **{**_RUN_DEFAULT_KWARGS, "fetch_device_info": True}
  285. )
  286. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  287. "sys.argv",
  288. ["", "--mqtt-host", "localhost"],
  289. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": "21"}):
  290. switchbot_mqtt._cli._main()
  291. run_mock.assert_called_once_with(
  292. **{**_RUN_DEFAULT_KWARGS, "fetch_device_info": True}
  293. )
  294. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  295. "sys.argv",
  296. ["", "--mqtt-host", "localhost"],
  297. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": ""}):
  298. switchbot_mqtt._cli._main()
  299. run_mock.assert_called_once_with(
  300. **{**_RUN_DEFAULT_KWARGS, "fetch_device_info": False}
  301. )
  302. with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
  303. "sys.argv",
  304. ["", "--mqtt-host", "localhost"],
  305. ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": " "}):
  306. switchbot_mqtt._cli._main()
  307. run_mock.assert_called_once_with(
  308. **{**_RUN_DEFAULT_KWARGS, "fetch_device_info": True}
  309. )
  310. @pytest.mark.parametrize(
  311. ("additional_argv", "root_log_level", "log_format"),
  312. [
  313. ([], logging.INFO, "%(message)s"),
  314. (
  315. ["--debug"],
  316. logging.DEBUG,
  317. "%(asctime)s:%(levelname)s:%(name)s:%(funcName)s:%(message)s",
  318. ),
  319. ],
  320. )
  321. def test__main_log_config(
  322. additional_argv: typing.List[str], root_log_level: int, log_format: str
  323. ) -> None:
  324. with unittest.mock.patch(
  325. "sys.argv", ["", "--mqtt-host", "localhost"] + additional_argv
  326. ), unittest.mock.patch(
  327. "logging.basicConfig"
  328. ) as logging_basic_config_mock, unittest.mock.patch(
  329. "switchbot_mqtt._run"
  330. ):
  331. switchbot_mqtt._cli._main()
  332. logging_basic_config_mock.assert_called_once_with(
  333. level=root_log_level, format=log_format, datefmt="%Y-%m-%dT%H:%M:%S%z"
  334. )