168 lines
6.1 KiB
Python
168 lines
6.1 KiB
Python
"""CLI 测试 — monkeypatch 传输工厂，验证 JSON 输出与退出码。"""
|
||
import json
|
||
|
||
import pytest
|
||
from PIL import Image
|
||
|
||
from ppclock import cli
|
||
|
||
|
||
class FakeTransport:
|
||
instances = []
|
||
|
||
def __init__(self, address=None, timeout=10.0):
|
||
self.epd_writes = []
|
||
self.rxtx_writes = []
|
||
self.address = address
|
||
FakeTransport.instances.append(self)
|
||
|
||
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 request_device_id(self, timeout=8.0):
|
||
return "DEVICE12345678"
|
||
|
||
async def delay(self, seconds):
|
||
pass
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def fake_transport(monkeypatch):
|
||
FakeTransport.instances.clear()
|
||
monkeypatch.setattr(cli, "_make_transport", lambda args: FakeTransport(args.mac, args.timeout))
|
||
yield FakeTransport
|
||
|
||
|
||
def run_cli(capsys, argv):
|
||
code = cli.main(argv)
|
||
out = capsys.readouterr().out.strip()
|
||
return code, json.loads(out) if out.startswith("{") else out
|
||
|
||
|
||
class TestJsonContract:
|
||
def test_mode_json(self, capsys):
|
||
code, out = run_cli(capsys, ["--json", "mode", "clock1"])
|
||
assert code == 0
|
||
assert out["ok"] is True and out["cmd"] == "mode"
|
||
t = FakeTransport.instances[0]
|
||
assert t.rxtx_writes == [bytes.fromhex("e10201"), bytes([0xE2])]
|
||
|
||
def test_time_json(self, capsys):
|
||
code, out = run_cli(capsys, ["--json", "time", "2026-07-24T15:30:00"])
|
||
assert code == 0 and out["ok"]
|
||
t = FakeTransport.instances[0]
|
||
assert t.rxtx_writes[0][0] == 0xDD
|
||
assert t.rxtx_writes[-1] == bytes([0xE2])
|
||
|
||
def test_error_json_contract(self, capsys):
|
||
"""错误 schema：ok=false + error 字段，退出码 1"""
|
||
code, out = run_cli(capsys, ["--json", "countdown"])
|
||
assert code == 1
|
||
assert out["ok"] is False and "error" in out
|
||
|
||
def test_mac_passthrough(self, capsys):
|
||
run_cli(capsys, ["--json", "--mac", "AA:BB:CC:DD:EE:FF", "clear"])
|
||
assert FakeTransport.instances[0].address == "AA:BB:CC:DD:EE:FF"
|
||
|
||
|
||
class TestCommands:
|
||
def test_clear(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "clear"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].epd_writes[-1] == bytes([0x01])
|
||
|
||
def test_image_upload(self, capsys, tmp_path):
|
||
p = tmp_path / "white.png"
|
||
Image.new("RGB", (400, 300), (255, 255, 255)).save(p)
|
||
code, out = run_cli(capsys, ["--json", "image", str(p), "--mono"])
|
||
assert code == 0
|
||
t = FakeTransport.instances[0]
|
||
assert len(t.epd_writes) == 63 + 1
|
||
assert t.epd_writes[-1] == bytes([0x01])
|
||
assert out["data"]["bytes_bw"] == 15000
|
||
|
||
def test_image_slot_and_color(self, capsys, tmp_path):
|
||
p = tmp_path / "red.png"
|
||
Image.new("RGB", (400, 300), (255, 0, 0)).save(p)
|
||
code, out = run_cli(capsys, ["--json", "image", str(p), "--slot", "2"])
|
||
assert code == 0
|
||
t = FakeTransport.instances[0]
|
||
assert t.rxtx_writes[0] == bytes([0xEA, 0x02])
|
||
assert out["data"]["bytes_red"] == 15000
|
||
|
||
def test_countdown(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "countdown", "2026-12-31"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].rxtx_writes[0] == bytes(
|
||
[0xFF, 0x00, 0x20, 0x26, 0x12, 0x31])
|
||
|
||
def test_countdown_off(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "countdown", "--off"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].rxtx_writes[0] == bytes([0xFA])
|
||
|
||
def test_parking(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "parking", "12"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].rxtx_writes[0] == bytes([0xEF, 0x11, 0x22])
|
||
|
||
def test_sleep(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "sleep", "--on", "--start", "23", "--end", "7"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].rxtx_writes[0] == bytes([0xFB, 0x01, 0x23, 0x07])
|
||
|
||
def test_rotation(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "rotation", "--count", "4"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].rxtx_writes[0] == bytes.fromhex("EA0003")
|
||
|
||
def test_lut(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "lut", "0x05"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].rxtx_writes[0] == bytes([0xE6, 0x05])
|
||
|
||
def test_activate_show_id(self, capsys):
|
||
code, out = run_cli(capsys, ["--json", "activate", "--show-id"])
|
||
assert code == 0
|
||
assert out["data"]["device_id"] == "DEVICE12345678"
|
||
|
||
def test_wifi(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "wifi", "MyAP", "pw", "--city", "101010100"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].rxtx_writes[0].startswith(b"WIFI_SSID:MyAP")
|
||
|
||
def test_raw(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "raw", "e2"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].rxtx_writes == [bytes([0xE2])]
|
||
|
||
def test_raw_epd_channel(self, capsys):
|
||
code, _ = run_cli(capsys, ["--json", "raw", "01", "--channel", "epd"])
|
||
assert code == 0
|
||
assert FakeTransport.instances[0].epd_writes == [bytes([0x01])]
|
||
|
||
def test_scan(self, capsys, monkeypatch):
|
||
import ppclock.transport as tr
|
||
async def fake_scan(timeout=5.0, name_prefix="NRF-"):
|
||
return [{"name": "NRF-TEST", "address": "AA:BB", "rssi": -50}]
|
||
monkeypatch.setattr(tr, "scan", fake_scan)
|
||
code, out = run_cli(capsys, ["--json", "scan"])
|
||
assert code == 0
|
||
assert out["data"]["devices"][0]["name"] == "NRF-TEST"
|
||
|
||
def test_template_custom(self, capsys):
|
||
code, out = run_cli(capsys, ["--json", "template", "custom",
|
||
"--data", '{"text": "你好"}'])
|
||
assert code == 0
|
||
t = FakeTransport.instances[0]
|
||
assert len(t.epd_writes) == 63 + 63 + 1 # 默认三色双平面
|