153 lines
5.3 KiB
Python
153 lines
5.3 KiB
Python
"""BridgeTransport 测试 — 用内存假 bridge 服务端验证 JSONL 协议与同构接口。"""
|
|
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from ppclock.transports import bridge
|
|
|
|
|
|
class FakeBridgeServer:
|
|
"""模拟 tools/bridge_server.py 的 JSONL 协议。"""
|
|
|
|
def __init__(self):
|
|
self.received = []
|
|
self.device_id = "89794900980101"
|
|
|
|
async def start(self):
|
|
self.server = await asyncio.start_server(self._handle, "127.0.0.1", 0)
|
|
return self.server.sockets[0].getsockname()[1]
|
|
|
|
async def _handle(self, reader, writer):
|
|
async for line in reader:
|
|
req = json.loads(line)
|
|
self.received.append(req)
|
|
op, rid = req["op"], req["id"]
|
|
if op == "ping":
|
|
data = {"pong": True, "version": 1, "platform": "FakeOS"}
|
|
elif op == "scan":
|
|
data = {"devices": [{"name": "NRF-FAKE", "address": "AA:BB:CC:DD:EE:FF",
|
|
"rssi": -50}]}
|
|
elif op == "connect":
|
|
data = {"address": req["address"]}
|
|
elif op == "write":
|
|
data = {"sent": len(req["data"]) // 2}
|
|
elif op == "read":
|
|
data = {"data": "beef"}
|
|
elif op == "request_device_id":
|
|
data = {"device_id": self.device_id}
|
|
elif op == "ota":
|
|
for i in (1, 2):
|
|
writer.write((json.dumps({"event": "ota_progress", "block": i,
|
|
"total": 2}) + "\n").encode())
|
|
await writer.drain()
|
|
data = {"done": True}
|
|
elif op == "close":
|
|
data = {}
|
|
elif op == "fail_me":
|
|
writer.write((json.dumps({"id": rid, "ok": False,
|
|
"error": "RuntimeError: boom"}) + "\n").encode())
|
|
await writer.drain()
|
|
continue
|
|
else:
|
|
continue
|
|
writer.write((json.dumps({"id": rid, "ok": True, "data": data}) + "\n").encode())
|
|
await writer.drain()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def server():
|
|
s = FakeBridgeServer()
|
|
port = await s.start()
|
|
yield s, port
|
|
s.server.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_via_rpc(server):
|
|
s, port = server
|
|
devices = await bridge.scan("127.0.0.1", port)
|
|
assert devices[0]["name"] == "NRF-FAKE"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_autoscan(server):
|
|
"""无 address 时自动扫描选首台 NRF-（与 BLETransport 语义一致）"""
|
|
s, port = server
|
|
async with bridge.BridgeTransport(host="127.0.0.1", port=port) as t:
|
|
assert t.address == "AA:BB:CC:DD:EE:FF"
|
|
ops = [r["op"] for r in s.received]
|
|
assert ops[:2] == ["scan", "connect"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_explicit_address(server):
|
|
s, port = server
|
|
async with bridge.BridgeTransport(address="18:BC:5A:5D:BF:28",
|
|
host="127.0.0.1", port=port) as t:
|
|
assert t.address == "18:BC:5A:5D:BF:28"
|
|
conn = next(r for r in s.received if r["op"] == "connect")
|
|
assert conn["address"] == "18:BC:5A:5D:BF:28"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_channels_and_hex(server):
|
|
s, port = server
|
|
async with bridge.BridgeTransport(address="X", host="127.0.0.1", port=port) as t:
|
|
await t.write_epd(bytes([0x03, 0xFF]))
|
|
await t.write_rxtx(bytes([0xE2]), response=False)
|
|
w = [r for r in s.received if r["op"] == "write"]
|
|
assert w[0]["char"] == "epd" and w[0]["data"] == "03ff"
|
|
# RXTX 写走 alt(331f) 活通道（实机证据）
|
|
assert w[1]["char"] == "alt" and w[1]["data"] == "e2" and w[1]["response"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_rxtx(server):
|
|
s, port = server
|
|
async with bridge.BridgeTransport(address="X", host="127.0.0.1", port=port) as t:
|
|
assert await t.read_rxtx() == bytes([0xBE, 0xEF])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_request_device_id(server):
|
|
s, port = server
|
|
async with bridge.BridgeTransport(address="X", host="127.0.0.1", port=port) as t:
|
|
assert await t.request_device_id() == "89794900980101"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_ota_streams_progress(server):
|
|
s, port = server
|
|
async with bridge.BridgeTransport(address="X", host="127.0.0.1", port=port) as t:
|
|
events = []
|
|
result = await t.run_ota(b"\x70\x51" + b"\x00" * 100, on_progress=events.append)
|
|
assert result == {"done": True}
|
|
assert [e["block"] for e in events] == [1, 2]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_error_raises_bridge_error(server):
|
|
s, port = server
|
|
async with bridge.BridgeTransport(address="X", host="127.0.0.1", port=port) as t:
|
|
with pytest.raises(bridge.BridgeError, match="boom"):
|
|
await t._rpc("fail_me")
|
|
|
|
|
|
class TestParseVia:
|
|
def test_bridge_uri(self):
|
|
host, port = bridge.parse_via("bridge://192.168.61.35:8971")
|
|
assert (host, port) == ("192.168.61.35", 8971)
|
|
|
|
def test_default_port(self):
|
|
host, port = bridge.parse_via("bridge://192.168.61.35")
|
|
assert (host, port) == ("192.168.61.35", 8971)
|
|
|
|
def test_bad_scheme(self):
|
|
with pytest.raises(ValueError):
|
|
bridge.parse_via("tcp://x")
|
|
|
|
def test_local_is_none(self):
|
|
assert bridge.parse_via("local") is None
|