新增复合管控器(pCCU) + 惯导数据解析 + 设备控制页增强 + 0824协议文档归档
- 新增 pCCU 复合管控器程序(src/pCCU):FC/PM 双链路(UDP)、协议编解码器 (FcProtocol/PmProtocol/Frame/Message/MessageRegistry)、快照与落库(DbStore)、 内置 WebServer;挂入 src/CMakeLists.txt 与板卡 mission(h100.moos) - 惯导/深度数据:新增 uDevice_insData 消息解析,接入 UpperCommManager, pPowerMangerHost 注册监视,UpmsgTest 覆盖 - pPowerMangerHost 设备控制页增强(web/index.html) - 部署/运维脚本改为 4 个 systemd 服务(moosdb/pPowerManger/pPowerMangerHost/pCCU): deploy/clean-data/fetch-data 同步适配 - 协议文档:归档 0824 控制主机与复合管控器通信协议 xlsx 及关联 docx,移除旧 md - 测试:新增 pccu 单元/集成测试与 disSysTest Python 配电仿真 - 清理:移除未使用的 CCU 状态调试打印 coutMsg;.gitignore 补充编译产物/临时文件忽略
This commit is contained in:
@@ -0,0 +1,92 @@
|
||||
"""协议层单元测试(无第三方依赖,直接运行)。
|
||||
|
||||
用法:python3 tests/test_protocol.py
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from app import protocol # noqa: E402
|
||||
from app.protocol import (MsgDisHighVolBusFb, MsgDisHighAVolBusFb, # noqa: E402
|
||||
MsgDisLowMainBusFb, MsgDisLowBusFb,
|
||||
MsgDisHVBCmd, MsgDisHVBACmd, MsgDisLVMBCmd,
|
||||
MsgDisLVBCmd)
|
||||
|
||||
|
||||
def test_expected_sizes():
|
||||
# 与 C++ pmSysvariable.h 结构体大小对齐
|
||||
expected = {
|
||||
MsgDisHighVolBusFb: 53,
|
||||
MsgDisHighAVolBusFb: 25,
|
||||
MsgDisLowMainBusFb: 34,
|
||||
MsgDisLowBusFb: 29,
|
||||
MsgDisHVBCmd: 14,
|
||||
MsgDisHVBACmd: 6,
|
||||
MsgDisLVMBCmd: 7,
|
||||
MsgDisLVBCmd: 8,
|
||||
}
|
||||
for cls, size in expected.items():
|
||||
assert cls.body_size() == size, f"{cls.__name__} 应为 {size},实际 {cls.body_size()}"
|
||||
|
||||
|
||||
def test_roundtrip():
|
||||
for cls in (MsgDisHighVolBusFb, MsgDisHighAVolBusFb, MsgDisLowMainBusFb,
|
||||
MsgDisLowBusFb, MsgDisHVBCmd, MsgDisHVBACmd, MsgDisLVMBCmd,
|
||||
MsgDisLVBCmd):
|
||||
m = cls()
|
||||
for i, name in enumerate(m.field_names()):
|
||||
setattr(m, name, (i + 1) % 256)
|
||||
data = m.pack()
|
||||
assert len(data) == 6 + cls.body_size() + 1
|
||||
assert protocol.msg_id_of(data) == cls.MSG_ID
|
||||
m2 = cls()
|
||||
assert m2.unpack(data)
|
||||
for name in m.field_names():
|
||||
assert getattr(m2, name) == getattr(m, name), f"{cls.__name__}.{name}"
|
||||
|
||||
|
||||
def test_checksum_reject():
|
||||
for cls in (MsgDisHighVolBusFb, MsgDisHVBACmd):
|
||||
data = bytearray(cls().pack())
|
||||
data[6] ^= 0xFF
|
||||
assert not cls().unpack(bytes(data))
|
||||
|
||||
|
||||
def test_bad_start_byte_reject():
|
||||
data = bytearray(MsgDisHVBCmd().pack())
|
||||
data[0] = 0x00
|
||||
assert protocol.msg_id_of(bytes(data)) is None
|
||||
assert not MsgDisHVBCmd().unpack(bytes(data))
|
||||
|
||||
|
||||
def test_unknown_id():
|
||||
data = protocol.MsgDisHVBCmd().pack()
|
||||
data = data[0:2] + b'\x00\x00' + data[4:] # 把 id 改为 0x0000
|
||||
assert protocol.msg_id_of(data) == 0x0000
|
||||
assert protocol.make_message(0x0000) is None
|
||||
assert protocol.make_message(MsgDisHVBCmd.MSG_ID).MSG_ID == MsgDisHVBCmd.MSG_ID
|
||||
|
||||
|
||||
def test_short_frame_reject():
|
||||
data = MsgDisHVBCmd().pack()[:10]
|
||||
assert not MsgDisHVBCmd().unpack(data)
|
||||
|
||||
|
||||
def run_all():
|
||||
tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
|
||||
failed = 0
|
||||
for t in tests:
|
||||
try:
|
||||
t()
|
||||
print(f" [OK] {t.__name__}")
|
||||
except Exception as e:
|
||||
failed += 1
|
||||
print(f" [FAIL] {t.__name__}: {e}")
|
||||
print("TEST:", "PASS" if failed == 0 else f"FAIL ({failed})")
|
||||
return 1 if failed else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(run_all())
|
||||
Reference in New Issue
Block a user