test_switchbot_curtain_motor.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156
  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 unittest.mock
  20. import bluepy.btle
  21. import pytest
  22. import switchbot_mqtt
  23. # pylint: disable=protected-access,
  24. # pylint: disable=too-many-arguments; these are tests, no API
  25. @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff", "aa:bb:cc:11:22:33"])
  26. @pytest.mark.parametrize("retry_count", (2, 3))
  27. @pytest.mark.parametrize(
  28. ("message_payload", "action_name"),
  29. [
  30. (b"open", "switchbot.SwitchbotCurtain.open"),
  31. (b"OPEN", "switchbot.SwitchbotCurtain.open"),
  32. (b"Open", "switchbot.SwitchbotCurtain.open"),
  33. (b"close", "switchbot.SwitchbotCurtain.close"),
  34. (b"CLOSE", "switchbot.SwitchbotCurtain.close"),
  35. (b"Close", "switchbot.SwitchbotCurtain.close"),
  36. (b"stop", "switchbot.SwitchbotCurtain.stop"),
  37. (b"STOP", "switchbot.SwitchbotCurtain.stop"),
  38. (b"Stop", "switchbot.SwitchbotCurtain.stop"),
  39. ],
  40. )
  41. @pytest.mark.parametrize("command_successful", [True, False])
  42. def test_execute_command(
  43. caplog, mac_address, retry_count, message_payload, action_name, command_successful
  44. ):
  45. with unittest.mock.patch(
  46. "switchbot.SwitchbotCurtain.__init__", return_value=None
  47. ) as device_init_mock, caplog.at_level(logging.INFO):
  48. actor = switchbot_mqtt._CurtainMotor(
  49. mac_address=mac_address, retry_count=retry_count
  50. )
  51. with unittest.mock.patch.object(
  52. actor, "report_state"
  53. ) as report_mock, unittest.mock.patch(
  54. action_name, return_value=command_successful
  55. ) as action_mock:
  56. actor.execute_command(
  57. mqtt_client="dummy", mqtt_message_payload=message_payload
  58. )
  59. device_init_mock.assert_called_once_with(mac=mac_address, retry_count=retry_count)
  60. action_mock.assert_called_once_with()
  61. if command_successful:
  62. assert caplog.record_tuples == [
  63. (
  64. "switchbot_mqtt",
  65. logging.INFO,
  66. "switchbot curtain {} {}".format(
  67. mac_address,
  68. {b"open": "opening", b"close": "closing", b"stop": "stopped"}[
  69. message_payload.lower()
  70. ],
  71. ),
  72. )
  73. ]
  74. report_mock.assert_called_once_with(
  75. mqtt_client="dummy",
  76. # https://www.home-assistant.io/integrations/cover.mqtt/#state_opening
  77. state={b"open": b"opening", b"close": b"closing", b"stop": b""}[
  78. message_payload.lower()
  79. ],
  80. )
  81. else:
  82. assert caplog.record_tuples == [
  83. (
  84. "switchbot_mqtt",
  85. logging.ERROR,
  86. "failed to {} switchbot curtain {}".format(
  87. message_payload.decode().lower(), mac_address
  88. ),
  89. )
  90. ]
  91. report_mock.assert_not_called()
  92. @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff"])
  93. @pytest.mark.parametrize("message_payload", [b"OEFFNEN", b""])
  94. def test_execute_command_invalid_payload(caplog, mac_address, message_payload):
  95. with unittest.mock.patch(
  96. "switchbot.SwitchbotCurtain"
  97. ) as device_mock, caplog.at_level(logging.INFO):
  98. actor = switchbot_mqtt._CurtainMotor(mac_address=mac_address, retry_count=7)
  99. with unittest.mock.patch.object(actor, "report_state") as report_mock:
  100. actor.execute_command(
  101. mqtt_client="dummy", mqtt_message_payload=message_payload
  102. )
  103. device_mock.assert_called_once_with(mac=mac_address, retry_count=7)
  104. assert not device_mock().mock_calls # no methods called
  105. report_mock.assert_not_called()
  106. assert caplog.record_tuples == [
  107. (
  108. "switchbot_mqtt",
  109. logging.WARNING,
  110. "unexpected payload {!r} (expected 'OPEN', 'CLOSE', or 'STOP')".format(
  111. message_payload
  112. ),
  113. )
  114. ]
  115. @pytest.mark.parametrize("mac_address", ["aa:bb:cc:dd:ee:ff"])
  116. @pytest.mark.parametrize("message_payload", [b"OPEN", b"CLOSE", b"STOP"])
  117. def test_execute_command_bluetooth_error(caplog, mac_address, message_payload):
  118. """
  119. paho.mqtt.python>=1.5.1 no longer implicitly suppresses exceptions in callbacks.
  120. verify pySwitchbot catches exceptions raised in bluetooth stack.
  121. https://github.com/Danielhiversen/pySwitchbot/blob/0.8.0/switchbot/__init__.py#L48
  122. https://github.com/Danielhiversen/pySwitchbot/blob/0.8.0/switchbot/__init__.py#L94
  123. """
  124. with unittest.mock.patch(
  125. "bluepy.btle.Peripheral",
  126. side_effect=bluepy.btle.BTLEDisconnectError(
  127. "Failed to connect to peripheral {}, addr type: random".format(mac_address)
  128. ),
  129. ), caplog.at_level(logging.ERROR):
  130. switchbot_mqtt._CurtainMotor(
  131. mac_address=mac_address, retry_count=10
  132. ).execute_command(mqtt_client="dummy", mqtt_message_payload=message_payload)
  133. assert caplog.record_tuples == [
  134. (
  135. "switchbot",
  136. logging.ERROR,
  137. "Switchbot communication failed. Stopping trying.",
  138. ),
  139. (
  140. "switchbot_mqtt",
  141. logging.ERROR,
  142. "failed to {} switchbot curtain {}".format(
  143. message_payload.decode().lower(), mac_address
  144. ),
  145. ),
  146. ]