refactor: store topic levels in tuples instead of lists (see `_MQTTControlledActor._get_mqtt_message_callbacks`)

switchbot_mqtt/_actors/__init__.py

@@ -33,37 +33,37 @@ from switchbot_mqtt._utils import (
 _LOGGER = logging.getLogger(__name__)
 
 # "homeassistant" for historic reason, may be parametrized in future
-_TOPIC_LEVELS_PREFIX: typing.List[_MQTTTopicLevel] = ["homeassistant"]
-_BUTTON_TOPIC_LEVELS_PREFIX = _TOPIC_LEVELS_PREFIX + [
+_TOPIC_LEVELS_PREFIX: typing.Tuple[_MQTTTopicLevel] = ("homeassistant",)
+_BUTTON_TOPIC_LEVELS_PREFIX = _TOPIC_LEVELS_PREFIX + (
     "switch",
     "switchbot",
     _MQTTTopicPlaceholder.MAC_ADDRESS,
-]
-_CURTAIN_TOPIC_LEVELS_PREFIX = _TOPIC_LEVELS_PREFIX + [
+)
+_CURTAIN_TOPIC_LEVELS_PREFIX = _TOPIC_LEVELS_PREFIX + (
     "cover",
     "switchbot-curtain",
     _MQTTTopicPlaceholder.MAC_ADDRESS,
-]
+)
 
 
 class _ButtonAutomator(_MQTTControlledActor):
     # https://www.home-assistant.io/integrations/switch.mqtt/
 
-    MQTT_COMMAND_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ["set"]
-    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + [
-        "request-device-info"
-    ]
-    MQTT_STATE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ["state"]
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + [
-        "battery-percentage"
-    ]
+    MQTT_COMMAND_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ("set",)
+    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + (
+        "request-device-info",
+    )
+    MQTT_STATE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ("state",)
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + (
+        "battery-percentage",
+    )
     # for downward compatibility (will be removed in v3):
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS_LEGACY = _TOPIC_LEVELS_PREFIX + [
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS_LEGACY = _TOPIC_LEVELS_PREFIX + (
         "cover",
         "switchbot",
         _MQTTTopicPlaceholder.MAC_ADDRESS,
         "battery-percentage",
-    ]
+    )
 
     def __init__(
         self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
@@ -121,19 +121,19 @@ class _ButtonAutomator(_MQTTControlledActor):
 class _CurtainMotor(_MQTTControlledActor):
 
     # https://www.home-assistant.io/integrations/cover.mqtt/
-    MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ["set"]
+    MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("set",)
     _MQTT_SET_POSITION_TOPIC_LEVELS = tuple(_CURTAIN_TOPIC_LEVELS_PREFIX) + (
         "position",
         "set-percent",
     )
-    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + [
-        "request-device-info"
-    ]
-    MQTT_STATE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ["state"]
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + [
-        "battery-percentage"
-    ]
-    _MQTT_POSITION_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ["position"]
+    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (
+        "request-device-info",
+    )
+    MQTT_STATE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("state",)
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (
+        "battery-percentage",
+    )
+    _MQTT_POSITION_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("position",)
 
     @classmethod
     def get_mqtt_position_topic(cls, mac_address: str) -> str:

switchbot_mqtt/_actors/_base.py

@@ -48,10 +48,14 @@ class _MQTTCallbackUserdata:
 
 
 class _MQTTControlledActor(abc.ABC):
-    MQTT_COMMAND_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
-    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
-    MQTT_STATE_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.List[_MQTTTopicLevel] = NotImplemented
+    MQTT_COMMAND_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented
+    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS: typing.Tuple[
+        _MQTTTopicLevel, ...
+    ] = NotImplemented
+    MQTT_STATE_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.Tuple[
+        _MQTTTopicLevel, ...
+    ] = NotImplemented
 
     @classmethod
     def get_mqtt_update_device_info_topic(cls, mac_address: str) -> str:
@@ -221,10 +225,10 @@ class _MQTTControlledActor(abc.ABC):
         # callbacks with same topic pattern
         # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L2304
         # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/matcher.py#L19
-        callbacks = {tuple(cls.MQTT_COMMAND_TOPIC_LEVELS): cls._mqtt_command_callback}
+        callbacks = {cls.MQTT_COMMAND_TOPIC_LEVELS: cls._mqtt_command_callback}
         if enable_device_info_update_topic:
             callbacks[
-                tuple(cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS)
+                cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS
             ] = cls._mqtt_update_device_info_callback
         return callbacks
 
@@ -246,7 +250,7 @@ class _MQTTControlledActor(abc.ABC):
     def _mqtt_publish(
         self,
         *,
-        topic_levels: typing.List[_MQTTTopicLevel],
+        topic_levels: typing.Iterable[_MQTTTopicLevel],
         payload: bytes,
         mqtt_client: paho.mqtt.client.Client,
     ) -> None:

tests/test_mqtt.py

@@ -176,8 +176,8 @@ def test__run_authentication_missing_username(
 
 def _mock_actor_class(
     *,
-    command_topic_levels: typing.List[_MQTTTopicLevel] = NotImplemented,
-    request_info_levels: typing.List[_MQTTTopicLevel] = NotImplemented,
+    command_topic_levels: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented,
+    request_info_levels: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented,
 ) -> typing.Type:
     class _ActorMock(_MQTTControlledActor):
         MQTT_COMMAND_TOPIC_LEVELS = command_topic_levels
@@ -217,7 +217,7 @@ def _mock_actor_class(
 @pytest.mark.parametrize("payload", [b"", b"whatever"])
 def test__mqtt_update_device_info_callback(
     caplog: _pytest.logging.LogCaptureFixture,
-    topic_levels: typing.List[_MQTTTopicLevel],
+    topic_levels: typing.Tuple[_MQTTTopicLevel, ...],
     topic: bytes,
     expected_mac_address: str,
     payload: bytes,
@@ -257,7 +257,7 @@ def test__mqtt_update_device_info_callback_ignore_retained(
     caplog: _pytest.logging.LogCaptureFixture,
 ) -> None:
     ActorMock = _mock_actor_class(
-        request_info_levels=[_MQTTTopicPlaceholder.MAC_ADDRESS, "request"]
+        request_info_levels=(_MQTTTopicPlaceholder.MAC_ADDRESS, "request")
     )
     message = MQTTMessage(topic=b"aa:bb:cc:dd:ee:ff/request")
     message.payload = b""
@@ -339,7 +339,7 @@ def test__mqtt_update_device_info_callback_ignore_retained(
 @pytest.mark.parametrize("fetch_device_info", [True, False])
 def test__mqtt_command_callback(
     caplog: _pytest.logging.LogCaptureFixture,
-    command_topic_levels: typing.List[_MQTTTopicLevel],
+    command_topic_levels: typing.Tuple[_MQTTTopicLevel, ...],
     topic: bytes,
     payload: bytes,
     expected_mac_address: str,
@@ -391,7 +391,7 @@ def test__mqtt_command_callback_password(
     mac_address: str, expected_password: typing.Optional[str]
 ) -> None:
     ActorMock = _mock_actor_class(
-        command_topic_levels=["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS]
+        command_topic_levels=("switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS)
     )
     message = MQTTMessage(topic=b"switchbot/" + mac_address.encode())
     message.payload = b"whatever"
@@ -565,7 +565,7 @@ def test__mqtt_command_callback_ignore_retained(
 @pytest.mark.parametrize("return_code", [MQTT_ERR_SUCCESS, MQTT_ERR_QUEUE_SIZE])
 def test__report_state(
     caplog: _pytest.logging.LogCaptureFixture,
-    state_topic_levels: typing.List[_MQTTTopicLevel],
+    state_topic_levels: typing.Tuple[_MQTTTopicLevel, ...],
     mac_address: str,
     expected_topic: str,
     state: bytes,

tests/test_switchbot_curtain_motor.py

@@ -82,13 +82,13 @@ def test__report_position(
     ):
         actor._report_position(mqtt_client="dummy")
     publish_mock.assert_called_once_with(
-        topic_levels=[
+        topic_levels=(
             "homeassistant",
             "cover",
             "switchbot-curtain",
             switchbot_mqtt._utils._MQTTTopicPlaceholder.MAC_ADDRESS,
             "position",
-        ],
+        ),
         payload=expected_payload,
         mqtt_client="dummy",
     )