""" 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)