"""协议镜像 + Hub（经进程内假 broker 对接模拟板子）+ 自动恢复状态机。

和固件测试同一个口径：断言打在「板子那头真的收到 / 执行了什么」上（SimBoard.received、
它的 rails 状态），不看 Hub 自己回了什么 —— 重传换了 seq 的现象恰恰是 Hub 这边一切正常、
手机被按了两遍。"""
import asyncio
import json

import pytest

from prrack import protocol
from prrack.bus import MemBroker, MemBus, topic_matches
from prrack.db import Db
from prrack.hub import CmdTimeout, Hub
from prrack.recovery import GAVE_UP, HARD, IDLE, SOFT, Params, PhoneRecovery, RecoveryManager
from prrack.simboard import HUNG, NORMAL, OFF, SimBoard

B = "b81f3fc9c8bc"


def run(coro):
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(coro)
    finally:  # 收掉 Hub 的后台任务（flusher / maint），免得退出时刷一屏 pending 警告
        tasks = asyncio.all_tasks(loop)
        for t in tasks:
            t.cancel()
        loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))
        loop.close()


async def settle():
    for _ in range(5):
        await asyncio.sleep(0)


def rig(tmp_path, present=(True,) * 5, timeout=0.05):
    broker = MemBroker()
    db = Db(str(tmp_path / "t.db"))
    hub = Hub(db, MemBus(broker), attempt_timeout=timeout)
    sim = SimBoard(MemBus(broker), B, present=present)
    return hub, sim, db


# ---- 协议镜像 -----------------------------------------------------------------

def test_build_command_whitelist_and_ranges():
    assert protocol.build_command({"cmd": "key", "rail": 3, "keys": "vdn+pwr", "ms": 2000}) == \
        {"cmd": "key", "rail": 3, "keys": "vdn+pwr", "ms": 2000}
    bad = [
        {"cmd": "wifi", "ssid": "x"},                       # 只收 USB 的命令不给界面
        {"cmd": "events"},                                   # 服务器自己同步
        {"cmd": "set_voltage", "rail": 1, "mv": 4011},       # SR-13 之上
        {"cmd": "set_voltage", "rail": 1, "mv": "3900"},
        {"cmd": "key", "rail": 1, "keys": "pwr+home", "ms": 100},
        {"cmd": "key", "rail": 1, "keys": "pwr", "ms": 15001},
        {"cmd": "power", "rail": 6, "on": True},
        {"cmd": "power", "rail": True, "on": True},         # bool 不是 int
        {"cmd": "vbus", "rail": 1, "on": 1},
        {"cmd": "fan", "duty": 1001},
    ]
    for b in bad:
        with pytest.raises(protocol.CmdError):
            protocol.build_command(b)


def test_seq_is_strictly_increasing_and_in_range():
    t = [1700000000.0]
    g = protocol.SeqGen(clock=lambda: t[0])
    a = [g.next() for _ in range(5)]
    assert a == sorted(set(a)) and all(1 <= x <= 0xFFFFFFFF for x in a)
    t[0] -= 100  # 时钟往回跳也不许重复
    assert g.next() > a[-1] or g.next() == 1


def test_event_table_matches_firmware_numbers():
    # firmware/main/evlog.h 冻结的数字（抽几个钉住）
    assert protocol.EVENT_TYPES[3][0] == "OVER_VOLT"
    assert protocol.EVENT_TYPES[12][0] == "REBOOT"
    assert protocol.EVENT_TYPES[13][0] == "CMD_REJECT"
    assert protocol.EVENT_TYPES[16][0] == "PD_LIMIT"
    assert protocol.CMD_NAMES[9] == "power" and protocol.CMD_NAMES[12] == "net"
    assert protocol.EVENT_TYPES[18][0] == "OTA" and protocol.CMD_NAMES[13] == "ota"  # C2
    assert protocol.EVENT_TYPES[19][0] == "BOOT_CHECK" and protocol.BOOT_CHECK_FAILED == 0x80  # 0.1.2


def test_topic_matching():
    assert topic_matches("phonerail/+/rsp", "phonerail/%s/rsp" % B)
    assert not topic_matches("phonerail/+/rsp", "phonerail/%s/tele" % B)
    assert topic_matches("phonerail/#", "phonerail/x/y")
    assert protocol.parse_topic("phonerail/%s/tele" % B) == (B, "tele")
    assert protocol.parse_topic("evse/123/state") is None


# ---- Hub ---------------------------------------------------------------------

def test_tele_is_ingested_and_board_discovered(tmp_path):
    hub, sim, db = rig(tmp_path)

    async def go():
        await hub.start()
        sim.start()
        sim.tick(1000)
        await settle()
        hub.flush()

    run(go())
    snap = hub.snapshot(B)
    assert snap["online"] and snap["fw"] == "0.1.0-sim"
    assert snap["status"]["rails"][0]["en"] is True
    assert db.q("SELECT COUNT(*) c FROM tele")[0]["c"] == 5
    assert len(snap["phones"]) == 5 and snap["phones"][0]["auto_recover"] == 0


def test_command_round_trip_and_audit(tmp_path):
    hub, sim, db = rig(tmp_path)

    async def go():
        await hub.start()
        sim.start()
        return await hub.call(B, {"cmd": "power", "rail": 2, "on": False}, "alice")

    rsp = run(go())
    assert rsp["ok"] is True
    assert sim.rails[1].en is False
    row = db.q("SELECT * FROM audit")[0]
    assert row["user"] == "alice" and row["rail"] == 2 and row["ok"] == 1


def test_retransmit_reuses_the_same_seq_and_executes_once(tmp_path):
    """⭐ 回应丢了一次：Hub 用**同一个 seq** 重发，板子判重，只执行一次。"""
    hub, sim, db = rig(tmp_path)
    sim.drop_next_replies = 1

    async def go():
        await hub.start()
        sim.start()
        return await hub.call(B, {"cmd": "power_cycle", "rail": 1}, "bob")

    rsp = run(go())
    assert rsp["ok"] is True and rsp.get("dup") is True
    sent = [r for r in sim.received if r.get("cmd") == "power_cycle"]
    assert len(sent) == 2 and sent[0]["seq"] == sent[1]["seq"]
    assert [e[3] for e in sim.events].count(12) == 1  # 重启事件只记了一次


def test_timeout_is_audited_and_raised(tmp_path):
    hub, sim, db = rig(tmp_path)
    sim.drop_next_replies = 99

    async def go():
        await hub.start()
        sim.start()
        with pytest.raises(CmdTimeout):
            await hub.call(B, {"cmd": "power", "rail": 1, "on": True}, "carol")

    run(go())
    row = db.q("SELECT * FROM audit")[0]
    assert row["ok"] == 0 and json.loads(row["response"])["err"] == "timeout"
    assert len({r["seq"] for r in sim.received}) == 1  # 三次都是同一个 seq


def test_events_are_synced_incrementally(tmp_path):
    hub, sim, db = rig(tmp_path)
    for i in range(18):
        sim._ev(1, 8, i)  # 加上 BOOT、PD_LIMIT 共 20 条：要分 3 页

    async def go():
        await hub.start()
        sim.start()
        sim.tick(1000)
        for _ in range(20):
            await settle()
        sim._ev(2, 3, 0, 4200, 1)  # 又来一条过压
        sim.tick(2000)
        for _ in range(20):
            await settle()

    run(go())
    seqs = [r["seq"] for r in db.q("SELECT seq FROM events ORDER BY seq")]
    assert seqs == list(range(21))
    assert db.q("SELECT ev_cursor FROM boards")[0]["ev_cursor"] == 21


def test_empty_slot_reply_passes_through(tmp_path):
    hub, sim, db = rig(tmp_path, present=(True, True, True, False, True))

    async def go():
        await hub.start()
        sim.start()
        return await hub.call(B, {"cmd": "power", "rail": 4, "on": True}, "dave")

    assert run(go()) == {"seq": sim.received[-1]["seq"], "ok": False, "err": "empty"}


# ---- 自动恢复：状态机 ------------------------------------------------------------

P = Params(window_s=60, flat_ua=5000, low_ua=30000, settle_s=30, max_per_hour=3)


def feed(ph, t0, secs, ua_fn):
    out = []
    for s in range(secs):
        st, act = ph.feed(t0 + s * 1000, ua_fn(s), 3)
        if st or act:
            out.append((t0 + s * 1000, st, act))
    return out


def test_flat_current_escalates_soft_then_hard_then_gives_up():
    ph = PhoneRecovery(P)
    r = feed(ph, 0, 61, lambda s: 412000)
    assert r[-1][1] == SOFT and r[-1][2][1] == {"cmd": "key", "rail": 3, "keys": "pwr", "ms": 10000}
    t = r[-1][0]
    # settle 期间的样本一概不看（哪怕是平线）
    assert feed(ph, t + 1000, 29, lambda s: 412000) == []
    r = feed(ph, t + 30000, 62, lambda s: 412000)
    assert r[-1][1] == HARD and r[-1][2][1] == {"cmd": "power_cycle", "rail": 3}
    t = r[-1][0]
    r = feed(ph, t + 30000, 62, lambda s: 412000)
    assert r[-1][1] == GAVE_UP and r[-1][2] is None
    assert feed(ph, t + 200000, 100, lambda s: 412000) == []  # 放弃之后再也不动


def test_recovers_after_soft_when_current_moves_again():
    ph = PhoneRecovery(P)
    t = feed(ph, 0, 61, lambda s: 412000)[-1][0]
    r = feed(ph, t + 30000, 70, lambda s: 300000 + (s % 7) * 40000)
    assert r[-1][1] == IDLE and ph.reason == "recovered"


def test_low_current_is_detected_and_normal_is_not():
    ph = PhoneRecovery(P)
    assert feed(ph, 0, 200, lambda s: 300000 + (s % 5) * 50000) == []
    ph2 = PhoneRecovery(P)
    assert feed(ph2, 0, 61, lambda s: 800 + s % 3)[-1][2][0] == "low"


# 2026-10-06 按实测改的平线判据（recovery.py 模块头的表）：下面三条钉的是「改之前判错」的那几种形状。

def test_flat_tolerates_occasional_spikes():
    """真卡死（10-04 第 1 路）：平在 ~520 mA，偶尔冒一个 +150 mA 的尖峰。max − min 判据一个窗口都抓不到。"""
    ph = PhoneRecovery(Params(window_s=60, settle_s=30))
    r = feed(ph, 0, 61, lambda s: 520000 + (s % 3) * 4000 + (150000 if s % 25 == 0 else 0))
    assert r and r[-1][1] == SOFT and r[-1][2][0] == "flat"


def test_sleeping_phone_is_healthy():
    """空闲睡着：平在 ~10 mA，每 25 s 被长轮询叫醒一秒（~250 mA）。既不是平线（低）也不是过低（有尖峰）。"""
    ph = PhoneRecovery(Params(window_s=300, settle_s=30))
    assert feed(ph, 0, 900, lambda s: 250000 if s % 25 == 0 else 9000 + s % 4 * 1000) == []


def test_asleep_and_never_waking_is_low():
    """睡下去之后再也没醒（代理不轮询了）：5 分钟里一个尖峰都没有 ⇒ 过低。"""
    ph = PhoneRecovery(Params(window_s=300, settle_s=30))
    r = feed(ph, 0, 301, lambda s: 9000 + s % 4 * 1000)
    assert r and r[-1][2][0] == "low"


def test_flat_but_low_level_is_not_flat():
    """平在 60 mA（低于 flat_min_ua、高于 low_ua）：不下结论 —— 没见过卡死的手机掉到这么低。"""
    ph = PhoneRecovery(Params(window_s=60, settle_s=30))
    assert feed(ph, 0, 200, lambda s: 60000 + s % 3 * 1000) == []


def test_rate_limit_gives_up():
    ph = PhoneRecovery(Params(window_s=60, settle_s=30, max_per_hour=2))
    t = 0
    for _ in range(2):  # 两次软重启都「恢复」了，但每次很快又卡
        assert feed(ph, t, 65, lambda s: 412000)[-1][1] == SOFT
        t += 65000  # 下一段接着上一段喂完的时刻往后走（时间不倒退）
        # 前 30 s 还在 settle 里不算数，之后要攒满一个 60 s 窗口才判「恢复」
        assert feed(ph, t, 100, lambda s: 300000 + (s % 7) * 40000)[-1][1] == IDLE
        t += 100000
    r = feed(ph, t, 65, lambda s: 412000)
    assert r[-1][1] == GAVE_UP and ph.reason == "rate_limit"


# ---- 自动恢复：接到 Hub + 模拟板上 -------------------------------------------------

def test_auto_recovery_end_to_end_with_sim(tmp_path):
    """⭐ 模拟板上第 3 台卡死：开了自动恢复 → 5 分钟后下长按电源 → 板子执行 → 恢复。
    没开自动恢复的第 2 台（也卡死）一个命令都不收到。"""
    hub, sim, db = rig(tmp_path)
    clock = [1_000_000]
    rec = RecoveryManager(hub, lambda: clock[0])
    sim.set_phone(3, HUNG)
    sim.set_phone(2, HUNG)

    async def go():
        await hub.start()
        sim.start()
        await settle()  # 等 Hub 处理完上线消息、建好这块板的 phones 行
        db.x("UPDATE phones SET auto_recover=1 WHERE rail=3")
        rec.reload()
        for s in range(300 + 130 + 320):
            clock[0] += 1000
            sim.tick(clock[0])
            await settle()

    run(go())
    keys = [r for r in sim.received if r.get("cmd") == "key"]
    assert len(keys) == 1 and keys[0]["rail"] == 3 and keys[0]["ms"] == 10000
    assert not [r for r in sim.received if r.get("rail") == 2]
    assert sim.rails[2].phone == NORMAL
    states = [r["state"] for r in db.q("SELECT state FROM recovery_log ORDER BY id")]
    assert states == [SOFT, IDLE]
    assert db.q("SELECT user FROM audit")[0]["user"] == "auto-recovery"


def test_auto_recovery_ignores_rails_that_are_off(tmp_path):
    hub, sim, db = rig(tmp_path)
    clock = [0]
    rec = RecoveryManager(hub, lambda: clock[0])
    sim.set_phone(1, OFF)
    sim.rails[0].en = False  # 这一路本来就关着：电流 0 不是故障

    async def go():
        await hub.start()
        sim.start()
        await settle()
        db.x("UPDATE phones SET auto_recover=1")
        rec.reload()
        for s in range(400):
            clock[0] += 1000
            sim.tick(clock[0])
            await settle()

    run(go())
    assert not [r for r in sim.received if r.get("cmd") in ("key", "power_cycle")]


def test_flash_mode_command_and_event_numbers():
    # 固件 0.1.5：命令 14 / 事件 21（cmd.h / evlog.h 冻结编号）
    assert protocol.CMD_NAMES[14] == "flash_mode" and protocol.EVENT_TYPES[21][0] == "CHG_MODE"
    assert protocol.EVENT_TYPES[19][0] == "BOOT_CHECK" and protocol.EVENT_TYPES[20][0] == "BLADE_MUTE"
    assert protocol.build_command({"cmd": "flash_mode", "rail": 2, "on": True}) ==         {"cmd": "flash_mode", "rail": 2, "on": True}
    assert protocol.build_command({"cmd": "flash_mode", "rail": 2, "on": True, "min": 45})["min"] == 45
    assert "min" not in protocol.build_command({"cmd": "flash_mode", "rail": 2, "on": False, "min": 5})
    for bad in ({"cmd": "flash_mode", "rail": 2, "on": True, "min": 0},
                {"cmd": "flash_mode", "rail": 2, "on": True, "min": 121},
                {"cmd": "flash_mode", "rail": 2, "on": True, "min": True},
                {"cmd": "flash_mode", "rail": 6, "on": True},
                {"cmd": "flash_mode", "rail": 2}):
        with pytest.raises(protocol.CmdError):
            protocol.build_command(bad)


def test_hub_command_and_event_numbers():
    # 固件 0.1.14：命令 15 / 事件 22（cmd.h / evlog.h 冻结编号）
    assert protocol.CMD_NAMES[15] == "hub" and protocol.EVENT_TYPES[22][0] == "HUB"
    assert protocol.HUB_MODES == ("auto", "on") and protocol.HUB_WHY_NAMES[2] == "有手机开着 VBUS"
    assert protocol.build_command({"cmd": "hub", "mode": "on"}) == {"cmd": "hub", "mode": "on"}
    assert protocol.build_command({"cmd": "hub", "min": 0}) == {"cmd": "hub", "min": 0}
    assert protocol.build_command({"cmd": "hub", "mode": "auto", "min": 30}) ==         {"cmd": "hub", "mode": "auto", "min": 30}
    for bad in ({"cmd": "hub"}, {"cmd": "hub", "mode": "off"}, {"cmd": "hub", "min": 241},
                {"cmd": "hub", "min": -1}, {"cmd": "hub", "min": True}):
        with pytest.raises(protocol.CmdError):
            protocol.build_command(bad)


def test_flash_mode_extend_command():
    assert protocol.build_command({"cmd": "flash_mode", "rail": 1, "add": 30}) ==         {"cmd": "flash_mode", "rail": 1, "add": 30}
    for bad in ({"cmd": "flash_mode", "rail": 1, "add": 0},
                {"cmd": "flash_mode", "rail": 1, "add": 121},
                {"cmd": "flash_mode", "rail": 1, "add": 30, "on": True},
                {"cmd": "flash_mode", "rail": 1, "add": 30, "min": 10}):
        with pytest.raises(protocol.CmdError):
            protocol.build_command(bad)


# ---- 耗电量 -------------------------------------------------------------------

def test_energy_is_rolled_up_per_local_day_and_old_days_are_not_shrunk(tmp_path):
    import time
    db = Db(str(tmp_path / "e.db"))
    day_ms = lambda y, m, d, h=12: int(time.mktime((y, m, d, h, 0, 0, 0, 0, -1)) * 1000)
    d1, d2 = day_ms(2026, 10, 4), day_ms(2026, 10, 5)
    # 10-04：两个满分钟 12 V × 0.5 A = 6 W ⇒ 720 W·s；10-05：一个只有 30 个样本的分钟 12 W ⇒ 360 W·s
    db.many("INSERT INTO board_tele_min (ts, board_id, in_mv_avg, in_ua_avg, n) VALUES (?,?,?,?,?)",
            [(d1, B, 12000, 500000, 60), (d1 + 60000, B, 12000, 500000, 60), (d2, B, 12000, 1000000, 30),
             (d2 + 60000, B, None, None, 60)])  # 输入侧 INA226 没读到的那一分钟：不计，也不算时长
    db.many("INSERT INTO tele_min (ts, board_id, rail, mv_avg, ua_avg, ua_min, ua_max, n) VALUES (?,?,?,?,?,?,?,?)",
            [(d1, B, 2, 4000, 250000, 0, 0, 60)])  # 第 2 路 1 W × 60 s
    db.energy_rollup(days_back=1, now=d2)
    days = db.energy(B)
    assert [d["day"] for d in days] == ["2026-10-04", "2026-10-05"]
    assert (days[0]["in_mws"], days[0]["secs"]) == (720000, 120)
    assert (days[1]["in_mws"], days[1]["secs"]) == (360000, 30)
    assert days[0]["rails"][1] == {"mws": 60000, "secs": 60} and days[0]["rails"][0] is None
    assert db.energy_total(B)["mws"] == 1080000 and db.energy_total(B)["since"] == "2026-10-04"

    # 分钟数据被清掉一半之后再滚：窗口内的日子照实重算，窗口外的旧日子保持原值
    db.x("DELETE FROM board_tele_min WHERE ts=?", (d1,))
    db.energy_rollup(days_back=0, now=d2)
    assert db.energy(B)[0]["in_mws"] == 720000
    db.energy_rollup(days_back=1, now=d2)
    assert db.energy(B)[0]["in_mws"] == 360000
    assert len(db.energy(B, days=1)) == 1 and db.energy(B, days=1)[0]["day"] == "2026-10-05"


# ---- 手机自己报的那一半（fleet.py）---------------------------------------------

def test_fleet_phones_are_mapped_to_board_rails(tmp_path):
    from prrack.fleet import map_phones
    ph = lambda rail, **kw: dict({"rail": rail, "enabled": True, "online": True, "busy": False, "version": "0.8.0",
                                  "temp_board_mc": 28000, "temp_cpu_mc": 31000, "temp_at": 1700000000}, **kw)
    phones = [ph("B-1"), ph("B-3", busy=True), ph(""), ph("B-9"), ph("B-2", enabled=False),
              ph("B-5", temp_board_mc=None, temp_cpu_mc=None, temp_at=None)]
    one = map_phones(phones, [(B, "")])
    assert sorted(r for _, r in one) == [1, 3, 5]  # 没填机架位 / 路号越界 / 停用的不摆
    assert one[(B, 1)]["temp_board_mc"] == 28000 and one[(B, 1)]["temp_at"] == 1700000000000
    assert one[(B, 3)]["busy"] is True and one[(B, 5)]["temp_board_mc"] is None
    # 多块板：只认名字就是那个字母的；对不上不猜
    two = map_phones(phones + [ph("A-1")], [(B, "板 B"), ("aaaaaaaaaaaa", "")])
    assert sorted(two) == [(B, 1), (B, 3), (B, 5)]

    hub, sim, db = rig(tmp_path)
    got = []

    async def go():
        await hub.start()
        sim.start()
        sim.tick(1000)
        await settle()
        q = hub.subscribe_client()
        hub.set_phone_live(one)
        hub.set_phone_live(one)  # 没变：不再推
        while not q.empty():
            got.append(q.get_nowait())
        snap = hub.snapshot(B)
        assert [p["live"] and p["live"]["temp_cpu_mc"] for p in snap["phones"]] == [31000, None, 31000, None, None]
        hub.set_phone_live({})  # AppTask 读不到：清空，界面那一行消失
        assert all(p["live"] is None for p in hub.snapshot(B)["phones"])

    run(go())
    msgs = [m for m in got if m["t"] == "phones_live"]
    assert len(msgs) == 1 and set(msgs[0]["live"]) == {1, 3, 5}