zhou yang和Claude Opus 5.5
ca7435242f
原判据「窗口内 max − min < 5 mA」抓不到真卡死：10-04 第 1 路平在 520 mA 五个半小时， 偶发尖峰让 590 个窗口的 max − min 都在 17 mA 以上，一个都判不出来。 - 平线改成：p95 − p05 < flat_ua（30 mA）且中位数 ≥ flat_min_ua（100 mA，新增参数）。 下限是给空闲低功耗留的：睡着的手机平在 ~10 mA 是正常的。 - 过低不变（窗内最大值 < 30 mA）：睡着的手机每 25 s 被长轮询叫醒，窗内最大值都在 213 mA 以上。 - 真实遥测回放：两次真卡死（10-04 第 1 路、10-05 09:40 第 5 路平在 382 mA）都判出； 亮屏常醒 ~26 h × 4 路、空闲睡着 ~6 h × 5 路、试验 / OTA 期间零误判。 - 测试 +4（带尖峰的平线 / 睡着是健康的 / 睡下去再没醒 = 过低 / 低位平线不下结论），恢复页加「中位数不低于」。 依据和局限写在 recovery.py 模块头。已部署；五路的自动恢复开关仍然全关，还没真正触发过一次。 Co-Authored-By: Claude Opus 5.5 <[email protected]>
241 行
11 KiB
Python
241 行
11 KiB
Python
"""HTTP / WebSocket 接口：用 TestClient 起完整的应用，Hub 经进程内假 broker 连模拟板子。"""
|
||
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.hub import Hub
|
||
from prrack.recovery import RecoveryManager
|
||
from prrack.simboard import SimBoard
|
||
|
||
B = "b81f3fc9c8bc"
|
||
|
||
|
||
@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, present=(True, True, True, False, True))
|
||
dist = tmp_path / "dist"
|
||
(dist / "assets").mkdir(parents=True)
|
||
(dist / "index.html").write_text("<html>rack</html>", encoding="utf-8")
|
||
async def up():
|
||
await hub.start()
|
||
sim.start()
|
||
sim.tick(now_ms())
|
||
|
||
app = create_app(hub, rec, str(dist), startup=[up], shutdown=[hub.stop])
|
||
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, db
|
||
|
||
|
||
def test_boards_and_snapshot(env):
|
||
c, hub, sim, db = env
|
||
boards = c.get("/api/boards").json()
|
||
assert [b["id"] for b in boards] == [B]
|
||
assert boards[0]["online"] is True and len(boards[0]["phones"]) == 5
|
||
assert c.get("/api/boards/nope").status_code == 404
|
||
|
||
|
||
def test_command_records_the_portal_user(env):
|
||
c, hub, sim, db = env
|
||
r = c.post("/api/boards/%s/cmd" % B, json={"cmd": "power", "rail": 1, "on": False},
|
||
headers={"X-Auth-User": "zhouyang"})
|
||
assert r.status_code == 200 and r.json()["ok"] is True
|
||
assert sim.rails[0].en is False
|
||
a = c.get("/api/audit").json()[0]
|
||
assert a["user"] == "zhouyang" and a["request"]["cmd"] == "power" and a["ok"] == 1
|
||
# 没有门户头（服务器本机直接 curl）记作 local
|
||
c.post("/api/boards/%s/cmd" % B, json={"cmd": "power", "rail": 1, "on": True})
|
||
assert c.get("/api/audit").json()[0]["user"] == "local"
|
||
|
||
|
||
def test_bad_commands_never_reach_the_board(env):
|
||
c, hub, sim, db = env
|
||
n = len(sim.received)
|
||
for body in ({"cmd": "wifi", "ssid": "x"}, {"cmd": "set_voltage", "rail": 1, "mv": 4300},
|
||
{"cmd": "key", "rail": 1, "keys": "pwr", "ms": 60000}, {"cmd": "rm -rf"}):
|
||
r = c.post("/api/boards/%s/cmd" % B, json=body)
|
||
assert r.status_code == 400, body
|
||
assert len(sim.received) == n
|
||
|
||
|
||
def test_board_rejection_is_passed_through(env):
|
||
c, hub, sim, db = env
|
||
r = c.post("/api/boards/%s/cmd" % B, json={"cmd": "power_cycle", "rail": 4})
|
||
assert r.status_code == 200 and r.json()["err"] == "empty"
|
||
|
||
|
||
def test_timeout_is_504(env):
|
||
c, hub, sim, db = env
|
||
sim.drop_next_replies = 99
|
||
r = c.post("/api/boards/%s/cmd" % B, json={"cmd": "power", "rail": 2, "on": True})
|
||
assert r.status_code == 504
|
||
|
||
|
||
def test_phone_settings_toggle_auto_recovery(env):
|
||
c, hub, sim, db = env
|
||
r = c.patch("/api/boards/%s/rails/3" % B, json={"label": "A-03", "auto_recover": True})
|
||
assert r.status_code == 200
|
||
ph = [p for p in r.json()["phones"] if p["rail"] == 3][0]
|
||
assert ph["label"] == "A-03" and ph["auto_recover"] == 1
|
||
rec = c.get("/api/recovery").json()
|
||
assert rec["params"]["window_s"] == 300
|
||
r = c.put("/api/recovery/params", json={"window_s": 120, "flat_ua": -5, "bogus": 1})
|
||
assert r.json()["window_s"] == 120 and r.json()["flat_ua"] == 30000 # 负数不收
|
||
|
||
|
||
def test_history_after_telemetry(env):
|
||
c, hub, sim, db = env
|
||
for _ in range(3):
|
||
sim.tick(now_ms())
|
||
time.sleep(0.1)
|
||
hub.flush()
|
||
h = c.get("/api/boards/%s/rails/1/history" % B, params={"start": now_ms() - 60000}).json()
|
||
assert h["resolution"] == "raw" and len(h["points"]) >= 1
|
||
assert h["points"][0][2] is not None
|
||
|
||
|
||
def test_websocket_hello_and_live_tele(env):
|
||
c, hub, sim, db = env
|
||
with c.websocket_connect("/api/ws", headers={"X-Auth-User": "zhouyang"}) as ws:
|
||
hello = ws.receive_json()
|
||
assert hello["t"] == "hello" and hello["user"] == "zhouyang" and hello["boards"][0]["id"] == B
|
||
sim.tick(now_ms())
|
||
msg = ws.receive_json()
|
||
while msg["t"] != "tele":
|
||
msg = ws.receive_json()
|
||
assert msg["board"] == B and len(msg["status"]["rails"]) == 5
|
||
|
||
|
||
def test_spa_fallback_is_not_cached(env):
|
||
c, hub, sim, db = env
|
||
r = c.get("/some/deep/route")
|
||
assert r.status_code == 200 and "rack" in r.text
|
||
assert r.headers["cache-control"] == "no-cache"
|
||
|
||
|
||
def test_a_real_websocket_library_is_installed():
|
||
"""TestClient 自带 WS 实现，上面那条 WS 用例在没装 websockets 的机器上也是绿的；
|
||
而真 uvicorn 缺了它会把 /api/ws 当普通 GET（落到 SPA 兜底，回 200 HTML）。
|
||
deploy.sh 在服务器上跑这套测试，所以缺库会在换上去之前被拦住。"""
|
||
from prrack.main import ws_library_ok
|
||
assert ws_library_ok()
|
||
|
||
|
||
def test_board_history_raw_and_minute(env):
|
||
c, hub, sim, db = env
|
||
base = now_ms() // 60000 * 60000 - 3 * 60000
|
||
rows = [(base + i * 1000, B, 12000, 1500000, 35000 + i, 40000 - i, 300) for i in range(120)]
|
||
db.many("INSERT INTO board_tele VALUES (?,?,?,?,?,?,?)", rows)
|
||
h = c.get("/api/boards/%s/history" % B, params={"start": base, "end": base + 120000}).json()
|
||
assert h["resolution"] == "raw" and len(h["points"]) >= 100
|
||
p = h["points"][0]
|
||
assert p[1] == 35000 and p[4] == 40000 and p[7] == 300 and p[8] == 12000
|
||
# 分钟聚合：两个整分钟，每分钟 min/max 就是那一分钟里的首尾
|
||
assert db.downsample(base + 180000) > 0
|
||
m = db.board_history(B, base, base + 120000 + 1) # 跨度小也能强制走聚合：t0 早于原始保留期时
|
||
rows_min = db.q("SELECT * FROM board_tele_min WHERE board_id=? ORDER BY ts", (B,))
|
||
assert [r["t0_min"] for r in rows_min][:2] == [35000, 35060]
|
||
assert [r["t0_max"] for r in rows_min][:2] == [35059, 35119]
|
||
assert rows_min[0]["t1_min"] == 40000 - 59 and rows_min[0]["fan_avg"] == 300
|
||
assert m["resolution"] == "raw"
|
||
old = db.board_history(B, base - 10 * 86400000, base + 120000)
|
||
# 10 天跨度 ⇒ 桶约 9 分钟，两分钟落进同一桶：min 取两分钟的最小、max 取最大
|
||
assert old["resolution"] == "minute" and old["points"][0][2] == 35000
|
||
assert max(p[3] for p in old["points"]) == 35119
|
||
assert c.get("/api/boards/nope/history", params={"start": 0}).status_code == 404
|
||
|
||
|
||
def test_boot_check_event_is_named_and_only_the_failure_is_severe(env):
|
||
# 固件 0.1.2 LOG_EV_BOOT_CHECK：arg 低 4 位 = 按了几次，bit7 = 最终没开起来
|
||
c, hub, sim, db = env
|
||
now = now_ms()
|
||
db.many("INSERT INTO events (board_id, seq, t_ms, rail, type, arg, mv, ua, received) VALUES (?,?,?,?,?,?,?,?,?)",
|
||
[(B, 9001, 1000, 1, 19, 2, 3931, 210000, now),
|
||
(B, 9002, 2000, 1, 19, 0x83, 3931, 1500, now + 1)])
|
||
ev = {e["seq"]: e for e in c.get("/api/events", params={"board": B}).json()}
|
||
assert ev[9001]["code"] == "BOOT_CHECK" and ev[9001]["severe"] is False
|
||
assert "按了 2 次，开起来了" in ev[9001]["name"]
|
||
assert ev[9002]["severe"] is True and "按了 3 次，没开起来" in ev[9002]["name"]
|
||
|
||
|
||
def test_board_history_sums_rail_power(env):
|
||
# 五路主板合计功率：同一时刻各路 mv×ua 求和再按桶平均；没有主板遥测的时刻是 None
|
||
c, hub, sim, db = env
|
||
base = now_ms() // 60000 * 60000 - 3 * 60000
|
||
db.many("INSERT INTO board_tele VALUES (?,?,?,?,?,?,?)",
|
||
[(base + i * 1000, B, 12000, 1000000, 35000, 35000, 0) for i in range(120)])
|
||
# 前 60 s 有两路：4.0 V × 500 mA + 3.9 V × 1 A = 5.9 W；后 60 s 没有主板数据
|
||
db.many("INSERT INTO tele VALUES (?,?,?,?,?,?,?,?,?,?)",
|
||
[(base + i * 1000, B, r, mv, ua, 1, 1, 1, 1, 0)
|
||
for i in range(60) for r, mv, ua in ((1, 4000, 500000), (2, 3900, 1000000))])
|
||
h = db.board_history(B, base, base + 120000)
|
||
assert h["resolution"] == "raw"
|
||
assert {p[10] for p in h["points"] if p[0] < base + 60000} == {5900}
|
||
assert {p[10] for p in h["points"] if p[0] >= base + 60000} == {None}
|
||
assert db.downsample(base + 180000) > 0
|
||
old = db.board_history(B, base - 10 * 86400000, base + 120000)
|
||
assert old["resolution"] == "minute" and old["points"][0][10] == 5900
|
||
|
||
|
||
def test_flash_mode_round_trip_and_two_rail_limit(env):
|
||
c, hub, sim, db = env
|
||
post = lambda body: c.post("/api/boards/%s/cmd" % B, json=body).json()
|
||
for rail in (1, 2, 3):
|
||
assert post({"cmd": "power", "rail": rail, "on": True})["ok"] is True
|
||
r = post({"cmd": "flash_mode", "rail": 1, "on": True, "min": 10})
|
||
assert r["ok"] is True and r["left"] == 600
|
||
assert post({"cmd": "flash_mode", "rail": 2, "on": True})["ok"] is True
|
||
r = post({"cmd": "flash_mode", "rail": 3, "on": True})
|
||
assert r["ok"] is False and r["err"] == "state" # 第 3 路被板子拒
|
||
assert post({"cmd": "flash_mode", "rail": 1, "on": False})["left"] == 0
|
||
assert post({"cmd": "flash_mode", "rail": 3, "on": True})["ok"] is True # 腾出名额
|
||
st = post({"cmd": "status"})
|
||
assert [x["chg"] > 0 for x in st["rails"]][:3] == [False, True, True]
|
||
|
||
|
||
def test_flash_mode_extend_round_trip(env):
|
||
c, hub, sim, db = env
|
||
post = lambda body: c.post("/api/boards/%s/cmd" % B, json=body).json()
|
||
assert post({"cmd": "power", "rail": 1, "on": True})["ok"] is True
|
||
r = post({"cmd": "flash_mode", "rail": 1, "add": 30})
|
||
assert r["ok"] is False and r["err"] == "state" # 没开着不能加钟
|
||
assert post({"cmd": "flash_mode", "rail": 1, "on": True})["left"] == 3600 # 默认 1 小时
|
||
assert post({"cmd": "flash_mode", "rail": 1, "add": 30})["left"] == 5400
|
||
assert post({"cmd": "flash_mode", "rail": 1, "add": 120})["left"] == 7200 # 截断
|
||
now = 10 ** 13
|
||
db.many("INSERT INTO events (board_id, seq, t_ms, rail, type, arg, mv, ua, received) VALUES (?,?,?,?,?,?,?,?,?)",
|
||
[(B, 9101, 1, 1, 21, 0x80 | 30, 0, 0, now), (B, 9102, 2, 2, 21, 254, 0, 0, now)])
|
||
ev = {e["seq"]: e for e in c.get("/api/events", params={"board": B}).json()}
|
||
assert "加钟 30 分钟" in ev[9101]["name"]
|
||
assert "断电自动退出" in ev[9102]["name"] # 0.1.13：254 带 0x80 位，不能被当成「加钟 126 分钟」
|
||
|
||
|
||
def test_hub_round_trip_and_event_names(env):
|
||
# 固件 0.1.14：hub 命令经 /cmd 放行、回应带 hub 对象；事件 22 按 arg 命名、mV 列不当电压显示
|
||
c, hub, sim, db = env
|
||
post = lambda body: c.post("/api/boards/%s/cmd" % B, json=body).json()
|
||
r = post({"cmd": "hub", "mode": "on"})
|
||
assert r["ok"] is True and r["hub"]["mode"] == "on" and r["hub"]["on"] is True
|
||
r = post({"cmd": "hub", "mode": "auto", "min": 0})
|
||
assert r["hub"]["on"] is False and r["hub"]["why"] == "off"
|
||
assert post({"cmd": "status"})["hub"]["mode"] == "auto"
|
||
assert c.post("/api/boards/%s/cmd" % B, json={"cmd": "hub", "min": 999}).status_code == 400
|
||
now = 10 ** 13
|
||
db.many("INSERT INTO events (board_id, seq, t_ms, rail, type, arg, mv, ua, received) VALUES (?,?,?,?,?,?,?,?,?)",
|
||
[(B, 9201, 1, 0, 22, 1, 2, 0, now), (B, 9202, 2, 0, 22, 0, 0, 0, now),
|
||
(B, 9203, 3, 0, 22, 0x11, 7, 0, now)])
|
||
ev = {e["seq"]: e for e in c.get("/api/events", params={"board": B}).json()}
|
||
assert "唤醒：有手机开着 VBUS" in ev[9201]["name"] and ev[9201]["mv"] is None
|
||
assert "休眠" in ev[9202]["name"] and ev[9202]["severe"] is False
|
||
assert "模式改为常开" in ev[9203]["name"]
|