184 lines
6.3 KiB
Python
184 lines
6.3 KiB
Python
"""PPClient 门面 API 测试。"""
|
|
import datetime
|
|
|
|
import pytest
|
|
|
|
from ppclock.client import PPClient
|
|
|
|
|
|
class FakeTransport:
|
|
def __init__(self, firmware_revision=None):
|
|
self.epd_writes = []
|
|
self.rxtx_writes = []
|
|
self.address = "AA:BB:CC:DD:EE:FF"
|
|
self.firmware_revision = firmware_revision
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *exc):
|
|
return False
|
|
|
|
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 write_alt(self, data, response=True):
|
|
self.rxtx_writes.append(data)
|
|
|
|
async def read_rxtx(self):
|
|
return b"\x00"
|
|
|
|
async def request_device_id(self, timeout=8.0):
|
|
self.rxtx_writes.append(bytes([0xEF, 0xEF]))
|
|
return "81233F3C267112"
|
|
|
|
async def run_ota(self, image, on_progress=None):
|
|
return {"blocks": 1, "bytes": len(image)}
|
|
|
|
async def delay(self, seconds):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return PPClient(FakeTransport())
|
|
|
|
|
|
class TestLifecycle:
|
|
@pytest.mark.asyncio
|
|
async def test_context_manager(self, client):
|
|
async with client as dev:
|
|
assert dev is client
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_device_id(self, client):
|
|
assert await client.get_device_id() == "81233F3C267112"
|
|
|
|
|
|
class TestCommands:
|
|
@pytest.mark.asyncio
|
|
async def test_set_time_with_tz(self, client):
|
|
await client.set_time(tz=8)
|
|
assert client._t.rxtx_writes[0][0] == 0xDD
|
|
assert client._t.rxtx_writes[-1] == bytes([0xE2])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_mode(self, client):
|
|
await client.set_mode("clock1")
|
|
assert client._t.rxtx_writes == [bytes.fromhex("e10201"), bytes([0xE2])]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_image_pil(self, client):
|
|
from PIL import Image
|
|
img = Image.new("RGB", (400, 300), (255, 255, 255))
|
|
result = await client.upload_image(img, slot=0, algo="none")
|
|
assert result["bytes_bw"] == 15000
|
|
assert client._t.rxtx_writes[0] == bytes([0xEA, 0x00])
|
|
assert len(client._t.epd_writes) == 63 + 63 + 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_mono_non_hm42_sends_one_plane(self):
|
|
from PIL import Image
|
|
client = PPClient(FakeTransport("PP_da14585_4.2"))
|
|
result = await client.upload_image(
|
|
Image.new("RGB", (400, 300), "white"), slot=1,
|
|
algo="none", mono=True,
|
|
)
|
|
assert result["bytes_bw"] == 15000
|
|
assert result["bytes_red"] == 0
|
|
assert len(client._t.epd_writes) == 63 + 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_mono_hm42_appends_empty_red_plane(self):
|
|
from PIL import Image
|
|
client = PPClient(FakeTransport("HM42_AIO_V1.0.9"))
|
|
result = await client.upload_image(
|
|
Image.new("RGB", (400, 300), "white"), slot=1,
|
|
algo="none", mono=True,
|
|
)
|
|
assert result["bytes_bw"] == 15000
|
|
assert result["bytes_red"] == 15000
|
|
assert client._t.epd_writes[:3] == [
|
|
bytes.fromhex("8001"),
|
|
bytes.fromhex("0000"),
|
|
bytes.fromhex("020000"),
|
|
]
|
|
assert len(client._t.epd_writes) == 3 + 63 + 63 + 1
|
|
first_red = client._t.epd_writes[3 + 63]
|
|
assert first_red[:4] == bytes.fromhex("03000000")
|
|
assert set(first_red[4:]) == {0}
|
|
assert client._t.epd_writes[-1] == bytes.fromhex("0101")
|
|
assert client._t.rxtx_writes[-1] == bytes.fromhex("e2")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_render_template(self, client):
|
|
result = await client.render_template("custom", {"text": "你好"})
|
|
assert result["template"] == "custom"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_countdown_and_off(self, client):
|
|
await client.countdown(datetime.date(2026, 12, 31), prefix="目标")
|
|
await client.countdown_off()
|
|
assert client._t.rxtx_writes[-2] == bytes([0xFA])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_calendar_text(self, client):
|
|
await client.calendar_text("倒计时")
|
|
kinds = [w[0] for w in client._t.rxtx_writes if len(w) >= 1]
|
|
assert 0x99 in kinds and 0xFA in kinds
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activate_auto(self, client):
|
|
result = await client.activate_auto()
|
|
assert result["device_id"] == "81233F3C267112"
|
|
assert result["mac"] == "28bf5d5abc18"
|
|
# EF + keygen 码已下发
|
|
assert any(w[0] == 0xEF and len(w) == 7 for w in client._t.rxtx_writes)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parking_sleep_lut_wifi_raw(self, client):
|
|
await client.parking("13800138000")
|
|
await client.sleep(True, 23, 7)
|
|
await client.lut(0x05)
|
|
await client.wifi("AP", "pw", "101010100")
|
|
await client.raw(bytes([0xE2]))
|
|
w = client._t.rxtx_writes
|
|
assert w[0][0] == 0xEF # 停车牌帧
|
|
assert w[1] == bytes([0xE2]) # 停车牌提交
|
|
assert w[2] == bytes([0xFB, 0x01, 0x23, 0x07])
|
|
assert w[3] == bytes([0xE6, 0x05])
|
|
assert w[4].startswith(b"WIFI_SSID:")
|
|
assert w[5] == bytes([0xE2])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ota_path(self, client, tmp_path):
|
|
f = tmp_path / "fw.img"
|
|
f.write_bytes(b"\x70\x51" + b"\x00" * 100)
|
|
result = await client.ota(f.read_bytes())
|
|
assert result["bytes"] == 102
|
|
|
|
|
|
class TestFactories:
|
|
def test_via_local_returns_client(self):
|
|
c = PPClient.via_local(mac="AA:BB")
|
|
assert isinstance(c, PPClient)
|
|
|
|
def test_via_bridge_parses_host_port(self):
|
|
c = PPClient.via_bridge("192.168.61.35", mac="AA:BB")
|
|
assert c._t.host == "192.168.61.35"
|
|
assert c._t.port == 8971
|
|
c2 = PPClient.via_bridge("192.168.61.35:9999")
|
|
assert c2._t.port == 9999
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_bridge(self, monkeypatch):
|
|
from ppclock.transports import bridge as br
|
|
async def fake_scan(host, port, timeout=5.0):
|
|
return [{"name": "NRF-X", "address": "AA", "rssi": -1}]
|
|
monkeypatch.setattr(br, "scan", fake_scan)
|
|
devices = await PPClient.scan_devices("bridge://192.168.61.35")
|
|
assert devices[0]["name"] == "NRF-X"
|