223 lines
7.2 KiB
Python
223 lines
7.2 KiB
Python
"""DeviceManager 测试：按需连接/串行/空闲超时/掉线重连。"""
|
||
import asyncio
|
||
|
||
import pytest
|
||
from bleak.exc import BleakError
|
||
|
||
from ppclock.client import PPClient
|
||
from ppclock.device_manager import DeviceManager
|
||
from ppclock.transports.base import TransportError
|
||
|
||
|
||
class FakeTransport:
|
||
def __init__(self):
|
||
self.address = "AA:BB:CC:DD:EE:FF"
|
||
self.enters = 0
|
||
self.exits = 0
|
||
self.rxtx_writes = []
|
||
self.fail_next_write = False
|
||
# fail_next_write 触发时抛出的异常实例（默认 SDK 包装后的 TransportError；
|
||
# 真实硬件路径常是 bleak 裸抛的 BleakError，测试可替换）
|
||
self.fail_exc = TransportError("connection lost")
|
||
self.firmware_revision = "HM42_AIO_V1.0.9"
|
||
self.rxtx_uuid = "00001f1f-0000-1000-8000-00805f9b34fb"
|
||
|
||
async def __aenter__(self):
|
||
self.enters += 1
|
||
return self
|
||
|
||
async def __aexit__(self, *exc):
|
||
self.exits += 1
|
||
return False
|
||
|
||
async def write_epd(self, data, response=True):
|
||
pass
|
||
|
||
async def write_rxtx(self, data, response=True):
|
||
if self.fail_next_write:
|
||
self.fail_next_write = False
|
||
raise self.fail_exc
|
||
self.rxtx_writes.append(data)
|
||
|
||
async def read_rxtx(self):
|
||
return b"\x00"
|
||
|
||
async def request_device_id(self, timeout=8.0):
|
||
return "81233F3C267112"
|
||
|
||
async def run_ota(self, image, on_progress=None):
|
||
return {}
|
||
|
||
async def delay(self, seconds):
|
||
pass
|
||
|
||
|
||
def make_manager(transports, **kw):
|
||
"""client_factory 依次弹出预置 transport 包成 PPClient。"""
|
||
def factory(mac, timeout):
|
||
return PPClient(transports.pop(0))
|
||
kw.setdefault("client_factory", factory)
|
||
return DeviceManager(**kw)
|
||
|
||
|
||
class TestConnect:
|
||
@pytest.mark.asyncio
|
||
async def test_connect_on_demand_and_reuse(self):
|
||
t = FakeTransport()
|
||
m = make_manager([t])
|
||
assert not m.connected
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
await m.run(lambda dev: dev.set_mode("clock2"))
|
||
assert t.enters == 1 # 第二次复用连接
|
||
assert m.connected
|
||
await m.close()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_remembers_address(self):
|
||
t = FakeTransport()
|
||
m = make_manager([t], mac=None)
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
assert (await m.status())["mac"] == "AA:BB:CC:DD:EE:FF"
|
||
await m.close()
|
||
|
||
|
||
class TestSerialization:
|
||
@pytest.mark.asyncio
|
||
async def test_ops_are_serialized(self):
|
||
t = FakeTransport()
|
||
m = make_manager([t])
|
||
concurrent = 0
|
||
peak = 0
|
||
|
||
async def op(dev):
|
||
nonlocal concurrent, peak
|
||
concurrent += 1
|
||
peak = max(peak, concurrent)
|
||
await asyncio.sleep(0.02)
|
||
concurrent -= 1
|
||
return 1
|
||
|
||
await asyncio.gather(*(m.run(op) for _ in range(5)))
|
||
assert peak == 1
|
||
await m.close()
|
||
|
||
|
||
class TestIdleTimeout:
|
||
@pytest.mark.asyncio
|
||
async def test_idle_disconnect(self):
|
||
t = FakeTransport()
|
||
m = make_manager([t], idle_timeout=0.05)
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
assert m.connected
|
||
await asyncio.sleep(0.15)
|
||
assert not m.connected
|
||
assert t.exits == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_activity_resets_idle(self):
|
||
t = FakeTransport()
|
||
m = make_manager([t], idle_timeout=0.1)
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
await asyncio.sleep(0.06)
|
||
await m.run(lambda dev: dev.set_mode("clock2"))
|
||
await asyncio.sleep(0.06)
|
||
assert m.connected # 第二次活动续期
|
||
await m.close()
|
||
|
||
|
||
class TestReconnect:
|
||
@pytest.mark.asyncio
|
||
async def test_reconnect_once_on_transport_error(self):
|
||
t1, t2 = FakeTransport(), FakeTransport()
|
||
t1.fail_next_write = True
|
||
m = make_manager([t1, t2])
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
assert t1.enters == 1 and t2.enters == 1 # 掉线重连一次并成功
|
||
await m.close()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_retry_exhaustion_raises(self):
|
||
t1, t2 = FakeTransport(), FakeTransport()
|
||
t1.fail_next_write = True
|
||
t2.fail_next_write = True
|
||
m = make_manager([t1, t2])
|
||
with pytest.raises(TransportError):
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
await m.close()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reconnect_on_raw_bleak_error(self):
|
||
"""真实掉线 bleak 裸抛 BleakError（不经 SDK 包装），也要重连重试。"""
|
||
t1, t2 = FakeTransport(), FakeTransport()
|
||
t1.fail_next_write = True
|
||
t1.fail_exc = BleakError("Not connected")
|
||
m = make_manager([t1, t2])
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
assert t1.enters == 1 and t2.enters == 1 # 掉线重连一次并成功
|
||
await m.close()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_retry_exhaustion_raises_bleak_error(self):
|
||
"""重连后仍抛 BleakError → 原样上抛。"""
|
||
t1, t2 = FakeTransport(), FakeTransport()
|
||
for t in (t1, t2):
|
||
t.fail_next_write = True
|
||
t.fail_exc = BleakError("Not connected")
|
||
m = make_manager([t1, t2])
|
||
with pytest.raises(BleakError):
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
await m.close()
|
||
|
||
|
||
class TestEnterFailure:
|
||
@pytest.mark.asyncio
|
||
async def test_aenter_failure_cleans_up(self):
|
||
"""__aenter__ 抛异常时必须补 __aexit__（否则真实硬件泄漏半开连接）。"""
|
||
|
||
class HalfOpenClient:
|
||
def __init__(self):
|
||
self.exits = 0
|
||
|
||
async def __aenter__(self):
|
||
raise BleakError("connect boom")
|
||
|
||
async def __aexit__(self, *exc):
|
||
self.exits += 1
|
||
return False
|
||
|
||
client = HalfOpenClient()
|
||
m = DeviceManager(mac="AA:BB:CC:DD:EE:FF",
|
||
client_factory=lambda mac, timeout: client)
|
||
with pytest.raises(BleakError):
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
assert client.exits == 1
|
||
assert not m.connected
|
||
await m.close()
|
||
|
||
|
||
class TestStatus:
|
||
@pytest.mark.asyncio
|
||
async def test_status_fields(self):
|
||
t = FakeTransport()
|
||
m = make_manager([t], connect_timeout=90.0, idle_timeout=300.0)
|
||
s = await m.status()
|
||
assert s["connected"] is True # status 自身触发连接
|
||
assert s["device_id"] == "81233F3C267112"
|
||
assert s["connect_timeout"] == 90.0
|
||
assert s["idle_timeout"] == 300.0
|
||
assert s["idle_seconds"] is None # 无先前活动；status 自身不得把读数归零
|
||
assert s["firmware_revision"] == "HM42_AIO_V1.0.9"
|
||
assert s["rxtx_uuid"].startswith("00001f1f")
|
||
await m.close()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_status_idle_seconds_reflects_last_activity(self):
|
||
"""status 自身的 _touch 不得把 idle_seconds 归零（先算后摸）。"""
|
||
t = FakeTransport()
|
||
m = make_manager([t])
|
||
await m.run(lambda dev: dev.set_mode("clock1"))
|
||
await asyncio.sleep(0.12)
|
||
s = await m.status()
|
||
assert s["idle_seconds"] >= 0.1
|
||
await m.close()
|