zjk
54c32a180a
- 新增 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 补充编译产物/临时文件忽略
93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
"""协议层单元测试（无第三方依赖，直接运行）。
|
|
|
|
用法：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())
|