1
0

test_switchbot_curtain_motor_position.py 9.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287
  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) 2022 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 unittest.mock
  20. import aiomqtt
  21. import bleak
  22. import _pytest.logging # pylint: disable=import-private-name; typing
  23. import pytest
  24. # pylint: disable=import-private-name; internal
  25. from switchbot_mqtt._actors import _CurtainMotor
  26. # pylint: disable=protected-access
  27. @pytest.mark.asyncio
  28. @pytest.mark.parametrize(
  29. ("topic", "payload", "expected_mac_address", "expected_position_percent"),
  30. [
  31. (
  32. "home/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
  33. b"42",
  34. "aa:bb:cc:dd:ee:ff",
  35. 42,
  36. ),
  37. (
  38. "home/cover/switchbot-curtain/11:22:33:44:55:66/position/set-percent",
  39. b"0",
  40. "11:22:33:44:55:66",
  41. 0,
  42. ),
  43. (
  44. "home/cover/switchbot-curtain/11:22:33:44:55:66/position/set-percent",
  45. b"100",
  46. "11:22:33:44:55:66",
  47. 100,
  48. ),
  49. ],
  50. )
  51. @pytest.mark.parametrize("retry_count", (3, 42))
  52. async def test__mqtt_set_position_callback(
  53. caplog: _pytest.logging.LogCaptureFixture,
  54. topic: str,
  55. payload: bytes,
  56. expected_mac_address: str,
  57. retry_count: int,
  58. expected_position_percent: int,
  59. ) -> None:
  60. message = aiomqtt.Message(
  61. topic=topic, payload=payload, qos=0, retain=False, mid=0, properties=None
  62. )
  63. device = unittest.mock.Mock()
  64. device.address = expected_mac_address
  65. with unittest.mock.patch.object(
  66. bleak.BleakScanner, "find_device_by_address", return_value=device
  67. ), unittest.mock.patch(
  68. "switchbot.SwitchbotCurtain.__init__", return_value=None
  69. ) as device_init_mock, unittest.mock.patch(
  70. "switchbot.SwitchbotCurtain.set_position"
  71. ) as set_position_mock, caplog.at_level(
  72. logging.DEBUG
  73. ):
  74. await _CurtainMotor._mqtt_set_position_callback(
  75. mqtt_client=unittest.mock.Mock(),
  76. message=message,
  77. retry_count=retry_count,
  78. device_passwords={},
  79. fetch_device_info=False,
  80. mqtt_topic_prefix="home/",
  81. )
  82. device_init_mock.assert_called_once_with(
  83. device=device, password=None, retry_count=retry_count, reverse_mode=True
  84. )
  85. set_position_mock.assert_called_once_with(expected_position_percent)
  86. assert caplog.record_tuples == [
  87. (
  88. "switchbot_mqtt._actors",
  89. logging.DEBUG,
  90. f"received topic=home/cover/switchbot-curtain/{expected_mac_address}"
  91. f"/position/set-percent payload=b'{expected_position_percent}'",
  92. ),
  93. (
  94. "switchbot_mqtt._actors",
  95. logging.INFO,
  96. f"set position of switchbot curtain {expected_mac_address}"
  97. f" to {expected_position_percent}%",
  98. ),
  99. ]
  100. @pytest.mark.asyncio
  101. async def test__mqtt_set_position_callback_ignore_retained(
  102. caplog: _pytest.logging.LogCaptureFixture,
  103. ) -> None:
  104. message = aiomqtt.Message(
  105. topic="homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
  106. payload=b"42",
  107. qos=0,
  108. retain=True,
  109. mid=0,
  110. properties=None,
  111. )
  112. with unittest.mock.patch(
  113. "switchbot.SwitchbotCurtain"
  114. ) as device_init_mock, caplog.at_level(logging.INFO):
  115. await _CurtainMotor._mqtt_set_position_callback(
  116. mqtt_client=unittest.mock.Mock(),
  117. message=message,
  118. retry_count=3,
  119. device_passwords={},
  120. fetch_device_info=False,
  121. mqtt_topic_prefix="whatever",
  122. )
  123. device_init_mock.assert_not_called()
  124. assert caplog.record_tuples == [
  125. (
  126. "switchbot_mqtt._actors",
  127. logging.INFO,
  128. "ignoring retained message on topic"
  129. " homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
  130. ),
  131. ]
  132. @pytest.mark.asyncio
  133. async def test__mqtt_set_position_callback_unexpected_topic(
  134. caplog: _pytest.logging.LogCaptureFixture,
  135. ) -> None:
  136. message = aiomqtt.Message(
  137. topic="switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set",
  138. payload=b"42",
  139. qos=0,
  140. retain=False,
  141. mid=0,
  142. properties=None,
  143. )
  144. with unittest.mock.patch(
  145. "switchbot.SwitchbotCurtain"
  146. ) as device_init_mock, caplog.at_level(logging.INFO):
  147. await _CurtainMotor._mqtt_set_position_callback(
  148. mqtt_client=unittest.mock.Mock(),
  149. message=message,
  150. retry_count=3,
  151. device_passwords={},
  152. fetch_device_info=False,
  153. mqtt_topic_prefix="",
  154. )
  155. device_init_mock.assert_not_called()
  156. assert caplog.record_tuples == [
  157. (
  158. "switchbot_mqtt._actors.base",
  159. logging.WARN,
  160. "unexpected topic switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set",
  161. ),
  162. ]
  163. @pytest.mark.asyncio
  164. async def test__mqtt_set_position_callback_invalid_mac_address(
  165. caplog: _pytest.logging.LogCaptureFixture,
  166. ) -> None:
  167. message = aiomqtt.Message(
  168. topic="tnatsissaemoh/cover/switchbot-curtain/aa:bb:cc:dd:ee/position/set-percent",
  169. payload=b"42",
  170. qos=0,
  171. retain=False,
  172. mid=0,
  173. properties=None,
  174. )
  175. with unittest.mock.patch(
  176. "switchbot.SwitchbotCurtain"
  177. ) as device_init_mock, caplog.at_level(logging.INFO):
  178. await _CurtainMotor._mqtt_set_position_callback(
  179. mqtt_client=unittest.mock.Mock(),
  180. message=message,
  181. retry_count=3,
  182. device_passwords={},
  183. fetch_device_info=False,
  184. mqtt_topic_prefix="tnatsissaemoh/",
  185. )
  186. device_init_mock.assert_not_called()
  187. assert caplog.record_tuples == [
  188. (
  189. "switchbot_mqtt._actors.base",
  190. logging.WARN,
  191. "invalid mac address aa:bb:cc:dd:ee",
  192. ),
  193. ]
  194. @pytest.mark.asyncio
  195. @pytest.mark.parametrize("payload", [b"-1", b"123"])
  196. async def test__mqtt_set_position_callback_invalid_position(
  197. caplog: _pytest.logging.LogCaptureFixture,
  198. payload: bytes,
  199. ) -> None:
  200. message = aiomqtt.Message(
  201. topic="homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
  202. payload=payload,
  203. qos=0,
  204. retain=False,
  205. mid=0,
  206. properties=None,
  207. )
  208. with unittest.mock.patch.object(
  209. bleak.BleakScanner, "find_device_by_address"
  210. ), unittest.mock.patch(
  211. "switchbot.SwitchbotCurtain"
  212. ) as device_init_mock, caplog.at_level(
  213. logging.INFO
  214. ):
  215. await _CurtainMotor._mqtt_set_position_callback(
  216. mqtt_client=unittest.mock.Mock(),
  217. message=message,
  218. retry_count=3,
  219. device_passwords={},
  220. fetch_device_info=False,
  221. mqtt_topic_prefix="homeassistant/",
  222. )
  223. device_init_mock.assert_called_once()
  224. device_init_mock().set_position.assert_not_called()
  225. assert caplog.record_tuples == [
  226. (
  227. "switchbot_mqtt._actors",
  228. logging.WARN,
  229. f"invalid position {payload.decode()}%, ignoring message",
  230. ),
  231. ]
  232. @pytest.mark.asyncio
  233. async def test__mqtt_set_position_callback_command_failed(
  234. caplog: _pytest.logging.LogCaptureFixture,
  235. ) -> None:
  236. message = aiomqtt.Message(
  237. topic="cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
  238. payload=b"21",
  239. qos=0,
  240. retain=False,
  241. mid=0,
  242. properties=None,
  243. )
  244. device = unittest.mock.Mock()
  245. device.address = "aa:bb:cc:dd:ee:ff"
  246. with unittest.mock.patch.object(
  247. bleak.BleakScanner, "find_device_by_address", return_value=device
  248. ), unittest.mock.patch(
  249. "switchbot.SwitchbotCurtain.__init__", return_value=None
  250. ) as device_init_mock, unittest.mock.patch(
  251. "switchbot.SwitchbotCurtain.set_position", return_value=False
  252. ) as set_position_mock, caplog.at_level(
  253. logging.INFO
  254. ):
  255. await _CurtainMotor._mqtt_set_position_callback(
  256. mqtt_client=unittest.mock.Mock(),
  257. message=message,
  258. retry_count=3,
  259. device_passwords={},
  260. fetch_device_info=False,
  261. mqtt_topic_prefix="",
  262. )
  263. device_init_mock.assert_called_once()
  264. set_position_mock.assert_awaited_with(21)
  265. assert caplog.record_tuples == [
  266. (
  267. "switchbot_mqtt._actors",
  268. logging.ERROR,
  269. "failed to set position of switchbot curtain aa:bb:cc:dd:ee:ff",
  270. ),
  271. ]