wangfq
6c1f276002
- protocol.py: OTA 模块 — 命令常量/ota_crc32(ISO-HDLC)/ota_split_bin(256B×96KB)/ data_ota_* 构建器/状态·阶段·错误码描述 - main.py: OTA 页签 — bin 选择(≤96KB+CRC32 显示)/版本·force·slot 参数/ 下载(OtaDownloadThread: begin→data×N→end, 断点续传·缺片重定位·CRC 重发)/ 刷写(二次确认)/中止/状态查询; 进度条 + ota_report/ota_status 实时显示 - tests/test_ota_protocol.py: 22 断言 — CRC32 标准向量/分片边界/构建器/ OtaDeviceSim 设备状态机(顺序·幂等·乱序·单片CRC·全镜像复核·续传)/完整下载流 - devlog 置顶条目
317 lines
13 KiB
Python
317 lines
13 KiB
Python
"""
|
|
DLD960 IoT MQTT OTA 协议单测 (V1.08)
|
|
|
|
验证对象: dbn_mqtt_tool/protocol.py 的 OTA 模块
|
|
- ota_crc32: CRC-32/ISO-HDLC 标准向量
|
|
- ota_split_bin: 分片边界 (256B 协议常量, 96KB 上限)
|
|
- data_ota_*: 命令构建器字段
|
|
- OtaDeviceSim: 设备侧接收状态机模拟 (协议 §4.19~4.21 规则)
|
|
- 完整下载流: 正常/断点续传/缺片重定位/CRC 错重发/重复片幂等
|
|
|
|
运行: venv/bin/python -m unittest tests.test_ota_protocol -v
|
|
"""
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from dbn_mqtt_tool.protocol import (
|
|
ota_crc32, ota_split_bin, OTA_CHUNK_SIZE, OTA_MAX_SIZE,
|
|
data_ota_begin, data_ota_data, data_ota_end, data_ota_abort, data_ota_flash,
|
|
OTA_TARGET_LOOP, OTA_SLOT_A,
|
|
)
|
|
|
|
|
|
class OtaDeviceSim:
|
|
"""模拟固件 OTA 接收状态机 (协议 §4.19 ota_begin / §4.20 ota_data / §4.21 ota_end)
|
|
|
|
规则 (与协议文档一致):
|
|
- 顺序片 (offset == received): 单片 CRC32 校验 → 写暂存 → received += 256
|
|
- 重复片 (offset < received): 幂等回 code=0, 不重写
|
|
- 缺片 (offset > received): code=1, data.offset=received (要求续传)
|
|
- 单片 CRC 失败: code=1 (平台重发本片)
|
|
- ota_end: received!=size → code=1; 全镜像 CRC 复核 → ready
|
|
"""
|
|
|
|
def __init__(self, bin_data: bytes):
|
|
self.bin = bin_data
|
|
self.size = len(bin_data)
|
|
self.state = "idle"
|
|
self.received = 0
|
|
self.storage = bytearray(self.size)
|
|
|
|
def _base(self, offset: int) -> dict:
|
|
return {"target": OTA_TARGET_LOOP, "slot": OTA_SLOT_A,
|
|
"offset": offset, "received": self.received,
|
|
"size": self.size, "crc32": ota_crc32(self.bin),
|
|
"state": self.state}
|
|
|
|
def ota_begin(self, req: dict) -> dict:
|
|
# 镜像一致且已 ready → 直接返回可刷写
|
|
if self.state == "ready" and req["size"] == self.size \
|
|
and req["crc32"] == ota_crc32(self.bin):
|
|
return {"code": 0, "data": self._base(self.size)}
|
|
# downloading 中断 → 返回已接收偏移 (断点续传)
|
|
if self.state == "downloading":
|
|
return {"code": 0, "data": self._base(self.received)}
|
|
self.state = "downloading"
|
|
self.received = 0
|
|
self.storage = bytearray(self.size)
|
|
return {"code": 0, "data": self._base(0)}
|
|
|
|
def ota_data(self, req: dict) -> dict:
|
|
offset, crc, hexs = req["offset"], req["crc32"], req["data"]
|
|
data = bytes.fromhex(hexs)
|
|
if offset == self.received:
|
|
if ota_crc32(data) != crc:
|
|
return {"code": 1, "err_code": 1, "msg": "crc mismatch",
|
|
"data": self._base(self.received)}
|
|
self.storage[offset:offset + len(data)] = data
|
|
self.received = min(offset + len(data), self.size)
|
|
return {"code": 0, "data": self._base(offset)}
|
|
if offset < self.received:
|
|
return {"code": 0, "data": self._base(offset)} # 幂等
|
|
return {"code": 1, "err_code": 1, "msg": "gap/out-of-order",
|
|
"data": self._base(self.received)}
|
|
|
|
def ota_end(self, req: dict) -> dict:
|
|
if self.received != self.size:
|
|
return {"code": 1, "data": self._base(self.received)}
|
|
if ota_crc32(bytes(self.storage)) != req["crc32"]:
|
|
return {"code": 5, "err_code": 5, "data": {"crc_ok": False}}
|
|
self.state = "ready"
|
|
return {"code": 0, "data": {"crc_ok": True}}
|
|
|
|
|
|
class TestOtaCrc32(unittest.TestCase):
|
|
def test_standard_vector(self):
|
|
"""CRC-32/ISO-HDLC 标准校验向量: crc32('123456789') = 0xCBF43926"""
|
|
self.assertEqual(ota_crc32(b"123456789"), 0xCBF43926)
|
|
|
|
def test_empty(self):
|
|
self.assertEqual(ota_crc32(b""), 0x00000000)
|
|
|
|
def test_zlib_alias(self):
|
|
import zlib
|
|
for data in (b"", b"a", os.urandom(300)):
|
|
self.assertEqual(ota_crc32(data), zlib.crc32(data) & 0xFFFFFFFF)
|
|
|
|
|
|
class TestOtaSplitBin(unittest.TestCase):
|
|
def test_small_single_chunk(self):
|
|
data = b"\x00\x01\x02\x03"
|
|
chunks = ota_split_bin(data)
|
|
self.assertEqual(len(chunks), 1)
|
|
off, crc, hexs = chunks[0]
|
|
self.assertEqual(off, 0)
|
|
self.assertEqual(hexs, data.hex())
|
|
self.assertEqual(len(hexs), 8) # 4B → 8 hex
|
|
self.assertEqual(crc, ota_crc32(data))
|
|
|
|
def test_exact_chunk(self):
|
|
data = bytes(range(256))
|
|
chunks = ota_split_bin(data)
|
|
self.assertEqual(len(chunks), 1)
|
|
self.assertEqual(len(chunks[0][2]), 512) # 256B → 512 hex
|
|
|
|
def test_two_chunks(self):
|
|
data = bytes(range(256)) + b"\x00" # 257B → 2 片
|
|
chunks = ota_split_bin(data)
|
|
self.assertEqual(len(chunks), 2)
|
|
self.assertEqual(chunks[0][0], 0)
|
|
self.assertEqual(chunks[1][0], 256)
|
|
self.assertEqual(len(chunks[1][2]), 2) # 末片 1B
|
|
|
|
def test_offsets_256_aligned(self):
|
|
data = os.urandom(5000)
|
|
for off, _, _ in ota_split_bin(data):
|
|
self.assertEqual(off % OTA_CHUNK_SIZE, 0)
|
|
|
|
def test_roundtrip_reconstruct(self):
|
|
data = os.urandom(4097)
|
|
merged = b"".join(bytes.fromhex(hexs) for _, _, hexs in ota_split_bin(data))
|
|
self.assertEqual(merged, data)
|
|
|
|
def test_over_max_size(self):
|
|
with self.assertRaises(ValueError):
|
|
ota_split_bin(os.urandom(OTA_MAX_SIZE + 1))
|
|
|
|
|
|
class TestOtaBuilders(unittest.TestCase):
|
|
def test_begin(self):
|
|
d = data_ota_begin(46864, 0x12345678, version="1.1.0")
|
|
self.assertEqual(d, {"target": "loop", "size": 46864, "crc32": 0x12345678,
|
|
"version": "1.1.0", "force": False})
|
|
|
|
def test_begin_force(self):
|
|
d = data_ota_begin(100, 1, force=True)
|
|
self.assertTrue(d["force"])
|
|
|
|
def test_data(self):
|
|
d = data_ota_data(0, 0x9E3779B9, "abcd")
|
|
self.assertEqual(d, {"target": "loop", "offset": 0,
|
|
"crc32": 0x9E3779B9, "data": "abcd"})
|
|
|
|
def test_end_abort_flash(self):
|
|
self.assertEqual(data_ota_end(0x11223344),
|
|
{"target": "loop", "crc32": 0x11223344})
|
|
self.assertEqual(data_ota_abort(), {"target": "loop"})
|
|
self.assertEqual(data_ota_flash(slot="b", force=True),
|
|
{"target": "loop", "slot": "b", "force": True})
|
|
|
|
|
|
class TestDeviceSim(unittest.TestCase):
|
|
"""设备侧规则: 顺序/幂等/乱序/CRC/全镜像复核"""
|
|
|
|
def setUp(self):
|
|
self.bin_data = bytes(range(256)) * 20 # 5120B = 20 片
|
|
self.dev = OtaDeviceSim(self.bin_data)
|
|
|
|
def test_normal_flow(self):
|
|
dev = self.dev
|
|
chunks = ota_split_bin(self.bin_data)
|
|
resp = dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data)))
|
|
self.assertEqual(resp["code"], 0)
|
|
self.assertEqual(resp["data"]["offset"], 0)
|
|
for off, crc, hexs in chunks:
|
|
resp = dev.ota_data(data_ota_data(off, crc, hexs))
|
|
self.assertEqual(resp["code"], 0, f"片@{off} 应成功")
|
|
self.assertEqual(dev.received, len(self.bin_data))
|
|
resp = dev.ota_end(data_ota_end(ota_crc32(self.bin_data)))
|
|
self.assertEqual(resp["code"], 0)
|
|
self.assertTrue(resp["data"]["crc_ok"])
|
|
self.assertEqual(dev.state, "ready")
|
|
self.assertEqual(bytes(dev.storage), self.bin_data)
|
|
|
|
def test_resume_after_begin(self):
|
|
"""断点续传: downloading 中断后 begin 返回已接收偏移"""
|
|
dev = self.dev
|
|
chunks = ota_split_bin(self.bin_data)
|
|
dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data)))
|
|
for off, crc, hexs in chunks[:7]: # 只收 7 片
|
|
dev.ota_data(data_ota_data(off, crc, hexs))
|
|
self.assertEqual(dev.received, 7 * OTA_CHUNK_SIZE)
|
|
# 重新 begin → 续传点
|
|
resp = dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data)))
|
|
self.assertEqual(resp["data"]["offset"], 7 * OTA_CHUNK_SIZE)
|
|
# 从续传点继续
|
|
for off, crc, hexs in chunks[7:]:
|
|
dev.ota_data(data_ota_data(off, crc, hexs))
|
|
self.assertEqual(dev.received, len(self.bin_data))
|
|
resp = dev.ota_end(data_ota_end(ota_crc32(self.bin_data)))
|
|
self.assertTrue(resp["data"]["crc_ok"])
|
|
|
|
def test_dup_chunk_idempotent(self):
|
|
"""重复片幂等: 已收片重发 → code=0, received 不变"""
|
|
dev = self.dev
|
|
chunks = ota_split_bin(self.bin_data)
|
|
dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data)))
|
|
off0, crc0, hexs0 = chunks[0]
|
|
dev.ota_data(data_ota_data(off0, crc0, hexs0))
|
|
r1 = dev.ota_data(data_ota_data(off0, crc0, hexs0)) # 重复
|
|
self.assertEqual(r1["code"], 0)
|
|
self.assertEqual(r1["data"]["received"], OTA_CHUNK_SIZE)
|
|
|
|
def test_gap_rejected(self):
|
|
"""缺片: offset > received → code=1, data.offset=received"""
|
|
dev = self.dev
|
|
chunks = ota_split_bin(self.bin_data)
|
|
dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data)))
|
|
off2, crc2, hexs2 = chunks[2] # 跳过 0,1
|
|
resp = dev.ota_data(data_ota_data(off2, crc2, hexs2))
|
|
self.assertEqual(resp["code"], 1)
|
|
self.assertEqual(resp["data"]["offset"], 0) # 要求从 0 续传
|
|
|
|
def test_chunk_crc_error(self):
|
|
"""单片 CRC 错 → code=1, 不落盘"""
|
|
dev = self.dev
|
|
chunks = ota_split_bin(self.bin_data)
|
|
dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data)))
|
|
off0, _, hexs0 = chunks[0]
|
|
resp = dev.ota_data(data_ota_data(off0, 0xDEADBEEF, hexs0)) # 错 crc
|
|
self.assertEqual(resp["code"], 1)
|
|
self.assertEqual(dev.received, 0) # 未落盘
|
|
# 重发正确片 → 成功
|
|
resp = dev.ota_data(data_ota_data(off0, ota_crc32(self.bin_data[:256]), hexs0))
|
|
self.assertEqual(resp["code"], 0)
|
|
|
|
def test_end_incomplete(self):
|
|
"""ota_end 时 received != size → code=1"""
|
|
dev = self.dev
|
|
dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data)))
|
|
resp = dev.ota_end(data_ota_end(ota_crc32(self.bin_data)))
|
|
self.assertEqual(resp["code"], 1)
|
|
self.assertEqual(resp["data"]["offset"], 0)
|
|
|
|
def test_end_crc_fail(self):
|
|
"""全镜像 CRC 复核失败 → code=5 crc_ok=false"""
|
|
dev = self.dev
|
|
chunks = ota_split_bin(self.bin_data)
|
|
dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data)))
|
|
for off, crc, hexs in chunks:
|
|
dev.ota_data(data_ota_data(off, crc, hexs))
|
|
resp = dev.ota_end(data_ota_end(0x00000000)) # 错的全镜像 crc
|
|
self.assertEqual(resp["code"], 5)
|
|
self.assertFalse(resp["data"]["crc_ok"])
|
|
|
|
|
|
class TestDownloadFlow(unittest.TestCase):
|
|
"""工具侧下载流程 (等价 OtaDownloadThread 核心逻辑, 无 Qt 依赖)"""
|
|
|
|
def _run_flow(self, bin_data: bytes, simulate_gap: bool = False):
|
|
dev = OtaDeviceSim(bin_data)
|
|
chunks = ota_split_bin(bin_data)
|
|
total_crc = ota_crc32(bin_data)
|
|
|
|
# ota_begin
|
|
resp = dev.ota_begin(data_ota_begin(len(bin_data), total_crc))
|
|
assert resp["code"] == 0
|
|
offset = resp["data"]["offset"]
|
|
|
|
sent_offsets = []
|
|
gap_done = False
|
|
while offset < len(bin_data):
|
|
idx = offset // OTA_CHUNK_SIZE
|
|
off, crc, hexs = chunks[idx]
|
|
if simulate_gap and not gap_done and off == 256:
|
|
# 模拟一次丢包: 跳过该片直接发下一片 → 设备回 gap → 重定位
|
|
gap_done = True
|
|
resp = dev.ota_data(data_ota_data(chunks[idx + 1][0],
|
|
chunks[idx + 1][1],
|
|
chunks[idx + 1][2]))
|
|
self.assertEqual(resp["code"], 1)
|
|
offset = resp["data"]["offset"] # 重定位
|
|
continue
|
|
resp = dev.ota_data(data_ota_data(off, crc, hexs))
|
|
if resp["code"] == 0:
|
|
offset = resp["data"]["received"]
|
|
elif resp["code"] == 1:
|
|
offset = resp["data"]["offset"] # 重定位
|
|
else:
|
|
self.fail(f"意外 code={resp['code']}")
|
|
sent_offsets.append(off)
|
|
|
|
# ota_end
|
|
resp = dev.ota_end(data_ota_end(total_crc))
|
|
self.assertEqual(resp["code"], 0)
|
|
self.assertTrue(resp["data"]["crc_ok"])
|
|
self.assertEqual(dev.state, "ready")
|
|
self.assertEqual(bytes(dev.storage), bin_data)
|
|
return sent_offsets
|
|
|
|
def test_flow_normal(self):
|
|
bin_data = os.urandom(3000)
|
|
self._run_flow(bin_data)
|
|
|
|
def test_flow_with_gap_retransmit(self):
|
|
"""缺片重传: 丢 1 片后设备要求重定位, 工具补发, 最终全量一致"""
|
|
bin_data = os.urandom(3000)
|
|
sent = self._run_flow(bin_data, simulate_gap=True)
|
|
# 所有片最终都被发送过 (含补发)
|
|
self.assertIn(256, sent)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|