Skip to content

refactor: decouple B01 (Q7/Q10) protocol layer from transport layer#859

Open
allenporter wants to merge 1 commit into
Python-roborock:mainfrom
allenporter:decouple-b01-transport-layer
Open

refactor: decouple B01 (Q7/Q10) protocol layer from transport layer#859
allenporter wants to merge 1 commit into
Python-roborock:mainfrom
allenporter:decouple-b01-transport-layer

Conversation

@allenporter

@allenporter allenporter commented Jul 2, 2026

Copy link
Copy Markdown
Contributor

Decouple B01 (Q7 / Q10) Protocol and Transport Layers

This PR decouples protocol-specific codecs, encryption/decryption, map-key derivation, and payload formatting from the device traits layer for B01 version (Q7 and Q10) devices.

Rationale & Rationale for Home Assistant Test Fakes

Previously, Q7 and Q10 traits communicated directly with raw MqttChannel objects. This forced the traits (and their corresponding unit tests) to handle low-level packet encoding, encryption signing, AES ECB/CBC padding, and local keys.

By separating the traits layer from the transport layer, we introduce clean boundaries where traits communicate over simple plaintext RPC channels (Q7RpcChannel, Q7MapRpcChannel, and Q10RpcChannel).

A key motivation for this change is to improve the test fakes exposed to Home Assistant. Rather than forcing downstream integrations to mock deep internals or mock raw MQTT messages, Home Assistant will be able to instantiate the actual python-roborock device objects (using real traits classes) and simply feed them with plaintext channel fakes. This will happen in separate PR and will make testing the Home Assistant integration much more straightforward and reliable.


Architectural Changes

  • Unified Channel Layer: Introduced B01Q7Channel and B01Q10Channel wrapping MQTT transport. They implement the low-level Channel interface and directly conform to the plaintext RPC protocols (Q7RpcChannel, Q7MapRpcChannel, and Q10RpcChannel). All AES block padding, encryption, signature verification, and map key derivation details are fully encapsulated here.
  • Simplified Traits: Refactored the Q7 and Q10 traits to use the plaintext interfaces, removing all dependencies on raw messages, local keys, and transport codecs.
  • Simplified Unit Tests: Trait unit tests now use simple fake channels (FakeQ7Channel and FakeB01Q10Channel) that verify plaintext parameters and enqueue plaintext mock responses, eliminating the need for B01MessageBuilder in trait tests.
  • New Channel Tests: Created dedicated unit test suites (test_b01_q7_channel.py and test_b01_q10_channel.py) to verify encryption/decryption, padding, and packet routing in isolation.

@allenporter allenporter force-pushed the decouple-b01-transport-layer branch 3 times, most recently from b27b400 to a40dd0c Compare July 2, 2026 21:25
@allenporter allenporter force-pushed the decouple-b01-transport-layer branch from a40dd0c to f218dd4 Compare July 2, 2026 21:29
@allenporter allenporter marked this pull request as ready for review July 2, 2026 21:30
Copilot AI review requested due to automatic review settings July 2, 2026 21:30

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR refactors the B01 device stack (Q7/Q10) to decouple protocol/transport concerns from the traits layer by introducing plaintext RPC channel interfaces and unified transport wrappers, and updates tests to validate traits against plaintext fakes while moving encryption/codec validation into dedicated channel tests.

Changes:

  • Introduces unified transport wrappers (B01Q7Channel, B01Q10Channel) and plaintext RPC protocols (Q7RpcChannel/Q7MapRpcChannel, Q10RpcChannel) to isolate encoding/encryption/map-key logic from traits.
  • Refactors Q7/Q10 traits to depend on the new plaintext channel interfaces (removing direct raw MQTT/message handling).
  • Reworks and expands tests: trait tests use plaintext fake channels; new RPC channel tests cover low-level message formatting/decoding behaviors.

Reviewed changes

Copilot reviewed 23 out of 23 changed files in this pull request and generated 1 comment.

Show a summary per file
File Description
tests/devices/traits/b01/q7/test_map.py Updates Q7 map trait test to assert plaintext RPC method/params rather than encoded MQTT payload.
tests/devices/traits/b01/q7/test_map_content.py Updates Q7 map-content tests to use plaintext map payloads and RPC method assertions.
tests/devices/traits/b01/q7/test_init.py Refactors Q7 API tests to assert RPC calls via plaintext methods rather than decrypted payload inspection.
tests/devices/traits/b01/q7/test_clean_summary.py Migrates Q7 clean summary tests to new fake RPC channel + exception simulation.
tests/devices/traits/b01/q7/conftest.py Introduces FakeQ7Channel implementing plaintext RPC/map RPC protocols and updates fixtures.
tests/devices/traits/b01/q7/init.py Removes legacy encoded-message test builder (now covered by dedicated channel tests).
tests/devices/traits/b01/q10/test_vacuum.py Updates vacuum trait tests to assert plaintext command dispatch rather than raw DPS JSON payloads.
tests/devices/traits/b01/q10/test_status.py Updates streaming/refresh tests to operate on decoded Q10Message stream and plaintext command recording.
tests/devices/traits/b01/q10/test_settings.py Switches settings writer tests to a plaintext Q10 RPC channel fake and command-based assertions.
tests/devices/traits/b01/q10/test_remote.py Updates remote trait tests to assert command/params rather than JSON payload.
tests/devices/traits/b01/q10/test_map.py Updates map integration tests to push parsed map/trace packets into the decoded message stream.
tests/devices/traits/b01/q10/conftest.py Adds Q10 plaintext fakes (FakeQ10RpcChannel, FakeB01Q10Channel) and shared fixtures.
tests/devices/rpc/test_b01_q7_channel.py Adds dedicated B01 Q7 channel tests for encoded request formatting, decode behavior, and timeouts.
tests/devices/rpc/test_b01_q10_channel.py Adds dedicated B01 Q10 channel tests for request formatting and decoded stream behavior.
roborock/devices/traits/b01/q7/map.py Refactors Q7 map trait to use Q7RpcChannel.send_command() instead of low-level decoded-command helper.
roborock/devices/traits/b01/q7/map_content.py Refactors Q7 map-content trait to use plaintext map RPC command calls.
roborock/devices/traits/b01/q7/clean_summary.py Refactors Q7 clean-summary trait to use Q7RpcChannel.send_command().
roborock/devices/traits/b01/q7/init.py Updates Q7 API to accept separate RPC + map RPC channels; simplifies send() to delegate to RPC channel.
roborock/devices/traits/b01/q10/command.py Switches Q10 command trait from raw MQTT usage to Q10RpcChannel.
roborock/devices/traits/b01/q10/init.py Updates Q10 API to depend on unified channel (B01Q10Channel) and stream decoded messages via subscribe_stream().
roborock/devices/rpc/b01_q7_channel.py Introduces B01Q7Channel wrapper implementing plaintext RPC/map RPC; adds create_b01_q7_channel().
roborock/devices/rpc/b01_q10_channel.py Introduces B01Q10Channel wrapper + Q10RpcChannel protocol; adds create_b01_q10_channel().
roborock/devices/device_manager.py Wires device creation to instantiate and pass the new unified B01 channels into Q7/Q10 trait factories.
Comments suppressed due to low confidence (1)

tests/devices/traits/b01/q7/test_map.py:18

  • This refactor removed the dedicated assertion that Q7MapList.current_map_id prefers the entry marked cur=True. That selection logic is still relied on by MapContentTrait.refresh(), but it no longer appears to be covered by any unit test.
async def test_q7_map_refresh(q7_api: Q7PropertiesApi, fake_channel: FakeQ7Channel):
    """Test retrieving lists of saved maps."""
    fake_channel.response_queue.append({"map_list": [{"id": 111, "name": "Map 1"}]})
    await q7_api.map.refresh()

    assert len(fake_channel.published_commands) == 1
    command, params = fake_channel.published_commands[0]
    assert command == RoborockB01Q7Methods.GET_MAP_LIST
    assert params == {}
    assert q7_api.map.map_list == [Q7MapListEntry(id=111)]


💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Comment on lines +26 to +30
async def send_map_command(self, command: Any, params: Any = None) -> bytes:
self.published_commands.append((command, params))
if self.response_queue:
return self.response_queue.pop(0)
return b""
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants