1
0

test_switchbot_curtain_motor.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348
  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 # pylint: disable=import-private-name; typing
  22. import bluepy.btle
  23. import pytest
  24. # pylint: disable=import-private-name; internal
  25. import switchbot_mqtt._utils
  26. from switchbot_mqtt._actors import _CurtainMotor
  27. # pylint: disable=protected-access,
  28. # pylint: disable=too-many-arguments; these are tests, no API
  29. @pytest.mark.parametrize("mac_address", ["{MAC_ADDRESS}", "aa:bb:cc:dd:ee:ff"])
  30. def test_get_mqtt_battery_percentage_topic(mac_address: str) -> None:
  31. assert (
  32. _CurtainMotor.get_mqtt_battery_percentage_topic(
  33. prefix="homeassistant/", mac_address=mac_address
  34. )
  35. == f"homeassistant/cover/switchbot-curtain/{mac_address}/battery-percentage"
  36. )
  37. @pytest.mark.parametrize("mac_address", ["{MAC_ADDRESS}", "aa:bb:cc:dd:ee:ff"])
  38. def test_get_mqtt_position_topic(mac_address: str) -> None:
  39. assert (
  40. _CurtainMotor.get_mqtt_position_topic(prefix="prfx-", mac_address=mac_address)
  41. == f"prfx-cover/switchbot-curtain/{mac_address}/position"
  42. )
  43. @pytest.mark.parametrize(
  44. "mac_address",
  45. ("aa:bb:cc:dd:ee:ff", "aa:bb:cc:dd:ee:gg"),
  46. )
  47. @pytest.mark.parametrize(
  48. ("position", "expected_payload"), [(0, b"0"), (100, b"100"), (42, b"42")]
  49. )
  50. def test__report_position(
  51. caplog: _pytest.logging.LogCaptureFixture,
  52. mac_address: str,
  53. position: int,
  54. expected_payload: bytes,
  55. ) -> None:
  56. with unittest.mock.patch(
  57. "switchbot.SwitchbotCurtain.__init__", return_value=None
  58. ) as device_init_mock, caplog.at_level(logging.DEBUG):
  59. actor = _CurtainMotor(mac_address=mac_address, retry_count=7, password=None)
  60. device_init_mock.assert_called_once_with(
  61. mac=mac_address,
  62. retry_count=7,
  63. password=None,
  64. # > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
  65. # > [...] The parameter 'reverse_mode' reverse these values, [...]
  66. # > The parameter is default set to True so that the definition of position
  67. # > is the same as in Home Assistant.
  68. # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
  69. reverse_mode=True,
  70. )
  71. with unittest.mock.patch.object(
  72. actor, "_mqtt_publish"
  73. ) as publish_mock, unittest.mock.patch(
  74. "switchbot.SwitchbotCurtain.get_position", return_value=position
  75. ):
  76. actor._report_position(mqtt_client="dummy", mqtt_topic_prefix="topic-prefix")
  77. publish_mock.assert_called_once_with(
  78. topic_prefix="topic-prefix",
  79. topic_levels=(
  80. "cover",
  81. "switchbot-curtain",
  82. switchbot_mqtt._utils._MQTTTopicPlaceholder.MAC_ADDRESS,
  83. "position",
  84. ),
  85. payload=expected_payload,
  86. mqtt_client="dummy",
  87. )
  88. assert not caplog.record_tuples
  89. @pytest.mark.parametrize("position", ("", 'lambda: print("")'))
  90. def test__report_position_invalid(
  91. caplog: _pytest.logging.LogCaptureFixture, position: str
  92. ) -> None:
  93. with unittest.mock.patch(
  94. "switchbot.SwitchbotCurtain.__init__", return_value=None
  95. ), caplog.at_level(logging.DEBUG):
  96. actor = _CurtainMotor(
  97. mac_address="aa:bb:cc:dd:ee:ff", retry_count=3, password=None
  98. )
  99. with unittest.mock.patch.object(
  100. actor, "_mqtt_publish"
  101. ) as publish_mock, unittest.mock.patch(
  102. "switchbot.SwitchbotCurtain.get_position", return_value=position
  103. ), pytest.raises(
  104. ValueError
  105. ):
  106. actor._report_position(mqtt_client="dummy", mqtt_topic_prefix="dummy2")
  107. publish_mock.assert_not_called()
  108. @pytest.mark.parametrize("topic_prefix", ["", "homeassistant/"])
  109. @pytest.mark.parametrize(("battery_percent", "battery_percent_encoded"), [(42, b"42")])
  110. @pytest.mark.parametrize("report_position", [True, False])
  111. @pytest.mark.parametrize(("position", "position_encoded"), [(21, b"21")])
  112. def test__update_and_report_device_info(
  113. topic_prefix: str,
  114. report_position: bool,
  115. battery_percent: int,
  116. battery_percent_encoded: bytes,
  117. position: int,
  118. position_encoded: bytes,
  119. ) -> None:
  120. with unittest.mock.patch("switchbot.SwitchbotCurtain.__init__", return_value=None):
  121. actor = _CurtainMotor(mac_address="dummy", retry_count=21, password=None)
  122. actor._get_device()._switchbot_device_data = {
  123. "data": {"battery": battery_percent, "position": position}
  124. }
  125. mqtt_client_mock = unittest.mock.MagicMock()
  126. with unittest.mock.patch("switchbot.SwitchbotCurtain.update") as update_mock:
  127. actor._update_and_report_device_info(
  128. mqtt_client=mqtt_client_mock,
  129. mqtt_topic_prefix=topic_prefix,
  130. report_position=report_position,
  131. )
  132. update_mock.assert_called_once_with()
  133. assert mqtt_client_mock.publish.call_count == (1 + report_position)
  134. assert (
  135. unittest.mock.call(
  136. topic=topic_prefix + "cover/switchbot-curtain/dummy/battery-percentage",
  137. payload=battery_percent_encoded,
  138. retain=True,
  139. )
  140. in mqtt_client_mock.publish.call_args_list
  141. )
  142. if report_position:
  143. assert (
  144. unittest.mock.call(
  145. topic=topic_prefix + "cover/switchbot-curtain/dummy/position",
  146. payload=position_encoded,
  147. retain=True,
  148. )
  149. in mqtt_client_mock.publish.call_args_list
  150. )
  151. @pytest.mark.parametrize(
  152. "exception",
  153. [
  154. PermissionError("bluepy-helper failed to enable low energy mode..."),
  155. bluepy.btle.BTLEManagementError("test"),
  156. ],
  157. )
  158. def test__update_and_report_device_info_update_error(exception: Exception) -> None:
  159. actor = _CurtainMotor(mac_address="dummy", retry_count=21, password=None)
  160. mqtt_client_mock = unittest.mock.MagicMock()
  161. with unittest.mock.patch.object(
  162. actor._get_device(), "update", side_effect=exception
  163. ), pytest.raises(type(exception)):
  164. actor._update_and_report_device_info(
  165. mqtt_client_mock, mqtt_topic_prefix="dummy", report_position=True
  166. )
  167. mqtt_client_mock.publish.assert_not_called()
  168. @pytest.mark.parametrize("topic_prefix", ["topic-prfx"])
  169. @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff", "aa:bb:cc:11:22:33"])
  170. @pytest.mark.parametrize("password", ["pa$$word", None])
  171. @pytest.mark.parametrize("retry_count", (2, 3))
  172. @pytest.mark.parametrize(
  173. ("message_payload", "action_name"),
  174. [
  175. (b"open", "switchbot.SwitchbotCurtain.open"),
  176. (b"OPEN", "switchbot.SwitchbotCurtain.open"),
  177. (b"Open", "switchbot.SwitchbotCurtain.open"),
  178. (b"close", "switchbot.SwitchbotCurtain.close"),
  179. (b"CLOSE", "switchbot.SwitchbotCurtain.close"),
  180. (b"Close", "switchbot.SwitchbotCurtain.close"),
  181. (b"stop", "switchbot.SwitchbotCurtain.stop"),
  182. (b"STOP", "switchbot.SwitchbotCurtain.stop"),
  183. (b"Stop", "switchbot.SwitchbotCurtain.stop"),
  184. ],
  185. )
  186. @pytest.mark.parametrize("update_device_info", [True, False])
  187. @pytest.mark.parametrize("command_successful", [True, False])
  188. def test_execute_command(
  189. caplog: _pytest.logging.LogCaptureFixture,
  190. topic_prefix: str,
  191. mac_address: str,
  192. password: typing.Optional[str],
  193. retry_count: int,
  194. message_payload: bytes,
  195. action_name: str,
  196. update_device_info: bool,
  197. command_successful: bool,
  198. ) -> None:
  199. with unittest.mock.patch(
  200. "switchbot.SwitchbotCurtain.__init__", return_value=None
  201. ) as device_init_mock, caplog.at_level(logging.INFO):
  202. actor = _CurtainMotor(
  203. mac_address=mac_address, retry_count=retry_count, password=password
  204. )
  205. with unittest.mock.patch.object(
  206. actor, "report_state"
  207. ) as report_mock, unittest.mock.patch(
  208. action_name, return_value=command_successful
  209. ) as action_mock, unittest.mock.patch.object(
  210. actor, "_update_and_report_device_info"
  211. ) as update_device_info_mock:
  212. actor.execute_command(
  213. mqtt_client="dummy",
  214. mqtt_message_payload=message_payload,
  215. update_device_info=update_device_info,
  216. mqtt_topic_prefix=topic_prefix,
  217. )
  218. device_init_mock.assert_called_once_with(
  219. mac=mac_address, password=password, retry_count=retry_count, reverse_mode=True
  220. )
  221. action_mock.assert_called_once_with()
  222. if command_successful:
  223. state_str = {b"open": "opening", b"close": "closing", b"stop": "stopped"}[
  224. message_payload.lower()
  225. ]
  226. assert caplog.record_tuples == [
  227. (
  228. "switchbot_mqtt._actors",
  229. logging.INFO,
  230. f"switchbot curtain {mac_address} {state_str}",
  231. )
  232. ]
  233. report_mock.assert_called_once_with(
  234. mqtt_client="dummy",
  235. mqtt_topic_prefix=topic_prefix,
  236. # https://www.home-assistant.io/integrations/cover.mqtt/#state_opening
  237. state={b"open": b"opening", b"close": b"closing", b"stop": b""}[
  238. message_payload.lower()
  239. ],
  240. )
  241. else:
  242. assert caplog.record_tuples == [
  243. (
  244. "switchbot_mqtt._actors",
  245. logging.ERROR,
  246. f"failed to {message_payload.decode().lower()} switchbot curtain {mac_address}",
  247. )
  248. ]
  249. report_mock.assert_not_called()
  250. if update_device_info and command_successful:
  251. update_device_info_mock.assert_called_once_with(
  252. mqtt_client="dummy",
  253. report_position=(action_name == "switchbot.SwitchbotCurtain.stop"),
  254. mqtt_topic_prefix=topic_prefix,
  255. )
  256. else:
  257. update_device_info_mock.assert_not_called()
  258. @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff"])
  259. @pytest.mark.parametrize("password", ["secret"])
  260. @pytest.mark.parametrize("message_payload", [b"OEFFNEN", b""])
  261. def test_execute_command_invalid_payload(
  262. caplog: _pytest.logging.LogCaptureFixture,
  263. mac_address: str,
  264. password: str,
  265. message_payload: bytes,
  266. ) -> None:
  267. with unittest.mock.patch(
  268. "switchbot.SwitchbotCurtain"
  269. ) as device_mock, caplog.at_level(logging.INFO):
  270. actor = _CurtainMotor(mac_address=mac_address, retry_count=7, password=password)
  271. with unittest.mock.patch.object(actor, "report_state") as report_mock:
  272. actor.execute_command(
  273. mqtt_client="dummy",
  274. mqtt_message_payload=message_payload,
  275. update_device_info=True,
  276. mqtt_topic_prefix="dummy",
  277. )
  278. device_mock.assert_called_once_with(
  279. mac=mac_address, password=password, retry_count=7, reverse_mode=True
  280. )
  281. assert not device_mock().mock_calls # no methods called
  282. report_mock.assert_not_called()
  283. assert caplog.record_tuples == [
  284. (
  285. "switchbot_mqtt._actors",
  286. logging.WARNING,
  287. f"unexpected payload {message_payload!r} (expected 'OPEN', 'CLOSE', or 'STOP')",
  288. )
  289. ]
  290. @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff"])
  291. @pytest.mark.parametrize("message_payload", [b"OPEN", b"CLOSE", b"STOP"])
  292. def test_execute_command_bluetooth_error(
  293. caplog: _pytest.logging.LogCaptureFixture, mac_address: str, message_payload: bytes
  294. ) -> None:
  295. """
  296. paho.mqtt.python>=1.5.1 no longer implicitly suppresses exceptions in callbacks.
  297. verify pySwitchbot catches exceptions raised in bluetooth stack.
  298. https://github.com/Danielhiversen/pySwitchbot/blob/0.8.0/switchbot/__init__.py#L48
  299. https://github.com/Danielhiversen/pySwitchbot/blob/0.8.0/switchbot/__init__.py#L94
  300. """
  301. with unittest.mock.patch(
  302. "bluepy.btle.Peripheral",
  303. side_effect=bluepy.btle.BTLEDisconnectError(
  304. f"Failed to connect to peripheral {mac_address}, addr type: random"
  305. ),
  306. ), caplog.at_level(logging.ERROR):
  307. _CurtainMotor(
  308. mac_address=mac_address, retry_count=0, password="secret"
  309. ).execute_command(
  310. mqtt_client="dummy",
  311. mqtt_message_payload=message_payload,
  312. update_device_info=True,
  313. mqtt_topic_prefix="dummy",
  314. )
  315. assert len(caplog.records) == 2
  316. assert caplog.records[0].name == "switchbot"
  317. assert caplog.records[0].levelno == logging.ERROR
  318. assert caplog.records[0].msg.startswith(
  319. # pySwitchbot<0.11 had '.' suffix
  320. "Switchbot communication failed. Stopping trying",
  321. )
  322. assert caplog.record_tuples[1] == (
  323. "switchbot_mqtt._actors",
  324. logging.ERROR,
  325. f"failed to {message_payload.decode().lower()} switchbot curtain {mac_address}",
  326. )