refactor: split _run from _main; added tests

switchbot_mqtt/__init__.py

@@ -116,6 +116,18 @@ def _mqtt_on_message(
     _send_command(switchbot_mac_address=switchbot_mac_address, action=action)
 
 
+def _run(mqtt_host: str, mqtt_port: int) -> None:
+    # https://pypi.org/project/paho-mqtt/
+    mqtt_client = paho.mqtt.client.Client()
+    mqtt_client.on_connect = _mqtt_on_connect
+    mqtt_client.on_message = _mqtt_on_message
+    _LOGGER.info(
+        "connecting to MQTT broker %s:%d", mqtt_host, mqtt_port,
+    )
+    mqtt_client.connect(host=mqtt_host, port=mqtt_port)
+    mqtt_client.loop_forever()
+
+
 def _main() -> None:
     logging.basicConfig(
         level=logging.DEBUG,
@@ -129,12 +141,4 @@ def _main() -> None:
     argparser.add_argument("--mqtt-host", type=str, required=True)
     argparser.add_argument("--mqtt-port", type=int, default=1883)
     args = argparser.parse_args()
-    # https://pypi.org/project/paho-mqtt/
-    mqtt_client = paho.mqtt.client.Client()
-    mqtt_client.on_connect = _mqtt_on_connect
-    mqtt_client.on_message = _mqtt_on_message
-    _LOGGER.info(
-        "connecting to MQTT broker %s:%d", args.mqtt_host, args.mqtt_port,
-    )
-    mqtt_client.connect(host=args.mqtt_host, port=args.mqtt_port)
-    mqtt_client.loop_forever()
+    _run(mqtt_host=args.mqtt_host, mqtt_port=args.mqtt_port)

tests/test_mqtt.py

@@ -0,0 +1,27 @@
+import unittest.mock
+
+import pytest
+
+import switchbot_mqtt
+
+
+@pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
+@pytest.mark.parametrize("mqtt_port", [1833])
+def test__run(mqtt_host, mqtt_port):
+    with unittest.mock.patch(
+        "paho.mqtt.client.Client"
+    ) as mqtt_client_mock, unittest.mock.patch(
+        "switchbot_mqtt._mqtt_on_message"
+    ) as message_handler_mock:
+        # pylint: disable=protected-access
+        switchbot_mqtt._run(mqtt_host=mqtt_host, mqtt_port=mqtt_port)
+    mqtt_client_mock.assert_called_once_with()
+    mqtt_client_mock().connect.assert_called_once_with(host=mqtt_host, port=mqtt_port)
+    mqtt_client_mock().socket().getpeername.return_value = (mqtt_host, mqtt_port)
+    mqtt_client_mock().on_connect(mqtt_client_mock(), None, {}, 0)
+    mqtt_client_mock().subscribe.assert_called_once_with(
+        "homeassistant/switch/switchbot/+/set"
+    )
+    mqtt_client_mock().on_message(mqtt_client_mock(), None, "message")
+    message_handler_mock.assert_called_once()
+    mqtt_client_mock().loop_forever.assert_called_once_with()