"""固件库 + OTA 推送（firmware.py / api.py 的 /api/firmware、/api/boards/{id}/ota）。

断言打在「板子那头真的收到了什么」上（SimBoard.received），和 test_core 同一个口径。"""
import hashlib
import os
import stat
import struct
import time

import pytest
from fastapi.testclient import TestClient

from prrack.api import create_app
from prrack.bus import MemBroker, MemBus
from prrack.db import Db, now_ms
from prrack.firmware import SLOT_BYTES, FwError, FwStore, parse_image
from prrack.hub import Hub
from prrack.recovery import RecoveryManager
from prrack.simboard import SimBoard

B = "b81f3fc9c8bc"
BASE = "http://home.example:7765/fw"


def image(version="0.1.0-c2", project="phonerail", size=4096, salt=b"", warm=False) -> bytes:
    """最小的「像 ESP app 镜像」：镜像头 24 B + 段头 8 B + esp_app_desc_t。"""
    hdr = bytes([0xE9, 3, 2, 0x4F]) + b"\0" * 20 + b"\0" * 8
    desc = struct.pack("<II8s", 0xABCD5432, 0, b"\0" * 8)
    desc += version.encode().ljust(32, b"\0") + project.encode().ljust(32, b"\0")
    desc += b"13:31:32".ljust(16, b"\0") + b"Sep 27 2026".ljust(16, b"\0") + b"v5.4".ljust(32, b"\0")
    desc = desc.ljust(256, b"\0")
    if warm:
        desc += b"PRCAPS1\0" + struct.pack("<I", 1)
    body = hdr + desc + salt
    return body + b"\xAA" * (size - len(body))


# ---- 镜像解析 ---------------------------------------------------------------

def test_parse_image_reads_the_app_descriptor():
    info = parse_image(image())
    assert info == {"version": "0.1.0-c2", "project": "phonerail", "built": "Sep 27 2026 13:31:32",
                    "idf": "v5.4", "warm": False}
    assert parse_image(image(warm=True))["warm"] is True


@pytest.mark.parametrize("data, word", [
    (image(project="vtcore"), "vtcore"),                 # 传错了别的项目的 .bin
    (b"\x7fELF" + image()[4:], "0xE9"),                  # .elf
    (image()[:32] + b"\0" * 4000, "app 描述"),            # bootloader / 分区表：没有 app 描述
    (b"", "大小"),
    (image(size=SLOT_BYTES + 1), "大小"),                 # 超过一个槽
], ids=["other-project", "elf", "no-app-desc", "empty", "too-big"])
def test_parse_image_refuses(data, word):
    with pytest.raises(FwError) as e:
        parse_image(data)
    assert word in str(e.value)


# ---- 存储 -------------------------------------------------------------------

def test_store_names_files_by_sha_prefix_and_is_idempotent(tmp_path):
    db = Db(str(tmp_path / "a.db"))
    fw = FwStore(db, str(tmp_path / "fw"), BASE)
    data = image()
    sha = hashlib.sha256(data).hexdigest()
    row = fw.add(data, "C:\\build\\phonerail.bin", "zhouyang")
    assert row["sha256"] == sha and row["size"] == len(data) and row["filename"] == "phonerail.bin"
    assert row["url"] == "%s/%s.bin" % (BASE, sha[:16]) and len(row["url"]) <= 64
    p = tmp_path / "fw" / (sha[:16] + ".bin")
    assert p.read_bytes() == data
    if os.name == "posix":
        assert stat.S_IMODE(p.stat().st_mode) == 0o644  # nginx 读得到
    assert not [f for f in os.listdir(tmp_path / "fw") if f.endswith(".tmp")]
    fw.add(data, "again.bin", "someone")  # 同一个镜像再传：不新增、不改上传人
    assert len(fw.list()) == 1 and fw.list()[0]["uploaded_by"] == "zhouyang"

    cmd = fw.command(sha, True)
    assert cmd == {"cmd": "ota", "url": row["url"], "sha256": sha, "size": len(data), "reboot": True}
    with pytest.raises(FwError):
        fw.command(sha, 1)  # reboot 是 bool，不是 0/1（板子也是这么判的）
    with pytest.raises(FwError):
        fw.command(sha.upper(), False)
    with pytest.raises(FwError):
        fw.command("0" * 64, False)

    os.remove(p)  # 文件被手动删了：推不出去（不然板子会拿到 404）
    with pytest.raises(FwError) as e:
        fw.command(sha, False)
    assert "重新上传" in str(e.value)
    assert fw.delete(sha) and fw.list() == [] and not fw.delete(sha)


def test_store_refuses_a_base_url_the_board_cannot_take(tmp_path):
    with pytest.raises(ValueError):
        FwStore(Db(str(tmp_path / "a.db")), str(tmp_path / "fw"), "http://" + "x" * 60 + "/fw")


# ---- 接口 + 模拟板 ------------------------------------------------------------

@pytest.fixture
def env(tmp_path):
    broker = MemBroker()
    db = Db(str(tmp_path / "a.db"))
    hub = Hub(db, MemBus(broker), attempt_timeout=0.2)
    rec = RecoveryManager(hub, now_ms)
    sim = SimBoard(MemBus(broker), B)
    fw = FwStore(db, str(tmp_path / "fw"), BASE)

    async def up():
        await hub.start()
        sim.start()
        sim.tick(now_ms())

    app = create_app(hub, rec, None, startup=[up], shutdown=[hub.stop], fw=fw)
    with TestClient(app) as c:
        deadline = time.time() + 2
        while B not in hub.latest and time.time() < deadline:
            time.sleep(0.01)
        yield c, hub, sim


def upload(c, data, name="phonerail.bin"):
    return c.post("/api/firmware?filename=" + name, content=data,
                  headers={"Content-Type": "application/octet-stream", "X-Auth-User": "zhouyang"})


def wait_for(fn, t=2.0):
    deadline = time.time() + t
    while time.time() < deadline:
        if fn():
            return True
        time.sleep(0.01)
    return False


def test_upload_and_list(env):
    c, hub, sim = env
    r = upload(c, image(warm=True))
    assert r.status_code == 200 and r.json()["version"] == "0.1.0-c2" and r.json()["warm"] is True
    assert r.json()["uploaded_by"] == "zhouyang"
    lst = c.get("/api/firmware").json()
    assert lst["slot"] == SLOT_BYTES and [x["sha256"] for x in lst["items"]] == [r.json()["sha256"]]
    bad = upload(c, image(project="evse"))
    assert bad.status_code == 400 and "evse" in bad.json()["detail"]


def test_push_sends_ota_and_the_board_downloads_it(env):
    c, hub, sim = env
    data = image(size=900000)
    sha = upload(c, data).json()["sha256"]
    r = c.post("/api/boards/%s/ota" % B, json={"sha256": sha}, headers={"X-Auth-User": "zhouyang"})
    assert r.status_code == 200 and r.json()["ok"] is True and r.json()["act"] == "start"
    got = sim.received[-1]
    assert got["cmd"] == "ota" and got["sha256"] == sha and got["size"] == len(data)
    assert got["reboot"] is False  # ⚠️ 默认不重启：重启 = 五台手机断电一次
    assert got["url"] == "%s/%s.bin" % (BASE, sha[:16])
    a = c.get("/api/audit").json()[0]
    assert a["user"] == "zhouyang" and a["request"]["cmd"] == "ota" and a["ok"] == 1

    # 下载中再推：板子回 busy
    r = c.post("/api/boards/%s/ota" % B, json={"sha256": sha, "reboot": True})
    assert r.json()["ok"] is False and r.json()["err"] == "busy"

    for _ in range(4):
        sim.tick(now_ms())
    assert wait_for(lambda: (hub.latest[B]["status"] or {}).get("ota", {}).get("st") == "ready")

    # 同一个镜像已就绪 + reboot：不重下，直接「重启」到新版本，然后自证健康
    r = c.post("/api/boards/%s/ota" % B, json={"sha256": sha, "reboot": True})
    assert r.json()["act"] == "reboot"
    sim.tick(now_ms())
    assert wait_for(lambda: hub.latest[B]["status"]["fw"] == "ota-" + sha[:8])
    sim.tick(now_ms())
    assert wait_for(lambda: hub.latest[B]["status"]["ota"]["pend"] is False)


def test_push_is_validated_before_it_reaches_the_board(env):
    c, hub, sim = env
    n = len(sim.received)
    sha = upload(c, image()).json()["sha256"]
    assert c.post("/api/boards/%s/ota" % B, json={"sha256": "0" * 64}).status_code == 400
    assert c.post("/api/boards/%s/ota" % B, json={"sha256": sha, "reboot": 1}).status_code == 400
    assert c.post("/api/boards/nope/ota", json={"sha256": sha}).status_code == 404
    # 通用命令口不放行 ota：URL / 哈希只能由固件库生成
    assert c.post("/api/boards/%s/cmd" % B, json={"cmd": "ota", "url": "http://x/a.bin"}).status_code == 400
    assert [m for m in sim.received[n:] if m.get("cmd") == "ota"] == []


def test_no_fw_dir_means_no_ota(tmp_path):
    db = Db(str(tmp_path / "a.db"))
    hub = Hub(db, MemBus(MemBroker()))
    app = create_app(hub, RecoveryManager(hub, now_ms), None)
    with TestClient(app) as c:
        assert c.get("/api/firmware").status_code == 503


def test_ota_events_are_named_by_arg(env):
    c, hub, sim = env
    sim._ev(0, 18, 3, 3, 996064)  # 失败：哈希不对
    sim._ev(0, 18, 5)             # 回滚

    # 每次检查前都推一条遥测：Hub 正在同步上一批事件（启动时的 BOOT）时会跳过这一条遥测带来的
    # 「有新事件」，只推一次的话偶尔再也没人触发同步（服务器上 3 次里红过 1 次）。真板 1 s 一条，没有这个问题。
    def synced():
        sim.tick(now_ms())
        return len([e for e in c.get("/api/events").json() if e["type"] == 18]) == 2
    assert wait_for(synced)
    ev = {e["arg"]: e for e in c.get("/api/events").json() if e["type"] == 18}
    assert ev[3]["name"] == "固件升级（失败）：哈希不对" and ev[3]["severe"] is True
    assert ev[5]["name"] == "固件升级（回滚）" and ev[5]["severe"] is True


# ---- 自动清理 -------------------------------------------------------------------

def test_prune_removes_old_orphans_and_stale_tmp_only(tmp_path):
    from prrack.firmware import ORPHAN_KEEP_S, TMP_KEEP_S
    db = Db(str(tmp_path / "a.db"))
    d = tmp_path / "fw"
    fw = FwStore(db, str(d), BASE)
    lib = fw.add(image(), "phonerail.bin", "zhouyang")  # 界面上传的：登记在库里
    now = time.time()

    def put(name, age_s):
        f = d / name
        f.write_bytes(b"x")
        os.utime(f, (now - age_s, now - age_s))
        return name

    old_orphan = put("0123456789abcdef.bin", ORPHAN_KEEP_S + 60)   # ota_push.py 传的，放了 7 天多
    new_orphan = put("fedcba9876543210.bin", ORPHAN_KEEP_S - 60)   # 还没到 7 天
    old_tmp = put(".0123456789abcdef.bin.tmp", TMP_KEEP_S + 60)    # 上传中断留下的
    new_tmp = put(".fedcba9876543210.bin.tmp", 60)                 # 可能正在传
    other = put("README.txt", ORPHAN_KEEP_S * 10)                  # 不是本模块的名字：不碰
    os.utime(d / (lib["sha256"][:16] + ".bin"), (now - ORPHAN_KEEP_S * 10,) * 2)  # 库里的再老也不删

    gone = fw.prune(now)
    assert sorted(gone) == sorted([old_orphan, old_tmp])
    left = set(os.listdir(d))
    assert {new_orphan, new_tmp, other, lib["sha256"][:16] + ".bin"} <= left
    assert old_orphan not in left and old_tmp not in left
    assert fw.get(lib["sha256"])["present"] is True
    assert fw.prune(now) == []  # 幂等


def test_prune_after_library_delete_is_a_no_op(tmp_path):
    db = Db(str(tmp_path / "a.db"))
    fw = FwStore(db, str(tmp_path / "fw"), BASE)
    sha = fw.add(image(), "a.bin", "u")["sha256"]
    fw.delete(sha)  # 界面删除：文件与登记一起没了
    assert fw.prune() == []