139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
"""命令层测试 — 用 FakeTransport 验证高层操作的帧序列与顺序（证据 docs/protocol.md）。"""
|
|
import datetime
|
|
|
|
import pytest
|
|
|
|
from ppclock import commands as C
|
|
|
|
|
|
class FakeTransport:
|
|
def __init__(self, id_response="DEVICE12345678"):
|
|
self.epd_writes = []
|
|
self.rxtx_writes = []
|
|
self.id_response = id_response
|
|
self.delays = []
|
|
|
|
async def write_epd(self, data, response=True):
|
|
self.epd_writes.append(data)
|
|
|
|
async def write_rxtx(self, data, response=True):
|
|
self.rxtx_writes.append(data)
|
|
|
|
async def request_device_id(self, timeout=8.0):
|
|
self.rxtx_writes.append(bytes([0xEF, 0xEF]))
|
|
return self.id_response
|
|
|
|
async def delay(self, seconds):
|
|
self.delays.append(seconds)
|
|
|
|
|
|
@pytest.fixture
|
|
def t():
|
|
return FakeTransport()
|
|
|
|
|
|
class TestTimeAndMode:
|
|
@pytest.mark.asyncio
|
|
async def test_set_time_sends_dd_then_commit(self, t):
|
|
dt = datetime.datetime(2026, 7, 24, 15, 30, 0)
|
|
await C.set_time(t, dt)
|
|
assert t.rxtx_writes[0][0] == 0xDD
|
|
assert t.rxtx_writes[-1] == bytes([0xE2])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_mode_appends_commit(self, t):
|
|
await C.set_mode(t, "clock1")
|
|
assert t.rxtx_writes == [bytes.fromhex("e10201"), bytes([0xE2])]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_uses_epd_channel(self, t):
|
|
await C.clear(t)
|
|
assert t.epd_writes == [bytes([0x00, 0x00]), bytes([0x01])]
|
|
assert t.rxtx_writes == []
|
|
|
|
|
|
class TestUploadImage:
|
|
@pytest.mark.asyncio
|
|
async def test_full_image_sequence(self, t):
|
|
"""整图：bw块×63 → 红块×63 → 延时 → 01"""
|
|
bw = bytes(15000)
|
|
red = bytes([0xFF]) * 15000
|
|
await C.upload_image(t, bw, red, small=False)
|
|
assert t.epd_writes[0][:4] == bytes([0x03, 0xFF, 0x00, 0x00])
|
|
assert t.epd_writes[63][:2] == bytes([0x03, 0x00]) # 红平面首块
|
|
assert t.epd_writes[-1] == bytes([0x01])
|
|
assert len(t.epd_writes) == 63 + 63 + 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mono_image_no_red_blocks(self, t):
|
|
await C.upload_image(t, bytes(15000), None, small=False)
|
|
assert len(t.epd_writes) == 63 + 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_slot_select_before_upload(self, t):
|
|
await C.upload_image(t, bytes(15000), None, small=False, slot=2)
|
|
assert t.rxtx_writes[0] == bytes([0xEA, 0x02])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_small_image_ends_with_AA(self, t):
|
|
await C.upload_image(t, bytes(3500), None, small=True)
|
|
assert t.epd_writes[-1] == bytes([0xAA])
|
|
assert t.epd_writes[0][0] == 0x04
|
|
|
|
|
|
class TestOtherCommands:
|
|
@pytest.mark.asyncio
|
|
async def test_countdown_set(self, t):
|
|
d = datetime.date(2026, 12, 31)
|
|
await C.countdown_set(t, d, "clock")
|
|
assert t.rxtx_writes[0] == bytes([0xFF, 0x00, 0x20, 0x26, 0x12, 0x31])
|
|
assert t.rxtx_writes[-1] == bytes([0xE2])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_countdown_off(self, t):
|
|
await C.countdown_off(t)
|
|
assert t.rxtx_writes == [bytes([0xFA]), bytes([0xE2])]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parking(self, t):
|
|
await C.parking(t, "13800138000")
|
|
assert t.rxtx_writes[0] == bytes([0xEF]) + bytes(
|
|
int(f"{d}{d}", 16) for d in "13800138000"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sleep_on_off(self, t):
|
|
await C.sleep(t, True, 23, 7)
|
|
assert t.rxtx_writes[0] == bytes([0xFB, 0x01, 0x23, 0x07])
|
|
await C.sleep(t, False, 23, 7)
|
|
assert t.rxtx_writes[1] == bytes([0xFB, 0x00, 0x23, 0x07])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rotation(self, t):
|
|
await C.rotation(t, count=4)
|
|
assert t.rxtx_writes[0] == bytes.fromhex("EA0003")
|
|
await C.rotation(t, interval=5, unit="min")
|
|
assert t.rxtx_writes[1] == bytes.fromhex("e9010501")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activate(self, t):
|
|
await C.activate(t, "A1B2" * 8)
|
|
assert t.rxtx_writes[0] == bytes([0xEF]) + bytes.fromhex("A1B2" * 8)
|
|
assert t.rxtx_writes[-1] == bytes([0xE2])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wifi(self, t):
|
|
await C.wifi(t, "MyAP", "pw", "101010100")
|
|
assert t.rxtx_writes[0] == b"WIFI_SSID:MyAP;WIFI_PASS:pw;CITY:101010100"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lut(self, t):
|
|
await C.lut(t, 0x05)
|
|
assert t.rxtx_writes[0] == bytes([0xE6, 0x05])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_device_id(self, t):
|
|
dev_id = await C.get_device_id(t)
|
|
assert t.rxtx_writes[0] == bytes([0xEF, 0xEF])
|
|
assert dev_id == "DEVICE12345678"
|