dzxj_dtu/backend/tests/test_panel_registration.py

"""线架注册模型（地址段扫描 / UID 绑定 / 换架自动接管）单元测试。"""
import sys
import os
from unittest.mock import patch

import pytest

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import app as app_module  # noqa: E402
from app import app  # noqa: E402


def make_cards(port_map):
    """构造 24 路读卡结果；port_map: {port_id: 'hex12'}"""
    cards = []
    for ant in range(1, 25):
        hex_str = (port_map.get(ant) or '').lower()
        cards.append({
            'antenna': ant,
            'card_str': hex_str,
            'present': bool(hex_str),
            'uid': '' if not hex_str else ':'.join(hex_str[i:i + 2] for i in range(0, 12, 2)),
        })
    return cards


def read_result(cards, panel_uid=None):
    return {
        'success': True, 'device_address': 1, 'function_code': 3,
        'cards': cards, 'panel_uid': panel_uid, 'raw_data': 'TEST',
    }


class TestPanelRegistration:
    def setup_method(self):
        self._saved = {
            'stored': dict(app_module.address_config.stored_devices),
            'panel_config': dict(app_module.panel_config),
            'port_state': dict(app_module.port_state),
            'fail': dict(app_module.panel_fail_count),
            'address_range': app_module.dtu_config.get('address_range'),
            'port_mappings': app_module.dtu_config.get('port_mappings'),
            'last_seen': dict(app_module.device_last_seen),
        }
        self.mock_modbus = patch.object(app_module, 'modbus_client').start()
        self.mock_save_dev = patch.object(app_module, 'save_device_config').start()
        self.mock_save_dtu = patch.object(app_module, 'save_dtu_config').start()
        self.mock_register = patch.object(app_module, 'dtu_register').start()
        self.mock_panel_status = patch.object(app_module, 'dtu_publish_panel_status').start()

        app_module.address_config.stored_devices.clear()
        app_module.panel_config.clear()
        app_module.port_state.clear()
        app_module.panel_fail_count.clear()
        app_module.device_last_seen.clear()

    def teardown_method(self):
        patch.stopall()
        app_module.address_config.stored_devices.clear()
        app_module.address_config.stored_devices.update(self._saved['stored'])
        app_module.panel_config.clear()
        app_module.panel_config.update(self._saved['panel_config'])
        app_module.port_state.clear()
        app_module.port_state.update(self._saved['port_state'])
        app_module.panel_fail_count.clear()
        app_module.panel_fail_count.update(self._saved['fail'])
        app_module.device_last_seen.clear()
        app_module.device_last_seen.update(self._saved['last_seen'])
        if self._saved['address_range'] is None:
            app_module.dtu_config.pop('address_range', None)
        else:
            app_module.dtu_config['address_range'] = self._saved['address_range']
        if self._saved['port_mappings'] is None:
            app_module.dtu_config.pop('port_mappings', None)
        else:
            app_module.dtu_config['port_mappings'] = self._saved['port_mappings']

    def test_takeover_rebinds_uid_and_inherits_mappings(self):
        """换架: 原地址探测到新 UID -> 重绑地址、保留 panel_id/映射、自动下发卡号。"""
        old_uid = 'a1' * 12
        new_uid = 'b2' * 12
        pid = 'PANEL_dtu_001_5'
        app_module.dtu_config['address_range'] = {'start': 5, 'end': 5}
        app_module.dtu_config['port_mappings'] = {pid: {'3': 'AABBCCDDEEFF'}}
        app_module.address_config.stored_devices[old_uid] = 5
        app_module.panel_config[pid] = {'address': 5, 'position': 5, 'panel_uid': old_uid, 'status': 'offline'}
        app_module.panel_fail_count[pid] = 3

        self.mock_modbus.read_all_antenna_cards.return_value = read_result(
            make_cards({3: 'aabbccddeeff'}), panel_uid=new_uid)
        self.mock_modbus.set_card_mapping.return_value = {'success': True}

        app_module._sweep_panel_addresses()

        devices = app_module.address_config.stored_devices
        assert devices.get(new_uid) == 5
        assert old_uid not in devices
        assert app_module.panel_config[pid]['panel_uid'] == new_uid
        assert app_module.panel_config[pid]['status'] == 'online'
        assert pid not in app_module.panel_fail_count
        # 期望卡号已下发到硬件（继承原面板数据）
        args, _ = self.mock_modbus.set_card_mapping.call_args
        assert args[0] == 5
        assert args[1].get(3) == 'AABBCCDDEEFF'
        assert self.mock_save_dev.called
        assert self.mock_register.called

    def test_probe_same_uid_marks_back_online(self):
        """缺失探测回应原 UID -> 恢复在线，不改变绑定。"""
        uid = 'c3' * 12
        pid = 'PANEL_dtu_001_2'
        app_module.dtu_config['address_range'] = {'start': 2, 'end': 2}
        app_module.address_config.stored_devices[uid] = 2
        app_module.panel_config[pid] = {'address': 2, 'position': 2, 'panel_uid': uid, 'status': 'offline'}
        app_module.panel_fail_count[pid] = 3

        self.mock_modbus.read_all_antenna_cards.return_value = read_result(
            make_cards({}), panel_uid=uid)

        app_module._sweep_panel_addresses()

        assert app_module.address_config.stored_devices == {uid: 2}
        assert app_module.panel_config[pid]['status'] == 'online'
        assert pid not in app_module.panel_fail_count

    def test_unbound_probe_registers_new_panel(self):
        """未绑定地址探测到新线架 -> 绑定地址↔UID。"""
        new_uid = 'd4' * 12
        app_module.dtu_config['address_range'] = {'start': 4, 'end': 4}
        self.mock_modbus.read_all_antenna_cards.return_value = read_result(
            make_cards({}), panel_uid=new_uid)

        app_module._sweep_panel_addresses()

        assert app_module.address_config.stored_devices.get(new_uid) == 4
        assert 'PANEL_dtu_001_4' in app_module.panel_config
        assert self.mock_save_dev.called

    def test_missing_panel_suspends_unbound_probe(self):
        """有面板处于缺失/接管状态时，本轮不抽查未绑定地址。"""
        old_uid = 'e5' * 12
        pid = 'PANEL_dtu_001_2'
        app_module.dtu_config['address_range'] = {'start': 2, 'end': 3}
        app_module.address_config.stored_devices[old_uid] = 2
        app_module.panel_config[pid] = {'address': 2, 'position': 2, 'panel_uid': old_uid, 'status': 'offline'}
        app_module.panel_fail_count[pid] = 1
        self.mock_modbus.read_all_antenna_cards.return_value = {'error': 'timeout'}

        app_module._sweep_panel_addresses()

        probed = [c.args[0] for c in self.mock_modbus.read_all_antenna_cards.call_args_list]
        assert probed == [2]

    def test_read_failure_threshold_marks_offline(self):
        """连续失败达到阈值 -> 标记离线（并发布离线状态）。"""
        uid = 'f6' * 12
        pid = 'PANEL_dtu_001_3'
        app_module.dtu_config['address_range'] = {'start': 3, 'end': 3}
        app_module.address_config.stored_devices[uid] = 3
        app_module.panel_config[pid] = {'address': 3, 'position': 3, 'panel_uid': uid, 'status': 'online'}
        self.mock_modbus.read_all_antenna_cards.return_value = {'error': 'timeout'}

        for _ in range(app_module.PANEL_MISS_THRESHOLD):
            app_module._sweep_panel_addresses()

        assert app_module.panel_config[pid]['status'] == 'offline'
        assert app_module.panel_fail_count[pid] == app_module.PANEL_MISS_THRESHOLD
        self.mock_panel_status.assert_any_call(pid, 3, None, online=False)

    def test_panel_unbind_api(self):
        """解绑 API: 移除绑定与面板配置。"""
        client = app.test_client()
        uid = '99' * 12
        pid = 'PANEL_dtu_001_7'
        app_module.address_config.stored_devices[uid] = 7
        app_module.panel_config[pid] = {'address': 7, 'position': 7, 'panel_uid': uid, 'status': 'online'}

        resp = client.post('/api/panel/unbind', json={'address': 7})
        data = resp.get_json()

        assert resp.status_code == 200 and data['success']
        assert uid not in app_module.address_config.stored_devices
        assert pid not in app_module.panel_config
        assert self.mock_save_dev.called

    def test_update_dtu_config_address_range(self):
        """DTU 配置 API: 地址段写入并校验。"""
        client = app.test_client()
        resp = client.post('/api/dtu/config', json={'address_range': {'start': 5, 'end': 60}})
        assert resp.status_code == 200
        assert app_module.dtu_config['address_range'] == {'start': 5, 'end': 60}
        assert self.mock_save_dtu.called

        resp = client.post('/api/dtu/config', json={'address_range': {'start': 70, 'end': 10}})
        assert resp.status_code == 400
        assert app_module.dtu_config['address_range'] == {'start': 5, 'end': 60}

    def test_address_range_fallback(self):
        app_module.dtu_config['address_range'] = {'start': 60, 'end': 10}
        assert app_module._address_range() == (1, 50)
        app_module.dtu_config['address_range'] = {'start': 3, 'end': 9}
        assert app_module._address_range() == (3, 9)
        app_module.dtu_config['address_range'] = None
        assert app_module._address_range() == (1, 50)


if __name__ == '__main__':
    pytest.main([__file__, '-v'])