1
0

test_mqtt.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609
  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.logging
  22. import pytest
  23. from paho.mqtt.client import MQTT_ERR_QUEUE_SIZE, MQTT_ERR_SUCCESS, MQTTMessage, Client
  24. import switchbot_mqtt
  25. import switchbot_mqtt._actors
  26. from switchbot_mqtt._actors import _ButtonAutomator, _CurtainMotor
  27. from switchbot_mqtt._actors._base import _MQTTCallbackUserdata, _MQTTControlledActor
  28. from switchbot_mqtt._utils import _MQTTTopicLevel, _MQTTTopicPlaceholder
  29. # pylint: disable=protected-access
  30. # pylint: disable=too-many-arguments; these are tests, no API
  31. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  32. @pytest.mark.parametrize("mqtt_port", [1833])
  33. @pytest.mark.parametrize("retry_count", [3, 21])
  34. @pytest.mark.parametrize(
  35. "device_passwords",
  36. [{}, {"11:22:33:44:55:66": "password", "aa:bb:cc:dd:ee:ff": "secret"}],
  37. )
  38. @pytest.mark.parametrize("fetch_device_info", [True, False])
  39. def test__run(
  40. caplog: _pytest.logging.LogCaptureFixture,
  41. mqtt_host: str,
  42. mqtt_port: int,
  43. retry_count: int,
  44. device_passwords: typing.Dict[str, str],
  45. fetch_device_info: bool,
  46. ) -> None:
  47. with unittest.mock.patch(
  48. "paho.mqtt.client.Client"
  49. ) as mqtt_client_mock, caplog.at_level(logging.DEBUG):
  50. switchbot_mqtt._run(
  51. mqtt_host=mqtt_host,
  52. mqtt_port=mqtt_port,
  53. mqtt_username=None,
  54. mqtt_password=None,
  55. retry_count=retry_count,
  56. device_passwords=device_passwords,
  57. fetch_device_info=fetch_device_info,
  58. )
  59. mqtt_client_mock.assert_called_once()
  60. assert not mqtt_client_mock.call_args[0]
  61. assert set(mqtt_client_mock.call_args[1].keys()) == {"userdata"}
  62. userdata = mqtt_client_mock.call_args[1]["userdata"]
  63. assert userdata == _MQTTCallbackUserdata(
  64. retry_count=retry_count,
  65. device_passwords=device_passwords,
  66. fetch_device_info=fetch_device_info,
  67. )
  68. assert not mqtt_client_mock().username_pw_set.called
  69. mqtt_client_mock().connect.assert_called_once_with(host=mqtt_host, port=mqtt_port)
  70. mqtt_client_mock().socket().getpeername.return_value = (mqtt_host, mqtt_port)
  71. with caplog.at_level(logging.DEBUG):
  72. mqtt_client_mock().on_connect(mqtt_client_mock(), userdata, {}, 0)
  73. subscribe_mock = mqtt_client_mock().subscribe
  74. assert subscribe_mock.call_count == (4 if fetch_device_info else 2)
  75. for topic in [
  76. "homeassistant/switch/switchbot/+/set",
  77. "homeassistant/cover/switchbot-curtain/+/set",
  78. ]:
  79. assert unittest.mock.call(topic) in subscribe_mock.call_args_list
  80. for topic in [
  81. "homeassistant/switch/switchbot/+/request-device-info",
  82. "homeassistant/cover/switchbot-curtain/+/request-device-info",
  83. ]:
  84. assert (
  85. unittest.mock.call(topic) in subscribe_mock.call_args_list
  86. ) == fetch_device_info
  87. mqtt_client_mock().loop_forever.assert_called_once_with()
  88. assert caplog.record_tuples[:2] == [
  89. (
  90. "switchbot_mqtt",
  91. logging.INFO,
  92. f"connecting to MQTT broker {mqtt_host}:{mqtt_port}",
  93. ),
  94. (
  95. "switchbot_mqtt",
  96. logging.DEBUG,
  97. f"connected to MQTT broker {mqtt_host}:{mqtt_port}",
  98. ),
  99. ]
  100. assert len(caplog.record_tuples) == (6 if fetch_device_info else 4)
  101. assert (
  102. "switchbot_mqtt._actors._base",
  103. logging.INFO,
  104. "subscribing to MQTT topic 'homeassistant/switch/switchbot/+/set'",
  105. ) in caplog.record_tuples
  106. assert (
  107. "switchbot_mqtt._actors._base",
  108. logging.INFO,
  109. "subscribing to MQTT topic 'homeassistant/cover/switchbot-curtain/+/set'",
  110. ) in caplog.record_tuples
  111. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  112. @pytest.mark.parametrize("mqtt_port", [1833])
  113. @pytest.mark.parametrize("mqtt_username", ["me"])
  114. @pytest.mark.parametrize("mqtt_password", [None, "secret"])
  115. def test__run_authentication(
  116. mqtt_host: str,
  117. mqtt_port: int,
  118. mqtt_username: str,
  119. mqtt_password: typing.Optional[str],
  120. ) -> None:
  121. with unittest.mock.patch("paho.mqtt.client.Client") as mqtt_client_mock:
  122. switchbot_mqtt._run(
  123. mqtt_host=mqtt_host,
  124. mqtt_port=mqtt_port,
  125. mqtt_username=mqtt_username,
  126. mqtt_password=mqtt_password,
  127. retry_count=7,
  128. device_passwords={},
  129. fetch_device_info=True,
  130. )
  131. mqtt_client_mock.assert_called_once_with(
  132. userdata=_MQTTCallbackUserdata(
  133. retry_count=7, device_passwords={}, fetch_device_info=True
  134. )
  135. )
  136. mqtt_client_mock().username_pw_set.assert_called_once_with(
  137. username=mqtt_username, password=mqtt_password
  138. )
  139. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  140. @pytest.mark.parametrize("mqtt_port", [1833])
  141. @pytest.mark.parametrize("mqtt_password", ["secret"])
  142. def test__run_authentication_missing_username(
  143. mqtt_host: str, mqtt_port: int, mqtt_password: str
  144. ) -> None:
  145. with unittest.mock.patch("paho.mqtt.client.Client"):
  146. with pytest.raises(ValueError):
  147. switchbot_mqtt._run(
  148. mqtt_host=mqtt_host,
  149. mqtt_port=mqtt_port,
  150. mqtt_username=None,
  151. mqtt_password=mqtt_password,
  152. retry_count=3,
  153. device_passwords={},
  154. fetch_device_info=True,
  155. )
  156. def _mock_actor_class(
  157. *,
  158. command_topic_levels: typing.List[_MQTTTopicLevel] = NotImplemented,
  159. request_info_levels: typing.List[_MQTTTopicLevel] = NotImplemented,
  160. ) -> typing.Type:
  161. class _ActorMock(_MQTTControlledActor):
  162. MQTT_COMMAND_TOPIC_LEVELS = command_topic_levels
  163. _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = request_info_levels
  164. def __init__(
  165. self, mac_address: str, retry_count: int, password: typing.Optional[str]
  166. ) -> None:
  167. super().__init__(
  168. mac_address=mac_address, retry_count=retry_count, password=password
  169. )
  170. def execute_command(
  171. self,
  172. mqtt_message_payload: bytes,
  173. mqtt_client: Client,
  174. update_device_info: bool,
  175. ) -> None:
  176. pass
  177. def _get_device(self) -> None:
  178. return None
  179. return _ActorMock
  180. @pytest.mark.parametrize(
  181. ("topic_levels", "topic", "expected_mac_address"),
  182. [
  183. (
  184. switchbot_mqtt._actors._ButtonAutomator._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
  185. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/request-device-info",
  186. "aa:bb:cc:dd:ee:ff",
  187. ),
  188. ],
  189. )
  190. @pytest.mark.parametrize("payload", [b"", b"whatever"])
  191. def test__mqtt_update_device_info_callback(
  192. caplog: _pytest.logging.LogCaptureFixture,
  193. topic_levels: typing.List[_MQTTTopicLevel],
  194. topic: bytes,
  195. expected_mac_address: str,
  196. payload: bytes,
  197. ) -> None:
  198. ActorMock = _mock_actor_class(request_info_levels=topic_levels)
  199. message = MQTTMessage(topic=topic)
  200. message.payload = payload
  201. callback_userdata = _MQTTCallbackUserdata(
  202. retry_count=21, # tested in test__mqtt_command_callback
  203. device_passwords={},
  204. fetch_device_info=True,
  205. )
  206. with unittest.mock.patch.object(
  207. ActorMock, "__init__", return_value=None
  208. ) as init_mock, unittest.mock.patch.object(
  209. ActorMock, "_update_and_report_device_info"
  210. ) as update_mock, caplog.at_level(
  211. logging.DEBUG
  212. ):
  213. ActorMock._mqtt_update_device_info_callback(
  214. "client_dummy", callback_userdata, message
  215. )
  216. init_mock.assert_called_once_with(
  217. mac_address=expected_mac_address, retry_count=21, password=None
  218. )
  219. update_mock.assert_called_once_with("client_dummy")
  220. assert caplog.record_tuples == [
  221. (
  222. "switchbot_mqtt._actors._base",
  223. logging.DEBUG,
  224. f"received topic={topic.decode()} payload={payload!r}",
  225. )
  226. ]
  227. def test__mqtt_update_device_info_callback_ignore_retained(
  228. caplog: _pytest.logging.LogCaptureFixture,
  229. ) -> None:
  230. ActorMock = _mock_actor_class(
  231. request_info_levels=[_MQTTTopicPlaceholder.MAC_ADDRESS, "request"]
  232. )
  233. message = MQTTMessage(topic=b"aa:bb:cc:dd:ee:ff/request")
  234. message.payload = b""
  235. message.retain = True
  236. with unittest.mock.patch.object(
  237. ActorMock, "__init__", return_value=None
  238. ) as init_mock, unittest.mock.patch.object(
  239. ActorMock, "execute_command"
  240. ) as execute_command_mock, caplog.at_level(
  241. logging.DEBUG
  242. ):
  243. ActorMock._mqtt_update_device_info_callback(
  244. "client_dummy",
  245. _MQTTCallbackUserdata(
  246. retry_count=21, device_passwords={}, fetch_device_info=True
  247. ),
  248. message,
  249. )
  250. init_mock.assert_not_called()
  251. execute_command_mock.assert_not_called()
  252. assert caplog.record_tuples == [
  253. (
  254. "switchbot_mqtt._actors._base",
  255. logging.DEBUG,
  256. "received topic=aa:bb:cc:dd:ee:ff/request payload=b''",
  257. ),
  258. ("switchbot_mqtt._actors._base", logging.INFO, "ignoring retained message"),
  259. ]
  260. @pytest.mark.parametrize(
  261. ("command_topic_levels", "topic", "payload", "expected_mac_address"),
  262. [
  263. (
  264. _ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  265. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  266. b"ON",
  267. "aa:bb:cc:dd:ee:ff",
  268. ),
  269. (
  270. _ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  271. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  272. b"OFF",
  273. "aa:bb:cc:dd:ee:ff",
  274. ),
  275. (
  276. _ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  277. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  278. b"on",
  279. "aa:bb:cc:dd:ee:ff",
  280. ),
  281. (
  282. _ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  283. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  284. b"off",
  285. "aa:bb:cc:dd:ee:ff",
  286. ),
  287. (
  288. _ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS,
  289. b"homeassistant/switch/switchbot/aa:01:23:45:67:89/set",
  290. b"ON",
  291. "aa:01:23:45:67:89",
  292. ),
  293. (
  294. ["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS],
  295. b"switchbot/aa:01:23:45:67:89",
  296. b"ON",
  297. "aa:01:23:45:67:89",
  298. ),
  299. (
  300. _CurtainMotor.MQTT_COMMAND_TOPIC_LEVELS,
  301. b"homeassistant/cover/switchbot-curtain/aa:01:23:45:67:89/set",
  302. b"OPEN",
  303. "aa:01:23:45:67:89",
  304. ),
  305. ],
  306. )
  307. @pytest.mark.parametrize("retry_count", (3, 42))
  308. @pytest.mark.parametrize("fetch_device_info", [True, False])
  309. def test__mqtt_command_callback(
  310. caplog: _pytest.logging.LogCaptureFixture,
  311. command_topic_levels: typing.List[_MQTTTopicLevel],
  312. topic: bytes,
  313. payload: bytes,
  314. expected_mac_address: str,
  315. retry_count: int,
  316. fetch_device_info: bool,
  317. ) -> None:
  318. ActorMock = _mock_actor_class(command_topic_levels=command_topic_levels)
  319. message = MQTTMessage(topic=topic)
  320. message.payload = payload
  321. callback_userdata = _MQTTCallbackUserdata(
  322. retry_count=retry_count,
  323. device_passwords={},
  324. fetch_device_info=fetch_device_info,
  325. )
  326. with unittest.mock.patch.object(
  327. ActorMock, "__init__", return_value=None
  328. ) as init_mock, unittest.mock.patch.object(
  329. ActorMock, "execute_command"
  330. ) as execute_command_mock, caplog.at_level(
  331. logging.DEBUG
  332. ):
  333. ActorMock._mqtt_command_callback("client_dummy", callback_userdata, message)
  334. init_mock.assert_called_once_with(
  335. mac_address=expected_mac_address, retry_count=retry_count, password=None
  336. )
  337. execute_command_mock.assert_called_once_with(
  338. mqtt_client="client_dummy",
  339. mqtt_message_payload=payload,
  340. update_device_info=fetch_device_info,
  341. )
  342. assert caplog.record_tuples == [
  343. (
  344. "switchbot_mqtt._actors._base",
  345. logging.DEBUG,
  346. f"received topic={topic.decode()} payload={payload!r}",
  347. )
  348. ]
  349. @pytest.mark.parametrize(
  350. ("mac_address", "expected_password"),
  351. [
  352. ("11:22:33:44:55:66", None),
  353. ("aa:bb:cc:dd:ee:ff", "secret"),
  354. ("11:22:33:dd:ee:ff", "äöü"),
  355. ],
  356. )
  357. def test__mqtt_command_callback_password(
  358. mac_address: str, expected_password: typing.Optional[str]
  359. ) -> None:
  360. ActorMock = _mock_actor_class(
  361. command_topic_levels=["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS]
  362. )
  363. message = MQTTMessage(topic=b"switchbot/" + mac_address.encode())
  364. message.payload = b"whatever"
  365. callback_userdata = _MQTTCallbackUserdata(
  366. retry_count=3,
  367. device_passwords={
  368. "11:22:33:44:55:77": "test",
  369. "aa:bb:cc:dd:ee:ff": "secret",
  370. "11:22:33:dd:ee:ff": "äöü",
  371. },
  372. fetch_device_info=True,
  373. )
  374. with unittest.mock.patch.object(
  375. ActorMock, "__init__", return_value=None
  376. ) as init_mock, unittest.mock.patch.object(
  377. ActorMock, "execute_command"
  378. ) as execute_command_mock:
  379. ActorMock._mqtt_command_callback("client_dummy", callback_userdata, message)
  380. init_mock.assert_called_once_with(
  381. mac_address=mac_address, retry_count=3, password=expected_password
  382. )
  383. execute_command_mock.assert_called_once_with(
  384. mqtt_client="client_dummy",
  385. mqtt_message_payload=b"whatever",
  386. update_device_info=True,
  387. )
  388. @pytest.mark.parametrize(
  389. ("topic", "payload"),
  390. [
  391. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff", b"on"),
  392. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/change", b"ON"),
  393. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set/suffix", b"ON"),
  394. ],
  395. )
  396. def test__mqtt_command_callback_unexpected_topic(
  397. caplog: _pytest.logging.LogCaptureFixture, topic: bytes, payload: bytes
  398. ) -> None:
  399. ActorMock = _mock_actor_class(
  400. command_topic_levels=_ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
  401. )
  402. message = MQTTMessage(topic=topic)
  403. message.payload = payload
  404. with unittest.mock.patch.object(
  405. ActorMock, "__init__", return_value=None
  406. ) as init_mock, unittest.mock.patch.object(
  407. ActorMock, "execute_command"
  408. ) as execute_command_mock, caplog.at_level(
  409. logging.DEBUG
  410. ):
  411. ActorMock._mqtt_command_callback(
  412. "client_dummy",
  413. _MQTTCallbackUserdata(
  414. retry_count=3, device_passwords={}, fetch_device_info=True
  415. ),
  416. message,
  417. )
  418. init_mock.assert_not_called()
  419. execute_command_mock.assert_not_called()
  420. assert caplog.record_tuples == [
  421. (
  422. "switchbot_mqtt._actors._base",
  423. logging.DEBUG,
  424. f"received topic={topic.decode()} payload={payload!r}",
  425. ),
  426. (
  427. "switchbot_mqtt._actors._base",
  428. logging.WARNING,
  429. f"unexpected topic {topic.decode()}",
  430. ),
  431. ]
  432. @pytest.mark.parametrize(("mac_address", "payload"), [("aa:01:23:4E:RR:OR", b"ON")])
  433. def test__mqtt_command_callback_invalid_mac_address(
  434. caplog: _pytest.logging.LogCaptureFixture, mac_address: str, payload: bytes
  435. ) -> None:
  436. ActorMock = _mock_actor_class(
  437. command_topic_levels=_ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
  438. )
  439. topic = f"homeassistant/switch/switchbot/{mac_address}/set".encode()
  440. message = MQTTMessage(topic=topic)
  441. message.payload = payload
  442. with unittest.mock.patch.object(
  443. ActorMock, "__init__", return_value=None
  444. ) as init_mock, unittest.mock.patch.object(
  445. ActorMock, "execute_command"
  446. ) as execute_command_mock, caplog.at_level(
  447. logging.DEBUG
  448. ):
  449. ActorMock._mqtt_command_callback(
  450. "client_dummy",
  451. _MQTTCallbackUserdata(
  452. retry_count=3, device_passwords={}, fetch_device_info=True
  453. ),
  454. message,
  455. )
  456. init_mock.assert_not_called()
  457. execute_command_mock.assert_not_called()
  458. assert caplog.record_tuples == [
  459. (
  460. "switchbot_mqtt._actors._base",
  461. logging.DEBUG,
  462. f"received topic={topic.decode()} payload={payload!r}",
  463. ),
  464. (
  465. "switchbot_mqtt._actors._base",
  466. logging.WARNING,
  467. f"invalid mac address {mac_address}",
  468. ),
  469. ]
  470. @pytest.mark.parametrize(
  471. ("topic", "payload"),
  472. [(b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set", b"ON")],
  473. )
  474. def test__mqtt_command_callback_ignore_retained(
  475. caplog: _pytest.logging.LogCaptureFixture, topic: bytes, payload: bytes
  476. ) -> None:
  477. ActorMock = _mock_actor_class(
  478. command_topic_levels=_ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
  479. )
  480. message = MQTTMessage(topic=topic)
  481. message.payload = payload
  482. message.retain = True
  483. with unittest.mock.patch.object(
  484. ActorMock, "__init__", return_value=None
  485. ) as init_mock, unittest.mock.patch.object(
  486. ActorMock, "execute_command"
  487. ) as execute_command_mock, caplog.at_level(
  488. logging.DEBUG
  489. ):
  490. ActorMock._mqtt_command_callback(
  491. "client_dummy",
  492. _MQTTCallbackUserdata(
  493. retry_count=4, device_passwords={}, fetch_device_info=True
  494. ),
  495. message,
  496. )
  497. init_mock.assert_not_called()
  498. execute_command_mock.assert_not_called()
  499. assert caplog.record_tuples == [
  500. (
  501. "switchbot_mqtt._actors._base",
  502. logging.DEBUG,
  503. f"received topic={topic.decode()} payload={payload!r}",
  504. ),
  505. ("switchbot_mqtt._actors._base", logging.INFO, "ignoring retained message"),
  506. ]
  507. @pytest.mark.parametrize(
  508. ("state_topic_levels", "mac_address", "expected_topic"),
  509. # https://www.home-assistant.io/docs/mqtt/discovery/#switches
  510. [
  511. (
  512. _ButtonAutomator.MQTT_STATE_TOPIC_LEVELS,
  513. "aa:bb:cc:dd:ee:ff",
  514. "homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/state",
  515. ),
  516. (
  517. ["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS, "state"],
  518. "aa:bb:cc:dd:ee:gg",
  519. "switchbot/aa:bb:cc:dd:ee:gg/state",
  520. ),
  521. ],
  522. )
  523. @pytest.mark.parametrize("state", [b"ON", b"CLOSE"])
  524. @pytest.mark.parametrize("return_code", [MQTT_ERR_SUCCESS, MQTT_ERR_QUEUE_SIZE])
  525. def test__report_state(
  526. caplog: _pytest.logging.LogCaptureFixture,
  527. state_topic_levels: typing.List[_MQTTTopicLevel],
  528. mac_address: str,
  529. expected_topic: str,
  530. state: bytes,
  531. return_code: int,
  532. ) -> None:
  533. # pylint: disable=too-many-arguments
  534. class _ActorMock(_MQTTControlledActor):
  535. MQTT_STATE_TOPIC_LEVELS = state_topic_levels
  536. def __init__(
  537. self, mac_address: str, retry_count: int, password: typing.Optional[str]
  538. ) -> None:
  539. super().__init__(
  540. mac_address=mac_address, retry_count=retry_count, password=password
  541. )
  542. def execute_command(
  543. self,
  544. mqtt_message_payload: bytes,
  545. mqtt_client: Client,
  546. update_device_info: bool,
  547. ) -> None:
  548. pass
  549. def _get_device(self) -> None:
  550. return None
  551. mqtt_client_mock = unittest.mock.MagicMock()
  552. mqtt_client_mock.publish.return_value.rc = return_code
  553. with caplog.at_level(logging.DEBUG):
  554. actor = _ActorMock(mac_address=mac_address, retry_count=3, password=None)
  555. actor.report_state(state=state, mqtt_client=mqtt_client_mock)
  556. mqtt_client_mock.publish.assert_called_once_with(
  557. topic=expected_topic, payload=state, retain=True
  558. )
  559. assert caplog.record_tuples[0] == (
  560. "switchbot_mqtt._actors._base",
  561. logging.DEBUG,
  562. f"publishing topic={expected_topic} payload={state!r}",
  563. )
  564. if return_code == MQTT_ERR_SUCCESS:
  565. assert not caplog.records[1:]
  566. else:
  567. assert caplog.record_tuples[1:] == [
  568. (
  569. "switchbot_mqtt._actors._base",
  570. logging.ERROR,
  571. f"Failed to publish MQTT message on topic {expected_topic} (rc={return_code})",
  572. )
  573. ]