feat(sdk): PPClient 门面 API(via_local/via_bridge/via_uri/scan_devices + 全设备操作)+ __version__ 0.1.0,165 测试全绿
This commit is contained in:
1 parent
a613c76114
commit
122023cdf4
3 files changed
+349
No files matched your search
@@ -0,0 +1,148 @@
|
||||
"""PPClient 门面 API 测试。"""
|
||||
import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from ppclock.client import PPClient
|
||||
|
||||
|
||||
class FakeTransport:
|
||||
def __init__(self):
|
||||
self.epd_writes = []
|
||||
self.rxtx_writes = []
|
||||
self.address = "AA:BB:CC:DD:EE:FF"
|
||||
|
||||
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_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"
|
||||
Reference in new issue
Block a user