"""V4 批量读卡协议 (modbus_rtu) 单元测试。 协议 V4: 一次性批量读 24 路卡号。 请求: [addr][0x03][0x00][0x02][0x00][0x48][CRC_L][CRC_H] -> 01 03 00 02 00 48 E4 3C 响应: [addr][0x03][0x90][144 数据字节][CRC_L][CRC_H] = 149 字节 每张卡 6 字节（硬件卡号，12 位十六进制）; 无卡标记 = FF FF FF FF FF FF 注: 硬件通信固定 6 字节卡号；对 MQTT 的 8 字节(16 位) jumper_uid 适配在 app.py 边界完成 (前 4 位 DTU 号 + 后 12 位卡号)，本测试只覆盖硬件通信层。 """ import sys import os from unittest.mock import MagicMock import pytest sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from modules.modbus_rtu import ( # noqa: E402 ModbusRTUClient, calculate_crc16, verify_crc16, CARD_REG_ADDR, CARD_REG_QUANTITY, CARD_DATA_BYTES, BYTES_PER_TAG, ANTENNA_COUNT, NO_TAG_BYTES, BULK_RESPONSE_LEN, BULK_BYTE_COUNT, ANTENNA_ADDRESSES, ) def build_bulk_response(addr: int, tag_data: bytes) -> bytes: """构造合法的 149 字节 V4 批量读响应帧。""" assert len(tag_data) == CARD_DATA_BYTES, f"tag_data 必须 {CARD_DATA_BYTES} 字节" frame = bytes([addr, 0x03, BULK_BYTE_COUNT]) + tag_data return frame + calculate_crc16(frame) def build_exception_response(addr: int, exception_code: int) -> bytes: """构造 Modbus 异常响应帧 (5 字节)。""" frame = bytes([addr, 0x83, exception_code]) return frame + calculate_crc16(frame) class TestReadAllAntennaCards: """V4 批量读解析。""" def setup_method(self): self.client = ModbusRTUClient(MagicMock()) def _mock_response(self, response_bytes: bytes): self.client.serial.ser = MagicMock() self.client.serial.send_and_wait.return_value = response_bytes def test_request_frame_matches_doc(self): """请求帧必须等于文档示例 01 03 00 02 00 48 E4 3C。""" self._mock_response(build_bulk_response(0x01, NO_TAG_BYTES * ANTENNA_COUNT)) self.client.read_all_antenna_cards(0x01) sent = self.client.serial.send_and_wait.call_args[0][0] assert sent == bytes([0x01, 0x03, 0x00, 0x02, 0x00, 0x48, 0xE4, 0x3C]) def test_min_response_bytes_is_full_frame(self): """必须要求完整 149 字节，避免截断。""" self._mock_response(build_bulk_response(0x01, NO_TAG_BYTES * ANTENNA_COUNT)) self.client.read_all_antenna_cards(0x01) kwargs = self.client.serial.send_and_wait.call_args.kwargs assert kwargs.get('min_response_bytes') == BULK_RESPONSE_LEN def test_no_tag_response(self): """24 路全无卡 (FF*144)。""" tag_data = NO_TAG_BYTES * ANTENNA_COUNT # 144 字节 0xFF self._mock_response(build_bulk_response(0x01, tag_data)) result = self.client.read_all_antenna_cards(0x01) assert 'error' not in result assert result['success'] is True assert result['device_address'] == 0x01 assert result['function_code'] == 0x03 assert len(result['cards']) == ANTENNA_COUNT for idx, card in enumerate(result['cards']): assert card['antenna'] == idx + 1 assert card['present'] is False assert card['card_str'] == '' assert card['uid'] == '' def test_tag_on_antenna_24(self): """24 号天线有卡 ED9A57F95001，其余无卡。""" tag = bytes([0xED, 0x9A, 0x57, 0xF9, 0x50, 0x01]) tag_data = NO_TAG_BYTES * 23 + tag # 138 + 6 = 144 self._mock_response(build_bulk_response(0x03, tag_data)) result = self.client.read_all_antenna_cards(0x03) assert 'error' not in result cards = result['cards'] assert len(cards) == 24 for i in range(23): assert cards[i]['present'] is False assert cards[23]['antenna'] == 24 assert cards[23]['present'] is True assert cards[23]['card_str'] == 'ed9a57f95001' assert cards[23]['uid'] == 'ed:9a:57:f9:50:01' def test_tag_on_antenna_1(self): """1 号天线有卡，其余无卡。""" tag = bytes([0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]) tag_data = tag + NO_TAG_BYTES * 23 self._mock_response(build_bulk_response(0x01, tag_data)) result = self.client.read_all_antenna_cards(0x01) assert result['cards'][0]['present'] is True assert result['cards'][0]['card_str'] == 'aabbccddeeff' assert result['cards'][0]['uid'] == 'aa:bb:cc:dd:ee:ff' for i in range(1, 24): assert result['cards'][i]['present'] is False def test_multiple_tags(self): """1/12/24 号天线同时有卡。""" tag1 = bytes([0x11, 0x22, 0x33, 0x44, 0x55, 0x66]) tag12 = bytes([0x77, 0x88, 0x99, 0xAA, 0xBB, 0xCC]) tag24 = bytes([0xDD, 0xEE, 0xFF, 0x00, 0x11, 0x22]) tag_data = bytearray(NO_TAG_BYTES * ANTENNA_COUNT) tag_data[0:6] = tag1 tag_data[11 * 6:12 * 6] = tag12 tag_data[23 * 6:24 * 6] = tag24 self._mock_response(build_bulk_response(0x01, bytes(tag_data))) result = self.client.read_all_antenna_cards(0x01) assert result['cards'][0]['card_str'] == '112233445566' assert result['cards'][11]['card_str'] == '778899aabbcc' assert result['cards'][23]['card_str'] == 'ddeeff001122' assert result['cards'][1]['present'] is False def test_crc_failure(self): """CRC 篡改 -> error。""" response = build_bulk_response(0x01, NO_TAG_BYTES * ANTENNA_COUNT) response = response[:-1] + bytes([response[-1] ^ 0xFF]) # 改 CRC 末字节 self._mock_response(response) result = self.client.read_all_antenna_cards(0x01) assert 'error' in result assert 'CRC' in result['error'] def test_short_frame(self): """截断响应 (<149) -> error。""" self._mock_response(bytes([0x01, 0x03, 0x90] + [0xFF] * 7)) result = self.client.read_all_antenna_cards(0x01) assert 'error' in result def test_exception_response(self): """Modbus 异常帧 (0x83) -> error 含 exception。""" self._mock_response(build_exception_response(0x01, 0x02)) result = self.client.read_all_antenna_cards(0x01) assert 'error' in result assert 'exception' in result['error'].lower() def test_unexpected_function_code(self): """非 0x03 响应 -> error。""" frame = bytes([0x01, 0x04, BULK_BYTE_COUNT]) + NO_TAG_BYTES * ANTENNA_COUNT self._mock_response(frame + calculate_crc16(frame)) result = self.client.read_all_antenna_cards(0x01) assert 'error' in result def test_serial_not_connected(self): """串口未连接 -> error。""" self.client.serial.ser = None result = self.client.read_all_antenna_cards(0x01) assert 'error' in result def test_empty_response(self): """空响应 -> error。""" self._mock_response(b'') result = self.client.read_all_antenna_cards(0x01) assert 'error' in result class TestReadAntennaCardSingle: """单卡提取 (向后兼容包装)。""" def setup_method(self): self.client = ModbusRTUClient(MagicMock()) self.client.serial.ser = MagicMock() def test_extract_antenna_1_with_tag(self): tag = bytes([0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x01]) tag_data = tag + NO_TAG_BYTES * 23 self.client.serial.send_and_wait.return_value = build_bulk_response(0x01, tag_data) result = self.client.read_antenna_card(0x01, 1) assert 'error' not in result assert result['antenna'] == 1 assert result['card_number_str'] == 'deadbeef0001' assert result['uid'] == 'de:ad:be:ef:00:01' assert result['card_number'] == int.from_bytes(tag, 'big') assert result['card_number_hex'] == f'0x{int.from_bytes(tag, "big"):012x}' def test_extract_antenna_no_tag(self): """无卡时 card_number_str 为空串（V2 是 16 个 0）。""" self.client.serial.send_and_wait.return_value = build_bulk_response( 0x01, NO_TAG_BYTES * 24) result = self.client.read_antenna_card(0x01, 5) assert 'error' not in result assert result['antenna'] == 5 assert result['card_number_str'] == '' assert result['uid'] == '' def test_extract_antenna_24_with_tag(self): tag = bytes([0xED, 0x9A, 0x57, 0xF9, 0x50, 0x01]) tag_data = NO_TAG_BYTES * 23 + tag self.client.serial.send_and_wait.return_value = build_bulk_response(0x03, tag_data) result = self.client.read_antenna_card(0x03, 24) assert 'error' not in result assert result['antenna'] == 24 assert result['card_number_str'] == 'ed9a57f95001' def test_invalid_antenna_number(self): result = self.client.read_antenna_card(0x01, 25) assert 'error' in result result = self.client.read_antenna_card(0x01, 0) assert 'error' in result def test_bulk_error_propagates(self): """批量读失败时，单卡读透传 error。""" self.client.serial.send_and_wait.return_value = b'' result = self.client.read_antenna_card(0x01, 1) assert 'error' in result def test_does_not_send_v2_frame(self): """单卡读不得发 V2 逐天线帧 (0x0002+(n-1)*4)。""" self.client.serial.send_and_wait.return_value = build_bulk_response( 0x01, NO_TAG_BYTES * 24) self.client.read_antenna_card(0x01, 10) sent = self.client.serial.send_and_wait.call_args[0][0] # 必须是 V4 批量帧 01 03 00 02 00 48 ... assert sent[1] == 0x03 assert sent[2:4] == bytes([0x00, 0x02]) assert sent[4:6] == bytes([0x00, 0x48]) # V2 第 10 天线会是 0x0026，不得出现 assert sent[2:4] != bytes([0x00, 0x26]) class TestAntennaAddressesMapping: """V4 天线地址映射。""" def test_all_antennas_share_card_register(self): """V4 下所有天线共享 0x0002。""" for ant in range(1, 25): assert ANTENNA_ADDRESSES[ant] == CARD_REG_ADDR def test_no_v2_per_antenna_addresses(self): """不得残留 V2 的逐天线地址 (0x0006/0x000A/...)。""" for ant in range(1, 25): assert ANTENNA_ADDRESSES[ant] != 0x0002 + (ant - 1) * 4 or ant == 1 class TestCrcHelpers: """CRC 辅助函数 (回归)。""" def test_request_crc_matches_doc(self): """文档示例请求帧 (01 03 00 02 00 48) CRC = E4 3C。""" req = bytes([0x01, 0x03, 0x00, 0x02, 0x00, 0x48]) assert calculate_crc16(req) == bytes([0xE4, 0x3C]) def test_no_tag_crc_is_not_doc_typo(self): """无标签响应 CRC 实为 8F 44，文档的 C9 A2 是笔误。""" frame = bytes([0x01, 0x03, 0x90]) + b'\xff' * 144 assert calculate_crc16(frame) == bytes([0x8F, 0x44]) def test_verify_crc16_valid_frame(self): frame = bytes([0x01, 0x03, 0x90]) + b'\xff' * 144 assert verify_crc16(frame + calculate_crc16(frame)) is True def test_verify_crc16_corrupted(self): frame = bytes([0x01, 0x03, 0x90]) + b'\xff' * 144 crc = bytearray(calculate_crc16(frame)) crc[0] ^= 0xFF assert verify_crc16(frame + bytes(crc)) is False if __name__ == '__main__': pytest.main([__file__, '-v'])