1
0

test_mqtt.py 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201
  1. import logging
  2. import unittest.mock
  3. import pytest
  4. from paho.mqtt.client import MQTT_ERR_QUEUE_SIZE, MQTT_ERR_SUCCESS, MQTTMessage
  5. import switchbot_mqtt
  6. # pylint: disable=protected-access
  7. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  8. @pytest.mark.parametrize("mqtt_port", [1833])
  9. def test__run(mqtt_host, mqtt_port):
  10. with unittest.mock.patch(
  11. "paho.mqtt.client.Client"
  12. ) as mqtt_client_mock, unittest.mock.patch(
  13. "switchbot_mqtt._mqtt_on_message"
  14. ) as message_handler_mock:
  15. switchbot_mqtt._run(
  16. mqtt_host=mqtt_host,
  17. mqtt_port=mqtt_port,
  18. mqtt_username=None,
  19. mqtt_password=None,
  20. )
  21. mqtt_client_mock.assert_called_once_with()
  22. assert not mqtt_client_mock().username_pw_set.called
  23. mqtt_client_mock().connect.assert_called_once_with(host=mqtt_host, port=mqtt_port)
  24. mqtt_client_mock().socket().getpeername.return_value = (mqtt_host, mqtt_port)
  25. mqtt_client_mock().on_connect(mqtt_client_mock(), None, {}, 0)
  26. mqtt_client_mock().subscribe.assert_called_once_with(
  27. "homeassistant/switch/switchbot/+/set"
  28. )
  29. mqtt_client_mock().on_message(mqtt_client_mock(), None, "message")
  30. message_handler_mock.assert_called_once()
  31. mqtt_client_mock().loop_forever.assert_called_once_with()
  32. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  33. @pytest.mark.parametrize("mqtt_port", [1833])
  34. @pytest.mark.parametrize("mqtt_username", ["me"])
  35. @pytest.mark.parametrize("mqtt_password", [None, "secret"])
  36. def test__run_authentication(mqtt_host, mqtt_port, mqtt_username, mqtt_password):
  37. with unittest.mock.patch("paho.mqtt.client.Client") as mqtt_client_mock:
  38. switchbot_mqtt._run(
  39. mqtt_host=mqtt_host,
  40. mqtt_port=mqtt_port,
  41. mqtt_username=mqtt_username,
  42. mqtt_password=mqtt_password,
  43. )
  44. mqtt_client_mock.assert_called_once_with()
  45. mqtt_client_mock().username_pw_set.assert_called_once_with(
  46. username=mqtt_username, password=mqtt_password,
  47. )
  48. @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
  49. @pytest.mark.parametrize("mqtt_port", [1833])
  50. @pytest.mark.parametrize("mqtt_password", ["secret"])
  51. def test__run_authentication_missing_username(mqtt_host, mqtt_port, mqtt_password):
  52. with unittest.mock.patch("paho.mqtt.client.Client"):
  53. with pytest.raises(ValueError):
  54. switchbot_mqtt._run(
  55. mqtt_host=mqtt_host,
  56. mqtt_port=mqtt_port,
  57. mqtt_username=None,
  58. mqtt_password=mqtt_password,
  59. )
  60. @pytest.mark.parametrize(
  61. ("topic", "payload", "expected_mac_address", "expected_action"),
  62. [
  63. (
  64. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  65. b"ON",
  66. "aa:bb:cc:dd:ee:ff",
  67. switchbot_mqtt._SwitchbotAction.ON,
  68. ),
  69. (
  70. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  71. b"OFF",
  72. "aa:bb:cc:dd:ee:ff",
  73. switchbot_mqtt._SwitchbotAction.OFF,
  74. ),
  75. (
  76. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  77. b"on",
  78. "aa:bb:cc:dd:ee:ff",
  79. switchbot_mqtt._SwitchbotAction.ON,
  80. ),
  81. (
  82. b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set",
  83. b"off",
  84. "aa:bb:cc:dd:ee:ff",
  85. switchbot_mqtt._SwitchbotAction.OFF,
  86. ),
  87. (
  88. b"homeassistant/switch/switchbot/aa:01:23:45:67:89/set",
  89. b"ON",
  90. "aa:01:23:45:67:89",
  91. switchbot_mqtt._SwitchbotAction.ON,
  92. ),
  93. ],
  94. )
  95. def test__mqtt_on_message(
  96. topic: bytes,
  97. payload: bytes,
  98. expected_mac_address: str,
  99. expected_action: switchbot_mqtt._SwitchbotAction,
  100. ):
  101. message = MQTTMessage(topic=topic)
  102. message.payload = payload
  103. with unittest.mock.patch("switchbot_mqtt._send_command") as send_command_mock:
  104. switchbot_mqtt._mqtt_on_message("client_dummy", None, message)
  105. send_command_mock.assert_called_once_with(
  106. mqtt_client="client_dummy",
  107. switchbot_mac_address=expected_mac_address,
  108. action=expected_action,
  109. )
  110. @pytest.mark.parametrize(
  111. ("topic", "payload"),
  112. [
  113. (b"homeassistant/switch/switchbot/aa:01:23:4E:RR:OR/set", b"ON"),
  114. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff", b"on"),
  115. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/change", b"ON"),
  116. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set", b""),
  117. (b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set", b"EIN"),
  118. ],
  119. )
  120. def test__mqtt_on_message_ignored(
  121. topic: bytes, payload: bytes,
  122. ):
  123. message = MQTTMessage(topic=topic)
  124. message.payload = payload
  125. with unittest.mock.patch("switchbot_mqtt._send_command") as send_command_mock:
  126. switchbot_mqtt._mqtt_on_message(None, None, message)
  127. assert not send_command_mock.called
  128. @pytest.mark.parametrize(
  129. ("topic", "payload"),
  130. [(b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/set", b"ON")],
  131. )
  132. def test__mqtt_on_message_ignored_retained(
  133. topic: bytes, payload: bytes,
  134. ):
  135. message = MQTTMessage(topic=topic)
  136. message.payload = payload
  137. message.retain = True
  138. with unittest.mock.patch("switchbot_mqtt._send_command") as send_command_mock:
  139. switchbot_mqtt._mqtt_on_message(None, None, message)
  140. assert not send_command_mock.called
  141. @pytest.mark.parametrize(
  142. ("switchbot_mac_address", "expected_topic"),
  143. # https://www.home-assistant.io/docs/mqtt/discovery/#switches
  144. [("aa:bb:cc:dd:ee:ff", "homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/state")],
  145. )
  146. @pytest.mark.parametrize(
  147. ("state", "expected_payload"),
  148. [
  149. (switchbot_mqtt._SwitchbotState.ON, b"ON"),
  150. (switchbot_mqtt._SwitchbotState.OFF, b"OFF"),
  151. ],
  152. )
  153. @pytest.mark.parametrize(
  154. "return_code", [MQTT_ERR_SUCCESS, MQTT_ERR_QUEUE_SIZE],
  155. )
  156. def test__report_state(
  157. caplog,
  158. state: switchbot_mqtt._SwitchbotState,
  159. switchbot_mac_address: str,
  160. expected_topic: str,
  161. expected_payload: bytes,
  162. return_code: int,
  163. ):
  164. # pylint: disable=too-many-arguments
  165. mqtt_client_mock = unittest.mock.MagicMock()
  166. mqtt_client_mock.publish.return_value.rc = return_code
  167. with caplog.at_level(logging.WARNING):
  168. switchbot_mqtt._report_state(
  169. mqtt_client=mqtt_client_mock,
  170. switchbot_mac_address=switchbot_mac_address,
  171. switchbot_state=state,
  172. )
  173. mqtt_client_mock.publish.assert_called_once_with(
  174. topic=expected_topic, payload=expected_payload, retain=True,
  175. )
  176. if return_code == MQTT_ERR_SUCCESS:
  177. assert len(caplog.records) == 0
  178. else:
  179. assert len(caplog.records) == 1
  180. assert caplog.record_tuples[0] == (
  181. "switchbot_mqtt",
  182. logging.ERROR,
  183. "failed to publish state (rc={})".format(return_code),
  184. )