zhou yang和Claude Opus 5.5
7bf30e3c38
- warm_esp.{h,c}：warm_restart() 记下开着的路（RTC_NOINIT，魔数 + 掩码 + 反码）→ gpio_hold 锁五根 EN → 重启；
目标镜像没有能力标记（第 288 字节 PRCAPS1，.rodata_custom_desc）就不锁 —— 老固件写不动锁住的 EN。
warm_take()：软件复位 + 标记对得上才接管，一次性；再读 EN 真电平核对
- bringup：adopt 掩码；rail_en_init_keep() 接管的路写高、五根一律放开 hold（崩溃时旧锁可能还在）；
刀片不在 / mux 失联的接管路立刻关（SLOT_EMPTY arg 1）；第二段已开着的路不重开、不等 PG
- pca9555_init_keep()：方向寄存器是本固件写的才留 VBUS（掉过电的片子锁存器是 0xFF），键 / EDL 清零
- HAL：hal_gpio_hold()；ESP 输出脚配 INPUT_OUTPUT（读得到真电平）；sim 给 hold 建模
- 事件 LOG_EV_OTA arg 7 = 接管；兜底定时器回滚故意不走接管
- 顶层 CMake：phonerail.h 进 CMAKE_CONFIGURE_DEPENDS（否则镜像头版本停在旧值）；-u g_phonerail_caps 防 gc
- test_bringup 加 6 例（接管 / 非计划复位全关 / 空槽 / mux 失联 / PCA 掉过电 / hold 冻结），全工程 448 例全绿
- rack：固件库识别能力标记，推送对话框按「支持 / 不支持重启不断电」给不同提示（已部署）
- 需求文档：SR-6 改为「非计划复位全关」+ 计划内重启例外；§10.9 新增「重启不断电」
- 板 B：台架镜像（空槽强制 CH1 在位，未提交）开 CH1 → OTA 带重启 → 事件 18/7 mv=1，CH1 一直开着；
现跑 0.1.0-c3。未测：示波器看重启期间 EN / 输出毛刺（要插刀片）
Co-Authored-By: Claude Opus 5.5 <[email protected]>
207 行
8.6 KiB
Python
207 行
8.6 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) # 回滚
|
|
sim.tick(now_ms())
|
|
assert wait_for(lambda: len([e for e in c.get("/api/events").json() if e["type"] == 18]) == 2)
|
|
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
|