feat(bridge): BLE 桥接能力——bridge_server.py(Windows原生) + BridgeTransport + --via/PPCLOCK_VIA，149 测试全绿

tests/test_bridge.py

@@ -0,0 +1,151 @@
+"""BridgeTransport 测试 — 用内存假 bridge 服务端验证 JSONL 协议与同构接口。"""
+import asyncio
+import json
+
+import pytest
+import pytest_asyncio
+
+from ppclock 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"
+    assert w[1]["char"] == "rxtx" 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

tests/test_cli.py

@@ -231,3 +231,55 @@ class TestActivateAuto:
         expected = keygen_code(bytes([0x9A, 0x01, 0x99, 0x50, 0x10, 0x20]))
         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.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.transport 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.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.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)