1
0

test_switchbot_curtain_motor_position.py 8.3 KB

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