test_mqtt.py 6.9 KB

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