fix: select RXTX channel by firmware family

tests/test_transport_local.py

@@ -0,0 +1,109 @@
+"""本机 BLE 传输层的固件通道选择测试。"""
+
+import pytest
+
+from ppclock import protocol as P
+from ppclock.transports import local
+
+
+class FakeCharacteristic:
+    def __init__(self, uuid, properties):
+        self.uuid = uuid
+        self.properties = properties
+
+
+class FakeServices:
+    def __init__(self):
+        self.characters = {
+            P.RXTX_CHAR_UUID_LEGACY: FakeCharacteristic(
+                P.RXTX_CHAR_UUID_LEGACY, ["write"]),
+            P.RXTX_CHAR_UUID: FakeCharacteristic(
+                P.RXTX_CHAR_UUID, ["read", "write", "notify"]),
+        }
+
+    def get_characteristic(self, uuid):
+        return self.characters.get(str(uuid).lower())
+
+
+class FakeBleakClient:
+    firmware_revision = "PP_da14585_4.2"
+    fail_firmware_read = False
+
+    def __init__(self, address, timeout):
+        self.address = address
+        self.timeout = timeout
+        self.services = FakeServices()
+        self.is_connected = False
+        self.notifications = []
+        self.writes = []
+
+    async def connect(self):
+        self.is_connected = True
+
+    async def disconnect(self):
+        self.is_connected = False
+
+    async def read_gatt_char(self, uuid):
+        if str(uuid).lower() == P.FIRMWARE_REVISION_UUID:
+            if self.fail_firmware_read:
+                raise RuntimeError("DIS unavailable")
+            return self.firmware_revision.encode()
+        return b""
+
+    async def write_gatt_char(self, uuid, data, response=True):
+        target = uuid.uuid if hasattr(uuid, "uuid") else uuid
+        self.writes.append((target, bytes(data), response))
+
+    async def start_notify(self, uuid, callback):
+        target = uuid.uuid if hasattr(uuid, "uuid") else uuid
+        self.notifications.append(target)
+
+    async def stop_notify(self, uuid):
+        self.notifications.remove(uuid)
+
+
+@pytest.mark.asyncio
+async def test_hm42_selects_write_only_1f1f(monkeypatch):
+    class HM42Client(FakeBleakClient):
+        firmware_revision = "HM42_AIO_V1.0.9"
+
+    monkeypatch.setattr(local, "BleakClient", HM42Client)
+    transport = local.BLETransport("18:BC:5A:7E:BB:1E")
+    await transport.connect()
+
+    assert transport.firmware_revision == "HM42_AIO_V1.0.9"
+    assert transport.rxtx_uuid == P.RXTX_CHAR_UUID_LEGACY
+    assert not transport._notify_started
+    await transport.write_rxtx(b"\xe1\x02\x02")
+    assert transport._client.writes[-1][0] == P.RXTX_CHAR_UUID_LEGACY
+    with pytest.raises(local.TransportError, match="不支持通知"):
+        await transport.request_device_id()
+    await transport.close()
+
+
+@pytest.mark.asyncio
+async def test_pp42_keeps_331f_with_notifications(monkeypatch):
+    monkeypatch.setattr(local, "BleakClient", FakeBleakClient)
+    transport = local.BLETransport("18:BC:5A:5D:BF:28")
+    await transport.connect()
+
+    assert transport.rxtx_uuid == P.RXTX_CHAR_UUID
+    assert transport._notify_started
+    assert transport._client.notifications == [P.RXTX_CHAR_UUID]
+    await transport.write_rxtx(b"\xe2")
+    assert transport._client.writes[-1][0] == P.RXTX_CHAR_UUID
+    await transport.close()
+
+
+@pytest.mark.asyncio
+async def test_unknown_firmware_preserves_331f_default(monkeypatch):
+    class NoDISClient(FakeBleakClient):
+        fail_firmware_read = True
+
+    monkeypatch.setattr(local, "BleakClient", NoDISClient)
+    transport = local.BLETransport("AA:BB:CC:DD:EE:FF")
+    await transport.connect()
+
+    assert transport.firmware_revision is None
+    assert transport.rxtx_uuid == P.RXTX_CHAR_UUID
+    await transport.close()