1
0

test_switchbot_button_automator.py 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179
  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. # pylint: disable=protected-access
  19. # pylint: disable=too-many-arguments; these are tests, no API
  20. # pylint: disable=duplicate-code; similarities with tests for curtain motor
  21. import logging
  22. import typing
  23. import unittest.mock
  24. import _pytest.logging # pylint: disable=import-private-name; typing
  25. import pytest
  26. # pylint: disable=import-private-name; internal
  27. from switchbot_mqtt._actors import _ButtonAutomator
  28. @pytest.mark.parametrize("prefix", ["homeassistant/", "prefix-", ""])
  29. @pytest.mark.parametrize("mac_address", ["{MAC_ADDRESS}", "aa:bb:cc:dd:ee:ff"])
  30. def test_get_mqtt_battery_percentage_topic(prefix: str, mac_address: str) -> None:
  31. assert (
  32. _ButtonAutomator.get_mqtt_battery_percentage_topic(
  33. prefix=prefix, mac_address=mac_address
  34. )
  35. == f"{prefix}switch/switchbot/{mac_address}/battery-percentage"
  36. )
  37. @pytest.mark.asyncio
  38. @pytest.mark.parametrize("topic_prefix", ["homeassistant/", "prefix-", ""])
  39. @pytest.mark.parametrize(("battery_percent", "battery_percent_encoded"), [(42, b"42")])
  40. async def test__update_and_report_device_info(
  41. topic_prefix: str, battery_percent: int, battery_percent_encoded: bytes
  42. ) -> None:
  43. device = unittest.mock.Mock()
  44. device.address = "dummy"
  45. with unittest.mock.patch("switchbot.Switchbot.__init__", return_value=None):
  46. actor = _ButtonAutomator(device=device, retry_count=21, password=None)
  47. actor._get_device().get_basic_info = unittest.mock.AsyncMock(
  48. return_value={"battery": battery_percent}
  49. )
  50. mqtt_client_mock = unittest.mock.AsyncMock()
  51. await actor._update_and_report_device_info(
  52. mqtt_client=mqtt_client_mock, mqtt_topic_prefix=topic_prefix
  53. )
  54. mqtt_client_mock.publish.assert_awaited_once_with(
  55. topic=f"{topic_prefix}switch/switchbot/dummy/battery-percentage",
  56. payload=battery_percent_encoded,
  57. retain=True,
  58. )
  59. @pytest.mark.asyncio
  60. @pytest.mark.parametrize("topic_prefix", ["homeassistant/"])
  61. @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff", "aa:bb:cc:11:22:33"])
  62. @pytest.mark.parametrize("password", (None, "secret"))
  63. @pytest.mark.parametrize("retry_count", (3, 21))
  64. @pytest.mark.parametrize(
  65. ("message_payload", "action_name"),
  66. [
  67. (b"on", "switchbot.Switchbot.turn_on"),
  68. (b"ON", "switchbot.Switchbot.turn_on"),
  69. (b"On", "switchbot.Switchbot.turn_on"),
  70. (b"off", "switchbot.Switchbot.turn_off"),
  71. (b"OFF", "switchbot.Switchbot.turn_off"),
  72. (b"Off", "switchbot.Switchbot.turn_off"),
  73. ],
  74. )
  75. @pytest.mark.parametrize("update_device_info", [True, False])
  76. @pytest.mark.parametrize("command_successful", [True, False])
  77. async def test_execute_command(
  78. caplog: _pytest.logging.LogCaptureFixture,
  79. topic_prefix: str,
  80. mac_address: str,
  81. password: typing.Optional[str],
  82. retry_count: int,
  83. message_payload: bytes,
  84. action_name: str,
  85. update_device_info: bool,
  86. command_successful: bool,
  87. ) -> None:
  88. # pylint: disable=too-many-locals
  89. device = unittest.mock.Mock()
  90. device.address = mac_address
  91. with unittest.mock.patch(
  92. "switchbot.Switchbot.__init__", return_value=None
  93. ) as device_init_mock, caplog.at_level(logging.INFO):
  94. actor = _ButtonAutomator(
  95. device=device, retry_count=retry_count, password=password
  96. )
  97. mqtt_client = unittest.mock.Mock()
  98. with unittest.mock.patch.object(
  99. actor, "report_state"
  100. ) as report_mock, unittest.mock.patch(
  101. action_name, return_value=command_successful
  102. ) as action_mock, unittest.mock.patch.object(
  103. actor, "_update_and_report_device_info"
  104. ) as update_device_info_mock:
  105. await actor.execute_command(
  106. mqtt_client=mqtt_client,
  107. mqtt_message_payload=message_payload,
  108. update_device_info=update_device_info,
  109. mqtt_topic_prefix=topic_prefix,
  110. )
  111. device_init_mock.assert_called_once_with(
  112. device=device, password=password, retry_count=retry_count
  113. )
  114. action_mock.assert_awaited_once_with()
  115. if command_successful:
  116. assert caplog.record_tuples == [
  117. (
  118. "switchbot_mqtt._actors",
  119. logging.INFO,
  120. f"switchbot {mac_address} turned {message_payload.decode().lower()}",
  121. )
  122. ]
  123. report_mock.assert_awaited_once_with(
  124. mqtt_client=mqtt_client,
  125. mqtt_topic_prefix=topic_prefix,
  126. state=message_payload.upper(),
  127. )
  128. assert update_device_info_mock.await_count == (1 if update_device_info else 0)
  129. else:
  130. assert caplog.record_tuples == [
  131. (
  132. "switchbot_mqtt._actors",
  133. logging.ERROR,
  134. f"failed to turn {message_payload.decode().lower()} switchbot {mac_address}",
  135. )
  136. ]
  137. report_mock.assert_not_called()
  138. update_device_info_mock.assert_not_called()
  139. @pytest.mark.asyncio
  140. @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff"])
  141. @pytest.mark.parametrize("message_payload", [b"EIN", b""])
  142. async def test_execute_command_invalid_payload(
  143. caplog: _pytest.logging.LogCaptureFixture, mac_address: str, message_payload: bytes
  144. ) -> None:
  145. device = unittest.mock.Mock()
  146. device.address = mac_address
  147. with unittest.mock.patch("switchbot.Switchbot") as device_mock, caplog.at_level(
  148. logging.INFO
  149. ):
  150. actor = _ButtonAutomator(device=device, retry_count=21, password=None)
  151. with unittest.mock.patch.object(actor, "report_state") as report_mock:
  152. await actor.execute_command(
  153. mqtt_client=unittest.mock.Mock(),
  154. mqtt_message_payload=message_payload,
  155. update_device_info=True,
  156. mqtt_topic_prefix="dummy",
  157. )
  158. device_mock.assert_called_once_with(device=device, retry_count=21, password=None)
  159. assert not device_mock().mock_calls # no methods called
  160. report_mock.assert_not_called()
  161. assert caplog.record_tuples == [
  162. (
  163. "switchbot_mqtt._actors",
  164. logging.WARNING,
  165. f"unexpected payload {message_payload!r} (expected 'ON' or 'OFF')",
  166. )
  167. ]