"""BridgeTransport 测试 — 用内存假 bridge 服务端验证 JSONL 协议与同构接口。"""
import asyncio
import json

import pytest
import pytest_asyncio

from ppclock.transports import bridge


class FakeBridgeServer:
    """模拟 ppclock.bridge_server 的 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