test_mqtt.py 14 KB

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