1
0

test_switchbot_curtain_motor.py 5.8 KB

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