zhou yang和Claude Opus 5.5
9858b6c98e
- 总览：每块板一块面板、五个刀片槽（状态灯 + DIN 读数 + 最近两分钟电流小曲线，平线 = 疑似卡死）；
只有异常 / 故障 / 执行中上色；顶上一句话概况，有问题直接列出是哪几路
- 单板页：新增「底板温度」（两路 NTC + 风扇占空比，上下两块共用时间轴，10 分钟 ~ 7 天）；
每路电流 / 电压去掉双 Y 轴，改上下两块，电压固定 3.5~4.1 V；名称 / 位置就地编辑
- 主题：theme.css 亮 / 暗两套变量（跟随系统，Element Plus 暗色变量同挂 html.dark），图表切换时重画；
读数用系统自带 DIN 系字体（不加载网络字体）；温度两色按本页亮暗底色跑过 dataviz validate_palette.js 全通过
- 后端：board_tele_min 分钟聚合（温度带 min/max，留 90 天）+ GET /api/boards/{id}/history（Db.board_history）
- 修一个偶发失败的测试（test_ota_events_are_named_by_arg：只推一次遥测，赶上 Hub 正在同步时会漏触发）；
后端 42 例本机 + 服务器全绿，已部署；线上已取到板 B 温度 24.4 / 26.0 °C
Co-Authored-By: Claude Opus 5.5 <[email protected]>
252 行
11 KiB
Python
252 行
11 KiB
Python
"""固件库 + 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() == []
|