dzxj_dtu/backend/tests/test_modbus_rtu.py

"""V4+ 带 UID 批量读卡协议 (modbus_rtu) 单元测试。

协议 V4+ (地址段轮询/线架绑定):
  请求 20B: [addr][0x03][0x00][0x02][0x00][0x48][12 字节 UID][CRC_L][CRC_H]
    UID = 已绑定线架 UID；uid=None 时为 12 字节全 0（探测未绑定地址）
  响应 161B: [addr][0x03][0x90][144 数据字节][12 字节 UID][CRC_L][CRC_H]
    UID = 线架自身 UID（用于绑定）
  每张卡 6 字节（硬件卡号，12 位十六进制）; 无卡标记 = FF FF FF FF FF FF

过渡期兼容: 旧固件 149B 响应（无 UID 字段）可解析但 panel_uid=None。

注: 硬件通信固定 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,
    build_read_cards_request,
    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,
    READ_WITH_UID_REQUEST_LEN,
    READ_WITH_UID_RESPONSE_LEN,
    ZERO_UID,
)

# 协议文档示例 UID: 18 00 40 00 14 00 00 59 59 54 30 56
PANEL_UID = bytes([0x18, 0x00, 0x40, 0x00, 0x14, 0x00, 0x00, 0x59, 0x59, 0x54, 0x30, 0x56])


def build_bulk_response(addr: int, tag_data: bytes, uid: bytes = PANEL_UID) -> bytes:
    """构造合法的 161 字节 V4+ 批量读响应帧（带线架 UID）。"""
    assert len(tag_data) == CARD_DATA_BYTES, f"tag_data 必须 {CARD_DATA_BYTES} 字节"
    assert len(uid) == 12, "uid 必须 12 字节"
    frame = bytes([addr, 0x03, BULK_BYTE_COUNT]) + tag_data + uid
    return frame + calculate_crc16(frame)


def build_legacy_bulk_response(addr: int, tag_data: bytes) -> bytes:
    """构造旧固件 149 字节响应帧（无 UID 字段）。"""
    assert len(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_with_zero_uid(self):
        """uid=None（探测）: 20 字节请求，UID 段全 0。"""
        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 len(sent) == READ_WITH_UID_REQUEST_LEN
        assert sent[:6] == bytes([0x01, 0x03, 0x00, 0x02, 0x00, 0x48])
        assert sent[6:18] == ZERO_UID
        assert sent[18:] == calculate_crc16(sent[:18])

    def test_request_frame_with_bound_uid(self):
        """传入绑定 UID: 请求帧尾部携带该 UID。"""
        self._mock_response(build_bulk_response(0x02, NO_TAG_BYTES * ANTENNA_COUNT))
        self.client.read_all_antenna_cards(0x02, uid=PANEL_UID)

        sent = self.client.serial.send_and_wait.call_args[0][0]
        assert len(sent) == READ_WITH_UID_REQUEST_LEN
        assert sent[6:18] == PANEL_UID
        assert sent[18:] == calculate_crc16(sent[:18])

    def test_build_read_cards_request_helper(self):
        assert build_read_cards_request(0x01) == build_read_cards_request(0x01, ZERO_UID)
        with pytest.raises(ValueError):
            build_read_cards_request(0x01, b'\x00' * 11)

    def test_min_response_bytes_is_full_frame(self):
        """必须要求完整 161 字节，避免截断。"""
        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') == READ_WITH_UID_RESPONSE_LEN

    def test_panel_uid_extracted(self):
        """161B 响应中的线架自身 UID 必须被提取（用于绑定）。"""
        self._mock_response(build_bulk_response(0x01, NO_TAG_BYTES * ANTENNA_COUNT, uid=PANEL_UID))

        result = self.client.read_all_antenna_cards(0x01)

        assert 'error' not in result
        assert result['panel_uid'] == PANEL_UID.hex()

    def test_zero_uid_response_gives_none(self):
        """响应的 UID 段全 0(无效) -> panel_uid=None。"""
        self._mock_response(build_bulk_response(0x01, NO_TAG_BYTES * ANTENNA_COUNT, uid=ZERO_UID))

        result = self.client.read_all_antenna_cards(0x01)

        assert 'error' not in result
        assert result['panel_uid'] is None

    def test_legacy_149_response_parsed_without_uid(self):
        """过渡期: 旧固件 149B 响应可解析，但 panel_uid=None。"""
        self._mock_response(build_legacy_bulk_response(0x01, NO_TAG_BYTES * ANTENNA_COUNT))

        result = self.client.read_all_antenna_cards(0x01)

        assert 'error' not in result
        assert result['panel_uid'] is None
        assert len(result['cards']) == ANTENNA_COUNT

    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'])