303 lines
11 KiB
Python
303 lines
11 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 write_alt(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.transports.local 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 # 默认三色双平面
|
||
|
||
|
||
class TestBatch:
|
||
def test_batch_jsonl(self, capsys, tmp_path):
|
||
f = tmp_path / "job.jsonl"
|
||
f.write_text(
|
||
'{"argv": ["mode", "clock1"]}\n'
|
||
'{"argv": ["parking", "12"]}\n'
|
||
'{"argv": ["countdown"]}\n' # 故意错误行
|
||
)
|
||
code, out = run_cli(capsys, ["--json", "batch", str(f)])
|
||
assert code == 0
|
||
results = out["data"]["results"]
|
||
assert [r["ok"] for r in results] == [True, True, False]
|
||
# 单连接：只有1个 transport 实例
|
||
assert len(FakeTransport.instances) == 1
|
||
|
||
def test_batch_stop_on_error(self, capsys, tmp_path):
|
||
f = tmp_path / "job.jsonl"
|
||
f.write_text(
|
||
'{"argv": ["countdown"]}\n'
|
||
'{"argv": ["mode", "clock1"]}\n'
|
||
)
|
||
code, out = run_cli(capsys, ["--json", "batch", str(f), "--stop-on-error"])
|
||
results = out["data"]["results"]
|
||
assert len(results) == 1 and results[0]["ok"] is False
|
||
|
||
|
||
class TestOta:
|
||
def test_ota_precheck_without_yes(self, capsys):
|
||
code, out = run_cli(capsys, ["--json", "ota", "app/firmware-PP_da14585_4.2_CH.img"])
|
||
assert code == 0
|
||
assert out["data"]["precheck"] == "ok"
|
||
assert out["data"]["body_size"] == 72164
|
||
|
||
def test_ota_rejects_bad_file(self, capsys, tmp_path):
|
||
bad = tmp_path / "bad.img"
|
||
bad.write_bytes(b"\x00" * 128)
|
||
code, out = run_cli(capsys, ["--json", "ota", str(bad)])
|
||
assert code == 1
|
||
assert out["ok"] is False
|
||
|
||
|
||
class TestActivateAuto:
|
||
def test_activate_auto_via_device_id(self, capsys):
|
||
"""EFEF 应答 ID '81233F3C267112' → MAC 28:BF:5D:5A:BC:18 → keygen → EF 下发"""
|
||
class IdTransport(FakeTransport):
|
||
async def request_device_id(self, timeout=8.0):
|
||
self.rxtx_writes.append(bytes([0xEF, 0xEF]))
|
||
return "81233F3C267112"
|
||
FakeTransport.instances.clear()
|
||
import ppclock.cli as c
|
||
orig = c._make_transport
|
||
c._make_transport = lambda args: IdTransport(args.mac, args.timeout)
|
||
try:
|
||
code, out = run_cli(capsys, ["--json", "activate", "--auto"])
|
||
finally:
|
||
c._make_transport = orig
|
||
assert code == 0
|
||
t = IdTransport.instances[0]
|
||
assert t.rxtx_writes[0] == bytes([0xEF, 0xEF])
|
||
# keygen(9A,01,99,50,10,20)：逐字节 rol2(((m<<1)&FF)^EF)%101
|
||
from ppclock.protocol import keygen_code
|
||
expected = keygen_code(bytes([0x28, 0xBF, 0x5D, 0x5A, 0xBC, 0x18]))
|
||
assert t.rxtx_writes[1] == bytes([0xEF]) + expected
|
||
assert out["data"]["code"] == expected.hex()
|
||
|
||
|
||
_ORIG_MAKE = cli._make_transport
|
||
|
||
|
||
class TestViaBridge:
|
||
def test_via_bridge_creates_bridge_transport(self):
|
||
import argparse
|
||
from ppclock.transports.bridge import BridgeTransport
|
||
args = argparse.Namespace(mac="AA:BB", timeout=5.0, via="bridge://192.168.61.35:8971")
|
||
t = _ORIG_MAKE(args)
|
||
assert isinstance(t, BridgeTransport)
|
||
assert (t.host, t.port) == ("192.168.61.35", 8971)
|
||
assert t.address == "AA:BB"
|
||
|
||
def test_via_default_port(self):
|
||
import argparse
|
||
args = argparse.Namespace(mac=None, timeout=5.0, via="bridge://10.0.0.2")
|
||
t = _ORIG_MAKE(args)
|
||
assert t.port == 8971
|
||
|
||
def test_via_local_creates_ble(self):
|
||
import argparse
|
||
from ppclock.transports.local import BLETransport
|
||
args = argparse.Namespace(mac=None, timeout=5.0, via=None)
|
||
assert isinstance(_ORIG_MAKE(args), BLETransport)
|
||
|
||
def test_scan_uses_bridge(self, capsys, monkeypatch):
|
||
import ppclock.transports.bridge as br
|
||
called = {}
|
||
|
||
async def fake_scan(host, port, timeout=5.0):
|
||
called["args"] = (host, port)
|
||
return [{"name": "NRF-BR", "address": "CC:DD", "rssi": -60}]
|
||
monkeypatch.setattr(br, "scan", fake_scan)
|
||
code, out = run_cli(capsys, ["--json", "--via", "bridge://10.0.0.9:9999", "scan"])
|
||
assert code == 0
|
||
assert called["args"] == ("10.0.0.9", 9999)
|
||
assert out["data"]["devices"][0]["name"] == "NRF-BR"
|
||
|
||
def test_env_var_fallback(self, capsys, monkeypatch):
|
||
import ppclock.transports.bridge as br
|
||
called = {}
|
||
|
||
async def fake_scan(host, port, timeout=5.0):
|
||
called["args"] = (host, port)
|
||
return []
|
||
monkeypatch.setattr(br, "scan", fake_scan)
|
||
monkeypatch.setenv("PPCLOCK_VIA", "bridge://10.0.0.7")
|
||
code, _ = run_cli(capsys, ["--json", "scan"])
|
||
assert code == 0
|
||
assert called["args"] == ("10.0.0.7", 8971)
|
||
|
||
|
||
class TestTimeTz:
|
||
def test_tz_offset_applied(self, capsys, monkeypatch):
|
||
"""--tz 8 应写入北京墙钟（与系统 UTC 本地时相差 8h）"""
|
||
import datetime as _dt
|
||
code, out = run_cli(capsys, ["--json", "time", "--tz", "8"])
|
||
assert code == 0
|
||
sent = FakeTransport.instances[0].rxtx_writes[0]
|
||
# dd 帧的 unix 字段（本地墙钟按 UTC 计）反解小时
|
||
unix_local = int.from_bytes(sent[1:5], "big")
|
||
wall = _dt.datetime.fromtimestamp(unix_local, _dt.timezone.utc)
|
||
utc_now = _dt.datetime.now(_dt.timezone.utc)
|
||
assert abs((wall - utc_now).total_seconds() - 8 * 3600) < 120
|