refactor: split `_parse_mqtt_topic` from `_MQTTControlledActor._mqtt_command_callback`

switchbot_mqtt/_actors/_base.py

@@ -30,6 +30,7 @@ from switchbot_mqtt._utils import (
     _mac_address_valid,
     _MQTTTopicLevel,
     _MQTTTopicPlaceholder,
+    _parse_mqtt_topic,
     _QueueLogHandler,
 )
 
@@ -151,20 +152,13 @@ class _MQTTControlledActor(abc.ABC):
         if message.retain:
             _LOGGER.info("ignoring retained message")
             return
-        topic_split = message.topic.split("/")
-        if len(topic_split) != len(cls.MQTT_COMMAND_TOPIC_LEVELS):
-            _LOGGER.warning("unexpected topic %s", message.topic)
+        try:
+            mac_address = _parse_mqtt_topic(
+                topic=message.topic, expected_levels=cls.MQTT_COMMAND_TOPIC_LEVELS
+            )[_MQTTTopicPlaceholder.MAC_ADDRESS]
+        except ValueError as exc:
+            _LOGGER.warning(str(exc), exc_info=False)
             return
-        mac_address = None
-        for given_part, expected_part in zip(
-            topic_split, cls.MQTT_COMMAND_TOPIC_LEVELS
-        ):
-            if expected_part == _MQTTTopicPlaceholder.MAC_ADDRESS:
-                mac_address = given_part
-            elif expected_part != given_part:
-                _LOGGER.warning("unexpected topic %s", message.topic)
-                return
-        assert mac_address
         if not _mac_address_valid(mac_address):
             _LOGGER.warning("invalid mac address %s", mac_address)
             return

switchbot_mqtt/_utils.py

@@ -45,6 +45,21 @@ def _join_mqtt_topic_levels(
     )
 
 
+def _parse_mqtt_topic(
+    topic: str, expected_levels: typing.List[_MQTTTopicLevel]
+) -> typing.Dict[_MQTTTopicPlaceholder, str]:
+    attrs: typing.Dict[_MQTTTopicPlaceholder, str] = {}
+    topic_split = topic.split("/")
+    if len(topic_split) != len(expected_levels):
+        raise ValueError(f"unexpected topic {topic}")
+    for given_part, expected_part in zip(topic_split, expected_levels):
+        if expected_part == _MQTTTopicPlaceholder.MAC_ADDRESS:
+            attrs[_MQTTTopicPlaceholder(expected_part)] = given_part
+        elif expected_part != given_part:
+            raise ValueError(f"unexpected topic {topic}")
+    return attrs
+
+
 class _QueueLogHandler(logging.Handler):
     """
     logging.handlers.QueueHandler drops exc_info

tests/test_mac_address.py

@@ -1,35 +0,0 @@
-# switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
-# compatible with home-assistant.io's MQTT Switch & Cover platform
-#
-# Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-import pytest
-import switchbot_mqtt._utils
-
-
-@pytest.mark.parametrize(
-    ("mac_address", "valid"),
-    [
-        ("aa:bb:cc:dd:ee:ff", True),
-        ("AA:BB:CC:DD:EE:FF", True),
-        ("AA:12:34:45:67:89", True),
-        ("aabbccddeeff", False),  # not supported by PySwitchbot
-        ("aa:bb:cc:dd:ee:gg", False),
-    ],
-)
-def test__mac_address_valid(mac_address, valid):
-    # pylint: disable=protected-access
-    assert switchbot_mqtt._utils._mac_address_valid(mac_address) == valid

tests/test_utils.py

@@ -0,0 +1,94 @@
+# switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
+# compatible with home-assistant.io's MQTT Switch & Cover platform
+#
+# Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import pytest
+
+from switchbot_mqtt._utils import (
+    _mac_address_valid,
+    _MQTTTopicPlaceholder,
+    _parse_mqtt_topic,
+)
+
+
+@pytest.mark.parametrize(
+    ("mac_address", "valid"),
+    [
+        ("aa:bb:cc:dd:ee:ff", True),
+        ("AA:BB:CC:DD:EE:FF", True),
+        ("AA:12:34:45:67:89", True),
+        ("aabbccddeeff", False),  # not supported by PySwitchbot
+        ("aa:bb:cc:dd:ee:gg", False),
+    ],
+)
+def test__mac_address_valid(mac_address, valid):
+    # pylint: disable=protected-access
+    assert _mac_address_valid(mac_address) == valid
+
+
+@pytest.mark.parametrize(
+    ("expected_levels", "topic", "expected_attrs"),
+    [
+        (
+            ["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS, "set"],
+            "switchbot/aa:bb:cc:dd:ee:ff/set",
+            {_MQTTTopicPlaceholder.MAC_ADDRESS: "aa:bb:cc:dd:ee:ff"},
+        ),
+        (
+            ["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS, "set"],
+            "switchbot//set",
+            {_MQTTTopicPlaceholder.MAC_ADDRESS: ""},
+        ),
+        (
+            ["prefix", _MQTTTopicPlaceholder.MAC_ADDRESS],
+            "prefix/aa:bb:cc:dd:ee:ff",
+            {_MQTTTopicPlaceholder.MAC_ADDRESS: "aa:bb:cc:dd:ee:ff"},
+        ),
+        (
+            [_MQTTTopicPlaceholder.MAC_ADDRESS],
+            "00:11:22:33:44:55",
+            {_MQTTTopicPlaceholder.MAC_ADDRESS: "00:11:22:33:44:55"},
+        ),
+    ],
+)
+def test__parse_mqtt_topic(expected_levels, topic, expected_attrs):
+    assert (
+        _parse_mqtt_topic(topic=topic, expected_levels=expected_levels)
+        == expected_attrs
+    )
+
+
+@pytest.mark.parametrize(
+    ("expected_levels", "topic"),
+    [
+        (
+            ["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS, "set"],
+            "switchbot/aa:bb:cc:dd:ee:ff",
+        ),
+        (
+            ["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS, "set"],
+            "switchbot/aa:bb:cc:dd:ee:ff/change",
+        ),
+        (
+            ["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS, "set"],
+            "switchbot/aa:bb:cc:dd:ee:ff/set/suffix",
+        ),
+    ],
+)
+def test__parse_mqtt_topic_fail(expected_levels, topic):
+    with pytest.raises(ValueError):
+        _parse_mqtt_topic(topic=topic, expected_levels=expected_levels)