test_mqtt.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418
  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 logging
  19. import typing
  20. import unittest.mock
  21. import pytest
  22. from paho.mqtt.client import MQTT_ERR_QUEUE_SIZE, MQTT_ERR_SUCCESS, MQTTMessage, Client
  23. import switchbot_mqtt
  24. # pylint: disable=protected-access
  25. # pylint: disable=too-many-arguments; these are tests, no API
  26. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  27. @pytest.mark.parametrize("mqtt_port", [1833])
  28. @pytest.mark.parametrize("retry_count", [3, 21])
  29. def test__run(caplog, mqtt_host, mqtt_port, retry_count):
  30. with unittest.mock.patch(
  31. "paho.mqtt.client.Client"
  32. ) as mqtt_client_mock, caplog.at_level(logging.DEBUG):
  33. switchbot_mqtt._run(
  34. mqtt_host=mqtt_host,
  35. mqtt_port=mqtt_port,
  36. mqtt_username=None,
  37. mqtt_password=None,
  38. retry_count=retry_count,
  39. )
  40. mqtt_client_mock.assert_called_once_with(
  41. userdata=switchbot_mqtt._MQTTCallbackUserdata(retry_count=retry_count)
  42. )
  43. assert not mqtt_client_mock().username_pw_set.called
  44. mqtt_client_mock().connect.assert_called_once_with(host=mqtt_host, port=mqtt_port)
  45. mqtt_client_mock().socket().getpeername.return_value = (mqtt_host, mqtt_port)
  46. with caplog.at_level(logging.DEBUG):
  47. mqtt_client_mock().on_connect(mqtt_client_mock(), None, {}, 0)
  48. assert mqtt_client_mock().subscribe.call_args_list == [
  49. unittest.mock.call("homeassistant/switch/switchbot/+/set"),
  50. unittest.mock.call("homeassistant/cover/switchbot-curtain/+/set"),
  51. ]
  52. assert mqtt_client_mock().message_callback_add.call_args_list == [
  53. unittest.mock.call(
  54. sub="homeassistant/switch/switchbot/+/set",
  55. callback=switchbot_mqtt._ButtonAutomator._mqtt_command_callback,
  56. ),
  57. unittest.mock.call(
  58. sub="homeassistant/cover/switchbot-curtain/+/set",
  59. callback=switchbot_mqtt._CurtainMotor._mqtt_command_callback,
  60. ),
  61. ]
  62. mqtt_client_mock().loop_forever.assert_called_once_with()
  63. assert caplog.record_tuples == [
  64. (
  65. "switchbot_mqtt",
  66. logging.INFO,
  67. "connecting to MQTT broker {}:{}".format(mqtt_host, mqtt_port),
  68. ),
  69. (
  70. "switchbot_mqtt",
  71. logging.DEBUG,
  72. "connected to MQTT broker {}:{}".format(mqtt_host, mqtt_port),
  73. ),
  74. (
  75. "switchbot_mqtt",
  76. logging.INFO,
  77. "subscribing to MQTT topic 'homeassistant/switch/switchbot/+/set'",
  78. ),
  79. (
  80. "switchbot_mqtt",
  81. logging.INFO,
  82. "subscribing to MQTT topic 'homeassistant/cover/switchbot-curtain/+/set'",
  83. ),
  84. ]
  85. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  86. @pytest.mark.parametrize("mqtt_port", [1833])
  87. @pytest.mark.parametrize("mqtt_username", ["me"])
  88. @pytest.mark.parametrize("mqtt_password", [None, "secret"])
  89. def test__run_authentication(mqtt_host, mqtt_port, mqtt_username, mqtt_password):
  90. with unittest.mock.patch("paho.mqtt.client.Client") as mqtt_client_mock:
  91. switchbot_mqtt._run(
  92. mqtt_host=mqtt_host,
  93. mqtt_port=mqtt_port,
  94. mqtt_username=mqtt_username,
  95. mqtt_password=mqtt_password,
  96. retry_count=7,
  97. )
  98. mqtt_client_mock.assert_called_once_with(
  99. userdata=switchbot_mqtt._MQTTCallbackUserdata(retry_count=7)
  100. )
  101. mqtt_client_mock().username_pw_set.assert_called_once_with(
  102. username=mqtt_username, password=mqtt_password
  103. )
  104. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  105. @pytest.mark.parametrize("mqtt_port", [1833])
  106. @pytest.mark.parametrize("mqtt_password", ["secret"])
  107. def test__run_authentication_missing_username(mqtt_host, mqtt_port, mqtt_password):
  108. with unittest.mock.patch("paho.mqtt.client.Client"):
  109. with pytest.raises(ValueError):
  110. switchbot_mqtt._run(
  111. mqtt_host=mqtt_host,
  112. mqtt_port=mqtt_port,
  113. mqtt_username=None,
  114. mqtt_password=mqtt_password,
  115. retry_count=3,
  116. )
  117. @pytest.mark.parametrize(
  118. ("command_topic_levels", "topic", "payload", "expected_mac_address"),
  119. [
  120. (
  121. switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  122. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  123. b"ON",
  124. "aa:bb:cc:dd:ee:ff",
  125. ),
  126. (
  127. switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  128. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  129. b"OFF",
  130. "aa:bb:cc:dd:ee:ff",
  131. ),
  132. (
  133. switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  134. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  135. b"on",
  136. "aa:bb:cc:dd:ee:ff",
  137. ),
  138. (
  139. switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  140. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  141. b"off",
  142. "aa:bb:cc:dd:ee:ff",
  143. ),
  144. (
  145. switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  146. b"homeassistant/switch/switchbot/aa:01:23:45:67:89/set",
  147. b"ON",
  148. "aa:01:23:45:67:89",
  149. ),
  150. (
  151. ["switchbot", switchbot_mqtt._MQTTTopicPlaceholder.MAC_ADDRESS],
  152. b"switchbot/aa:01:23:45:67:89",
  153. b"ON",
  154. "aa:01:23:45:67:89",
  155. ),
  156. (
  157. switchbot_mqtt._CurtainMotor.MQTT_COMMAND_TOPIC_LEVELS,
  158. b"homeassistant/cover/switchbot-curtain/aa:01:23:45:67:89/set",
  159. b"OPEN",
  160. "aa:01:23:45:67:89",
  161. ),
  162. ],
  163. )
  164. @pytest.mark.parametrize("retry_count", (3, 42))
  165. def test__mqtt_command_callback(
  166. caplog,
  167. command_topic_levels: typing.List[switchbot_mqtt._MQTTTopicLevel],
  168. topic: bytes,
  169. payload: bytes,
  170. expected_mac_address: str,
  171. retry_count: int,
  172. ):
  173. class _ActorMock(switchbot_mqtt._MQTTControlledActor):
  174. MQTT_COMMAND_TOPIC_LEVELS = command_topic_levels
  175. def __init__(self, mac_address, retry_count):
  176. super().__init__(mac_address=mac_address, retry_count=retry_count)
  177. def execute_command(self, mqtt_message_payload: bytes, mqtt_client: Client):
  178. pass
  179. message = MQTTMessage(topic=topic)
  180. message.payload = payload
  181. callback_userdata = switchbot_mqtt._MQTTCallbackUserdata(retry_count=retry_count)
  182. with unittest.mock.patch.object(
  183. _ActorMock, "__init__", return_value=None
  184. ) as init_mock, unittest.mock.patch.object(
  185. _ActorMock, "execute_command"
  186. ) as execute_command_mock, caplog.at_level(
  187. logging.DEBUG
  188. ):
  189. _ActorMock._mqtt_command_callback("client_dummy", callback_userdata, message)
  190. init_mock.assert_called_once_with(
  191. mac_address=expected_mac_address, retry_count=retry_count
  192. )
  193. execute_command_mock.assert_called_once_with(
  194. mqtt_client="client_dummy", mqtt_message_payload=payload
  195. )
  196. assert caplog.record_tuples == [
  197. (
  198. "switchbot_mqtt",
  199. logging.DEBUG,
  200. "received topic={} payload={!r}".format(topic.decode(), payload),
  201. )
  202. ]
  203. @pytest.mark.parametrize(
  204. ("topic", "payload"),
  205. [
  206. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff", b"on"),
  207. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/change", b"ON"),
  208. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set/suffix", b"ON"),
  209. ],
  210. )
  211. def test__mqtt_command_callback_unexpected_topic(caplog, topic: bytes, payload: bytes):
  212. class _ActorMock(switchbot_mqtt._MQTTControlledActor):
  213. MQTT_COMMAND_TOPIC_LEVELS = (
  214. switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
  215. )
  216. def __init__(self, mac_address, retry_count):
  217. super().__init__(mac_address=mac_address, retry_count=retry_count)
  218. def execute_command(self, mqtt_message_payload: bytes, mqtt_client: Client):
  219. pass
  220. message = MQTTMessage(topic=topic)
  221. message.payload = payload
  222. with unittest.mock.patch.object(
  223. _ActorMock, "__init__", return_value=None
  224. ) as init_mock, unittest.mock.patch.object(
  225. _ActorMock, "execute_command"
  226. ) as execute_command_mock, caplog.at_level(
  227. logging.DEBUG
  228. ):
  229. _ActorMock._mqtt_command_callback(
  230. "client_dummy", switchbot_mqtt._MQTTCallbackUserdata(retry_count=3), message
  231. )
  232. init_mock.assert_not_called()
  233. execute_command_mock.assert_not_called()
  234. assert caplog.record_tuples == [
  235. (
  236. "switchbot_mqtt",
  237. logging.DEBUG,
  238. "received topic={} payload={!r}".format(topic.decode(), payload),
  239. ),
  240. (
  241. "switchbot_mqtt",
  242. logging.WARNING,
  243. "unexpected topic {}".format(topic.decode()),
  244. ),
  245. ]
  246. @pytest.mark.parametrize(("mac_address", "payload"), [("aa:01:23:4E:RR:OR", b"ON")])
  247. def test__mqtt_command_callback_invalid_mac_address(
  248. caplog, mac_address: str, payload: bytes
  249. ):
  250. class _ActorMock(switchbot_mqtt._MQTTControlledActor):
  251. MQTT_COMMAND_TOPIC_LEVELS = (
  252. switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
  253. )
  254. def __init__(self, mac_address, retry_count):
  255. super().__init__(mac_address=mac_address, retry_count=retry_count)
  256. def execute_command(self, mqtt_message_payload: bytes, mqtt_client: Client):
  257. pass
  258. topic = "homeassistant/switch/switchbot/{}/set".format(mac_address).encode()
  259. message = MQTTMessage(topic=topic)
  260. message.payload = payload
  261. with unittest.mock.patch.object(
  262. _ActorMock, "__init__", return_value=None
  263. ) as init_mock, unittest.mock.patch.object(
  264. _ActorMock, "execute_command"
  265. ) as execute_command_mock, caplog.at_level(
  266. logging.DEBUG
  267. ):
  268. _ActorMock._mqtt_command_callback(
  269. "client_dummy",
  270. switchbot_mqtt._MQTTCallbackUserdata(retry_count=None),
  271. message,
  272. )
  273. init_mock.assert_not_called()
  274. execute_command_mock.assert_not_called()
  275. assert caplog.record_tuples == [
  276. (
  277. "switchbot_mqtt",
  278. logging.DEBUG,
  279. "received topic={} payload={!r}".format(topic.decode(), payload),
  280. ),
  281. (
  282. "switchbot_mqtt",
  283. logging.WARNING,
  284. "invalid mac address {}".format(mac_address),
  285. ),
  286. ]
  287. @pytest.mark.parametrize(
  288. ("topic", "payload"),
  289. [(b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set", b"ON")],
  290. )
  291. def test__mqtt_command_callback_ignore_retained(caplog, topic: bytes, payload: bytes):
  292. class _ActorMock(switchbot_mqtt._MQTTControlledActor):
  293. MQTT_COMMAND_TOPIC_LEVELS = (
  294. switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
  295. )
  296. def __init__(self, mac_address, retry_count):
  297. super().__init__(mac_address=mac_address, retry_count=retry_count)
  298. def execute_command(self, mqtt_message_payload: bytes, mqtt_client: Client):
  299. pass
  300. message = MQTTMessage(topic=topic)
  301. message.payload = payload
  302. message.retain = True
  303. with unittest.mock.patch.object(
  304. _ActorMock, "__init__", return_value=None
  305. ) as init_mock, unittest.mock.patch.object(
  306. _ActorMock, "execute_command"
  307. ) as execute_command_mock, caplog.at_level(
  308. logging.DEBUG
  309. ):
  310. _ActorMock._mqtt_command_callback(
  311. "client_dummy",
  312. switchbot_mqtt._MQTTCallbackUserdata(retry_count=None),
  313. message,
  314. )
  315. init_mock.assert_not_called()
  316. execute_command_mock.assert_not_called()
  317. assert caplog.record_tuples == [
  318. (
  319. "switchbot_mqtt",
  320. logging.DEBUG,
  321. "received topic={} payload={!r}".format(topic.decode(), payload),
  322. ),
  323. ("switchbot_mqtt", logging.INFO, "ignoring retained message"),
  324. ]
  325. @pytest.mark.parametrize(
  326. ("state_topic_levels", "mac_address", "expected_topic"),
  327. # https://www.home-assistant.io/docs/mqtt/discovery/#switches
  328. [
  329. (
  330. switchbot_mqtt._ButtonAutomator.MQTT_STATE_TOPIC_LEVELS,
  331. "aa:bb:cc:dd:ee:ff",
  332. "homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/state",
  333. ),
  334. (
  335. ["switchbot", switchbot_mqtt._MQTTTopicPlaceholder.MAC_ADDRESS, "state"],
  336. "aa:bb:cc:dd:ee:gg",
  337. "switchbot/aa:bb:cc:dd:ee:gg/state",
  338. ),
  339. ],
  340. )
  341. @pytest.mark.parametrize("state", [b"ON", b"CLOSE"])
  342. @pytest.mark.parametrize("return_code", [MQTT_ERR_SUCCESS, MQTT_ERR_QUEUE_SIZE])
  343. def test__report_state(
  344. caplog,
  345. state_topic_levels: typing.List[switchbot_mqtt._MQTTTopicLevel],
  346. mac_address: str,
  347. expected_topic: str,
  348. state: bytes,
  349. return_code: int,
  350. ):
  351. # pylint: disable=too-many-arguments
  352. class _ActorMock(switchbot_mqtt._MQTTControlledActor):
  353. MQTT_STATE_TOPIC_LEVELS = state_topic_levels
  354. def __init__(self, mac_address, retry_count):
  355. super().__init__(mac_address=mac_address, retry_count=retry_count)
  356. def execute_command(self, mqtt_message_payload: bytes, mqtt_client: Client):
  357. pass
  358. mqtt_client_mock = unittest.mock.MagicMock()
  359. mqtt_client_mock.publish.return_value.rc = return_code
  360. with caplog.at_level(logging.DEBUG):
  361. _ActorMock(mac_address=mac_address, retry_count=3).report_state(
  362. state=state, mqtt_client=mqtt_client_mock
  363. )
  364. mqtt_client_mock.publish.assert_called_once_with(
  365. topic=expected_topic, payload=state, retain=True
  366. )
  367. assert caplog.record_tuples[0] == (
  368. "switchbot_mqtt",
  369. logging.DEBUG,
  370. "publishing topic={} payload={!r}".format(expected_topic, state),
  371. )
  372. if return_code == MQTT_ERR_SUCCESS:
  373. assert not caplog.records[1:]
  374. else:
  375. assert caplog.record_tuples[1:] == [
  376. (
  377. "switchbot_mqtt",
  378. logging.ERROR,
  379. "Failed to publish MQTT message on topic {} (rc={})".format(
  380. expected_topic, return_code
  381. ),
  382. )
  383. ]