110 lines
3.4 KiB
Python
110 lines
3.4 KiB
Python
"""本机 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()
|