🦎 q7: narrow decoded command response type

Author: arduano

roborock/devices/rpc/b01_q7_channel.py

@@ -6,7 +6,7 @@
 import json
 import logging
 from collections.abc import Callable
-from typing import Any, TypeVar
+from typing import TypeAlias, TypeVar
 
 from roborock.devices.transport.mqtt_channel import MqttChannel
 from roborock.exceptions import RoborockException
@@ -16,6 +16,7 @@
 _LOGGER = logging.getLogger(__name__)
 _TIMEOUT = 10.0
 _T = TypeVar("_T")
+DecodedB01Response: TypeAlias = dict[str, object] | str
 
 
 def _matches_map_response(response_message: RoborockMessage, *, version: bytes | None) -> bytes | None:
@@ -61,11 +62,11 @@ def on_message(response_message: RoborockMessage) -> None:
 async def send_decoded_command(
     mqtt_channel: MqttChannel,
     request_message: Q7RequestMessage,
-) -> Any:
+) -> DecodedB01Response:
     """Send a command on the MQTT channel and get a decoded response."""
     _LOGGER.debug("Sending B01 MQTT command: %s", request_message)
 
-    def find_response(response_message: RoborockMessage) -> Any | None:
+    def find_response(response_message: RoborockMessage) -> DecodedB01Response | None:
         """Handle incoming messages and resolve the future."""
         try:
             decoded_dps = decode_rpc_response(response_message)

tests/devices/traits/b01/q7/test_init.py

@@ -111,6 +111,21 @@ async def test_send_decoded_command_error_code(fake_channel: FakeChannel, messag
         await send_decoded_command(fake_channel, Q7RequestMessage(dps=10000, command="prop.get", params=[]))  # type: ignore[arg-type]
 
 
+async def test_send_decoded_command_returns_dict_for_prop_get(
+    fake_channel: FakeChannel, message_builder: B01MessageBuilder
+):
+    """Property queries still return decoded dict payloads."""
+    message = message_builder.build({"status": 8})
+    fake_channel.response_queue.append(message)
+
+    result = await send_decoded_command(
+        cast(Any, fake_channel),
+        Q7RequestMessage(dps=10000, command="prop.get", params=[]),  # type: ignore[arg-type]
+    )
+
+    assert result == {"status": 8}
+
+
 async def test_send_decoded_command_allows_ok_string_ack(fake_channel: FakeChannel, message_builder: B01MessageBuilder):
     """Command ACKs may return plain string payloads like ``ok``."""
     message = message_builder.build("ok")