"""线架注册模型（地址段扫描 / 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'])