update & report device info when receiving msg on `homeassistant/{switch/switchbot,cover/switchbot-curtain}/MAC_ADDRESS/request-device-info` (requires `--fetch-device-info`)

CHANGELOG.md

@@ -5,6 +5,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added
+- MQTT messages on topic `homeassistant/switch/switchbot/MAC_ADDRESS/request-device-info`
+  and `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/request-device-info` trigger
+  update and reporting of device information (battery level, and curtains' position).
+  Requires `--fetch-device-info`.
 
 ## [2.1.0] - 2021-10-19
 ### Added

README.md

@@ -40,6 +40,8 @@ $ mosquitto_pub -h MQTT_BROKER -t homeassistant/switch/switchbot/aa:bb:cc:dd:ee:
 
 The command-line option `--fetch-device-info` enables battery level reports on topic
 `homeassistant/switch/switchbot/MAC_ADDRESS/battery-percentage` after every command.
+The report may be requested manually by sending a MQTT message to the topic
+`homeassistant/switch/switchbot/MAC_ADDRESS/request-device-info` (requires `--fetch-device-info`)
 
 ### Curtain Motor
 
@@ -53,6 +55,8 @@ The command-line option `--fetch-device-info` enables position reports on topic
 `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/position` after `STOP` commands
 and battery level reports on topic `homeassistant/cover/switchbot-curtain/MAC_ADDRESS/battery-percentage`
 after every command.
+These reports may be requested manually by sending a MQTT message to the topic
+`homeassistant/cover/switchbot-curtain/MAC_ADDRESS/request-device-info` (requires `--fetch-device-info`)
 
 ### Device Passwords
 

switchbot_mqtt/__init__.py

@@ -38,8 +38,14 @@ def _mqtt_on_connect(
     assert return_code == 0, return_code  # connection accepted
     mqtt_broker_host, mqtt_broker_port = mqtt_client.socket().getpeername()
     _LOGGER.debug("connected to MQTT broker %s:%d", mqtt_broker_host, mqtt_broker_port)
-    _ButtonAutomator.mqtt_subscribe(mqtt_client=mqtt_client)
-    _CurtainMotor.mqtt_subscribe(mqtt_client=mqtt_client)
+    _ButtonAutomator.mqtt_subscribe(
+        mqtt_client=mqtt_client,
+        enable_device_info_update_topic=userdata.fetch_device_info,
+    )
+    _CurtainMotor.mqtt_subscribe(
+        mqtt_client=mqtt_client,
+        enable_device_info_update_topic=userdata.fetch_device_info,
+    )
 
 
 def _run(

switchbot_mqtt/_actors/__init__.py

@@ -33,32 +33,32 @@ from switchbot_mqtt._utils import (
 _LOGGER = logging.getLogger(__name__)
 
 # "homeassistant" for historic reason, may be parametrized in future
-_MQTT_TOPIC_LEVELS_PREFIX: typing.List[_MQTTTopicLevel] = ["homeassistant"]
+_TOPIC_LEVELS_PREFIX: typing.List[_MQTTTopicLevel] = ["homeassistant"]
+_BUTTON_TOPIC_LEVELS_PREFIX = _TOPIC_LEVELS_PREFIX + [
+    "switch",
+    "switchbot",
+    _MQTTTopicPlaceholder.MAC_ADDRESS,
+]
+_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 = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "switch",
-        "switchbot",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "set",
-    ]
-    MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "switch",
-        "switchbot",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "state",
+    MQTT_COMMAND_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ["set"]
+    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + [
+        "request-device-info"
     ]
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "switch",
-        "switchbot",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "battery-percentage",
+    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 = _MQTT_TOPIC_LEVELS_PREFIX + [
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS_LEGACY = _TOPIC_LEVELS_PREFIX + [
         "cover",
         "switchbot",
         _MQTTTopicPlaceholder.MAC_ADDRESS,
@@ -121,30 +121,15 @@ class _ButtonAutomator(_MQTTControlledActor):
 class _CurtainMotor(_MQTTControlledActor):
     # https://www.home-assistant.io/integrations/cover.mqtt/
 
-    MQTT_COMMAND_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "cover",
-        "switchbot-curtain",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "set",
+    MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ["set"]
+    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + [
+        "request-device-info"
     ]
-    MQTT_STATE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "cover",
-        "switchbot-curtain",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "state",
-    ]
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "cover",
-        "switchbot-curtain",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "battery-percentage",
-    ]
-    _MQTT_POSITION_TOPIC_LEVELS = _MQTT_TOPIC_LEVELS_PREFIX + [
-        "cover",
-        "switchbot-curtain",
-        _MQTTTopicPlaceholder.MAC_ADDRESS,
-        "position",
+    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

@@ -56,9 +56,17 @@ 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
 
+    @classmethod
+    def get_mqtt_update_device_info_topic(cls, mac_address: str) -> str:
+        return _join_mqtt_topic_levels(
+            topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
+            mac_address=mac_address,
+        )
+
     @classmethod
     def get_mqtt_battery_percentage_topic(cls, mac_address: str) -> str:
         return _join_mqtt_topic_levels(
@@ -130,6 +138,51 @@ class _MQTTControlledActor(abc.ABC):
         self._update_device_info()
         self._report_battery_level(mqtt_client=mqtt_client)
 
+    @classmethod
+    def _init_from_topic(
+        cls,
+        userdata: _MQTTCallbackUserdata,
+        topic: str,
+        expected_topic_levels: typing.List[_MQTTTopicLevel],
+    ) -> typing.Optional["_MQTTControlledActor"]:
+        try:
+            mac_address = _parse_mqtt_topic(
+                topic=topic, expected_levels=expected_topic_levels
+            )[_MQTTTopicPlaceholder.MAC_ADDRESS]
+        except ValueError as exc:
+            _LOGGER.warning(str(exc), exc_info=False)
+            return None
+        if not _mac_address_valid(mac_address):
+            _LOGGER.warning("invalid mac address %s", mac_address)
+            return None
+        return cls(
+            mac_address=mac_address,
+            retry_count=userdata.retry_count,
+            password=userdata.device_passwords.get(mac_address, None),
+        )
+
+    @classmethod
+    def _mqtt_update_device_info_callback(
+        cls,
+        mqtt_client: paho.mqtt.client.Client,
+        userdata: _MQTTCallbackUserdata,
+        message: paho.mqtt.client.MQTTMessage,
+    ) -> None:
+        # pylint: disable=unused-argument; callback
+        # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
+        _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
+        if message.retain:
+            _LOGGER.info("ignoring retained message")
+            return
+        actor = cls._init_from_topic(
+            userdata=userdata,
+            topic=message.topic,
+            expected_topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
+        )
+        if actor:
+            # pylint: disable=protected-access; own instance
+            actor._update_and_report_device_info(mqtt_client)
+
     @abc.abstractmethod
     def execute_command(
         self,
@@ -152,40 +205,38 @@ class _MQTTControlledActor(abc.ABC):
         if message.retain:
             _LOGGER.info("ignoring retained message")
             return
-        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
-        if not _mac_address_valid(mac_address):
-            _LOGGER.warning("invalid mac address %s", mac_address)
-            return
-        actor = cls(
-            mac_address=mac_address,
-            retry_count=userdata.retry_count,
-            password=userdata.device_passwords.get(mac_address, None),
-        )
-        actor.execute_command(
-            mqtt_message_payload=message.payload,
-            mqtt_client=mqtt_client,
-            # consider calling update+report method directly when adding support for battery levels
-            update_device_info=userdata.fetch_device_info,
+        actor = cls._init_from_topic(
+            userdata=userdata,
+            topic=message.topic,
+            expected_topic_levels=cls.MQTT_COMMAND_TOPIC_LEVELS,
         )
+        if actor:
+            actor.execute_command(
+                mqtt_message_payload=message.payload,
+                mqtt_client=mqtt_client,
+                update_device_info=userdata.fetch_device_info,
+            )
 
     @classmethod
-    def mqtt_subscribe(cls, mqtt_client: paho.mqtt.client.Client) -> None:
-        command_topic = "/".join(
-            "+" if isinstance(l, _MQTTTopicPlaceholder) else l
-            for l in cls.MQTT_COMMAND_TOPIC_LEVELS
-        )
-        _LOGGER.info("subscribing to MQTT topic %r", command_topic)
-        mqtt_client.subscribe(command_topic)
-        mqtt_client.message_callback_add(
-            sub=command_topic,
-            callback=cls._mqtt_command_callback,
-        )
+    def mqtt_subscribe(
+        cls,
+        mqtt_client: paho.mqtt.client.Client,
+        *,
+        enable_device_info_update_topic: bool,
+    ) -> None:
+        topics = [(cls.MQTT_COMMAND_TOPIC_LEVELS, cls._mqtt_command_callback)]
+        if enable_device_info_update_topic:
+            topics.append(
+                (
+                    cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
+                    cls._mqtt_update_device_info_callback,
+                )
+            )
+        for topic_levels, callback in topics:
+            topic = _join_mqtt_topic_levels(topic_levels, mac_address="+")
+            _LOGGER.info("subscribing to MQTT topic %r", topic)
+            mqtt_client.subscribe(topic)
+            mqtt_client.message_callback_add(sub=topic, callback=callback)
 
     def _mqtt_publish(
         self,

switchbot_mqtt/_cli.py

@@ -74,6 +74,10 @@ def _main() -> None:
         " after every command. Additionally report curtain motors' position on"
         f" topic {_CurtainMotor.get_mqtt_position_topic(mac_address='MAC_ADDRESS')}"
         " after executing stop commands."
+        " When this option is enabled, the mentioned reports may also be requested"
+        " by sending a MQTT message to the topic"
+        f" {_ButtonAutomator.get_mqtt_update_device_info_topic(mac_address='MAC_ADDRESS')}"
+        f" or {_CurtainMotor.get_mqtt_update_device_info_topic(mac_address='MAC_ADDRESS')}."
         " This option can also be enabled by assigning a non-empty value to the"
         " environment variable FETCH_DEVICE_INFO.",
     )

tests/test_mqtt.py

@@ -54,34 +54,36 @@ def test__run(
             device_passwords=device_passwords,
             fetch_device_info=fetch_device_info,
         )
-    mqtt_client_mock.assert_called_once_with(
-        userdata=switchbot_mqtt._MQTTCallbackUserdata(
-            retry_count=retry_count,
-            device_passwords=device_passwords,
-            fetch_device_info=fetch_device_info,
-        )
+    mqtt_client_mock.assert_called_once()
+    assert not mqtt_client_mock.call_args[0]
+    assert set(mqtt_client_mock.call_args[1].keys()) == {"userdata"}
+    userdata = mqtt_client_mock.call_args[1]["userdata"]
+    assert userdata == switchbot_mqtt._MQTTCallbackUserdata(
+        retry_count=retry_count,
+        device_passwords=device_passwords,
+        fetch_device_info=fetch_device_info,
     )
     assert not mqtt_client_mock().username_pw_set.called
     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)
     with caplog.at_level(logging.DEBUG):
-        mqtt_client_mock().on_connect(mqtt_client_mock(), None, {}, 0)
-    assert mqtt_client_mock().subscribe.call_args_list == [
-        unittest.mock.call("homeassistant/switch/switchbot/+/set"),
-        unittest.mock.call("homeassistant/cover/switchbot-curtain/+/set"),
-    ]
-    assert mqtt_client_mock().message_callback_add.call_args_list == [
-        unittest.mock.call(
-            sub="homeassistant/switch/switchbot/+/set",
-            callback=switchbot_mqtt._ButtonAutomator._mqtt_command_callback,
-        ),
-        unittest.mock.call(
-            sub="homeassistant/cover/switchbot-curtain/+/set",
-            callback=switchbot_mqtt._CurtainMotor._mqtt_command_callback,
-        ),
-    ]
+        mqtt_client_mock().on_connect(mqtt_client_mock(), userdata, {}, 0)
+    subscribe_mock = mqtt_client_mock().subscribe
+    assert subscribe_mock.call_count == (4 if fetch_device_info else 2)
+    for topic in [
+        "homeassistant/switch/switchbot/+/set",
+        "homeassistant/cover/switchbot-curtain/+/set",
+    ]:
+        assert unittest.mock.call(topic) in subscribe_mock.call_args_list
+    for topic in [
+        "homeassistant/switch/switchbot/+/request-device-info",
+        "homeassistant/cover/switchbot-curtain/+/request-device-info",
+    ]:
+        assert (
+            unittest.mock.call(topic) in subscribe_mock.call_args_list
+        ) == fetch_device_info
     mqtt_client_mock().loop_forever.assert_called_once_with()
-    assert caplog.record_tuples == [
+    assert caplog.record_tuples[:2] == [
         (
             "switchbot_mqtt",
             logging.INFO,
@@ -92,17 +94,18 @@ def test__run(
             logging.DEBUG,
             f"connected to MQTT broker {mqtt_host}:{mqtt_port}",
         ),
-        (
-            "switchbot_mqtt._actors._base",
-            logging.INFO,
-            "subscribing to MQTT topic 'homeassistant/switch/switchbot/+/set'",
-        ),
-        (
-            "switchbot_mqtt._actors._base",
-            logging.INFO,
-            "subscribing to MQTT topic 'homeassistant/cover/switchbot-curtain/+/set'",
-        ),
     ]
+    assert len(caplog.record_tuples) == (6 if fetch_device_info else 4)
+    assert (
+        "switchbot_mqtt._actors._base",
+        logging.INFO,
+        "subscribing to MQTT topic 'homeassistant/switch/switchbot/+/set'",
+    ) in caplog.record_tuples
+    assert (
+        "switchbot_mqtt._actors._base",
+        logging.INFO,
+        "subscribing to MQTT topic 'homeassistant/cover/switchbot-curtain/+/set'",
+    ) in caplog.record_tuples
 
 
 @pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
@@ -148,10 +151,13 @@ def test__run_authentication_missing_username(mqtt_host, mqtt_port, mqtt_passwor
 
 
 def _mock_actor_class(
-    command_topic_levels: typing.List[_MQTTTopicLevel],
+    *,
+    command_topic_levels: typing.List[_MQTTTopicLevel] = NotImplemented,
+    request_info_levels: typing.List[_MQTTTopicLevel] = NotImplemented,
 ) -> typing.Type:
     class _ActorMock(switchbot_mqtt._actors._MQTTControlledActor):
         MQTT_COMMAND_TOPIC_LEVELS = command_topic_levels
+        _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = request_info_levels
 
         def __init__(self, mac_address, retry_count, password):
             super().__init__(
@@ -172,6 +178,88 @@ def _mock_actor_class(
     return _ActorMock
 
 
+@pytest.mark.parametrize(
+    ("topic_levels", "topic", "expected_mac_address"),
+    [
+        (
+            switchbot_mqtt._actors._ButtonAutomator._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
+            b"homeassistant/switch/switchbot/aa:bb:cc:dd:ee:ff/request-device-info",
+            "aa:bb:cc:dd:ee:ff",
+        ),
+    ],
+)
+@pytest.mark.parametrize("payload", [b"", b"whatever"])
+def test__mqtt_update_device_info_callback(
+    caplog,
+    topic_levels: typing.List[_MQTTTopicLevel],
+    topic: bytes,
+    expected_mac_address: str,
+    payload: bytes,
+):
+    ActorMock = _mock_actor_class(request_info_levels=topic_levels)
+    message = MQTTMessage(topic=topic)
+    message.payload = payload
+    callback_userdata = switchbot_mqtt._MQTTCallbackUserdata(
+        retry_count=21,  # tested in test__mqtt_command_callback
+        device_passwords={},
+        fetch_device_info=True,
+    )
+    with unittest.mock.patch.object(
+        ActorMock, "__init__", return_value=None
+    ) as init_mock, unittest.mock.patch.object(
+        ActorMock, "_update_and_report_device_info"
+    ) as update_mock, caplog.at_level(
+        logging.DEBUG
+    ):
+        ActorMock._mqtt_update_device_info_callback(
+            "client_dummy", callback_userdata, message
+        )
+    init_mock.assert_called_once_with(
+        mac_address=expected_mac_address, retry_count=21, password=None
+    )
+    update_mock.assert_called_once_with("client_dummy")
+    assert caplog.record_tuples == [
+        (
+            "switchbot_mqtt._actors._base",
+            logging.DEBUG,
+            f"received topic={topic.decode()} payload={payload!r}",
+        )
+    ]
+
+
+def test__mqtt_update_device_info_callback_ignore_retained(caplog):
+    ActorMock = _mock_actor_class(
+        request_info_levels=[_MQTTTopicPlaceholder.MAC_ADDRESS, "request"]
+    )
+    message = MQTTMessage(topic=b"aa:bb:cc:dd:ee:ff/request")
+    message.payload = b""
+    message.retain = True
+    with unittest.mock.patch.object(
+        ActorMock, "__init__", return_value=None
+    ) as init_mock, unittest.mock.patch.object(
+        ActorMock, "execute_command"
+    ) as execute_command_mock, caplog.at_level(
+        logging.DEBUG
+    ):
+        ActorMock._mqtt_update_device_info_callback(
+            "client_dummy",
+            switchbot_mqtt._MQTTCallbackUserdata(
+                retry_count=21, device_passwords={}, fetch_device_info=True
+            ),
+            message,
+        )
+    init_mock.assert_not_called()
+    execute_command_mock.assert_not_called()
+    assert caplog.record_tuples == [
+        (
+            "switchbot_mqtt._actors._base",
+            logging.DEBUG,
+            "received topic=aa:bb:cc:dd:ee:ff/request payload=b''",
+        ),
+        ("switchbot_mqtt._actors._base", logging.INFO, "ignoring retained message"),
+    ]
+
+
 @pytest.mark.parametrize(
     ("command_topic_levels", "topic", "payload", "expected_mac_address"),
     [
@@ -230,7 +318,7 @@ def test__mqtt_command_callback(
     retry_count: int,
     fetch_device_info: bool,
 ):
-    ActorMock = _mock_actor_class(command_topic_levels)
+    ActorMock = _mock_actor_class(command_topic_levels=command_topic_levels)
     message = MQTTMessage(topic=topic)
     message.payload = payload
     callback_userdata = switchbot_mqtt._MQTTCallbackUserdata(
@@ -272,7 +360,9 @@ def test__mqtt_command_callback(
     ],
 )
 def test__mqtt_command_callback_password(mac_address, expected_password):
-    ActorMock = _mock_actor_class(["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS])
+    ActorMock = _mock_actor_class(
+        command_topic_levels=["switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS]
+    )
     message = MQTTMessage(topic=b"switchbot/" + mac_address.encode())
     message.payload = b"whatever"
     callback_userdata = switchbot_mqtt._MQTTCallbackUserdata(
@@ -310,7 +400,7 @@ def test__mqtt_command_callback_password(mac_address, expected_password):
 )
 def test__mqtt_command_callback_unexpected_topic(caplog, topic: bytes, payload: bytes):
     ActorMock = _mock_actor_class(
-        switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
+        command_topic_levels=switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
     )
     message = MQTTMessage(topic=topic)
     message.payload = payload
@@ -349,7 +439,7 @@ def test__mqtt_command_callback_invalid_mac_address(
     caplog, mac_address: str, payload: bytes
 ):
     ActorMock = _mock_actor_class(
-        switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
+        command_topic_levels=switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
     )
     topic = f"homeassistant/switch/switchbot/{mac_address}/set".encode()
     message = MQTTMessage(topic=topic)
@@ -390,7 +480,7 @@ def test__mqtt_command_callback_invalid_mac_address(
 )
 def test__mqtt_command_callback_ignore_retained(caplog, topic: bytes, payload: bytes):
     ActorMock = _mock_actor_class(
-        switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
+        command_topic_levels=switchbot_mqtt._ButtonAutomator.MQTT_COMMAND_TOPIC_LEVELS
     )
     message = MQTTMessage(topic=topic)
     message.payload = payload