dzxj_dtu/backend/tests/test_app.py

import pytest
import sys
import os
from unittest.mock import MagicMock, patch
from flask import json
from flask_socketio import SocketIOTestClient

# 添加项目根目录到Python路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app import app, socketio, serial_client, mqtt_client


class TestApp:
    
    def setup_method(self):
        """每个测试方法执行前的设置"""
        # 创建Flask测试客户端
        self.client = app.test_client()
        app.config['TESTING'] = True
        
        # 创建SocketIO测试客户端
        self.socketio_client = SocketIOTestClient(app, socketio)
        
        # 模拟串口客户端和MQTT客户端
        self.mock_serial = patch('app.serial_client').start()
        self.mock_mqtt = patch('app.mqtt_client').start()
        
        # 模拟初始状态 (get_status 返回 dict)
        self.mock_serial.get_status.return_value = {'connected': False, 'config': None, 'has_error': False}
        self.mock_serial.current_config = None
        self.mock_mqtt.get_status.return_value = {'connected': False, 'broker': None}
        self.mock_mqtt.config = None
    
    def teardown_method(self):
        """每个测试方法执行后的清理"""
        # 停止所有模拟
        patch.stopall()
        # 断开SocketIO连接
        self.socketio_client.disconnect()
    
    def test_serial_connect_api(self):
        """测试串口连接API"""
        # 模拟连接成功
        self.mock_serial.connect.return_value = (True, "串口连接成功")
        
        # 发送POST请求
        response = self.client.post('/api/serial/connect', 
                                   data=json.dumps({
                                       'port': 'COM1',
                                       'baudrate': 9600
                                   }),
                                   content_type='application/json')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['success'] is True
        assert data['message'] == "串口连接成功"
        
        # 验证调用 (使用关键字参数)
        self.mock_serial.connect.assert_called_once_with(
            'COM1', baudrate=9600, timeout=0.1, bytesize=8, parity='N', stopbits=1
        )
    
    def test_serial_disconnect_api(self):
        """测试串口断开连接API"""
        # 模拟断开返回 (success, message)
        self.mock_serial.disconnect.return_value = (True, "已断开连接")
        
        # 发送POST请求
        response = self.client.post('/api/serial/disconnect')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['success'] is True
        assert data['message'] == "已断开连接"
        
        # 验证调用
        self.mock_serial.disconnect.assert_called_once()
    
    def test_serial_status_api(self):
        """测试获取串口状态API"""
        # 模拟串口已连接 (get_status 返回 dict)
        self.mock_serial.get_status.return_value = {'connected': True, 'config': None, 'has_error': False}
        
        # 发送GET请求
        response = self.client.get('/api/serial/status')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['connected'] is True
    
    def test_serial_send_api(self):
        """测试串口发送数据API"""
        # 模拟发送成功
        self.mock_serial.send_data.return_value = (True, "数据发送成功")
        
        # 发送POST请求 (endpoint 使用 'message' 字段)
        response = self.client.post('/api/serial/send',
                                  data=json.dumps({
                                      'message': 'test message'
                                  }),
                                  content_type='application/json')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['success'] is True
        
        # 验证调用
        self.mock_serial.send_data.assert_called_once_with('test message')
    
    def test_mqtt_connect_api(self):
        """测试MQTT连接API"""
        # 模拟连接成功
        self.mock_mqtt.connect.return_value = (True, "MQTT连接成功")
        
        # 发送POST请求
        mqtt_config = {
            'broker': 'localhost',
            'port': 1883,
            'username': '',
            'password': '',
            'client_id': 'test_client',
            'keepalive': 60,
            'use_tls': False
        }
        response = self.client.post('/api/mqtt/connect',
                                  data=json.dumps(mqtt_config),
                                  content_type='application/json')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['success'] is True
        assert data['message'] == "MQTT连接成功"
        
        # 验证调用 (endpoint 使用 broker= 关键字参数)
        self.mock_mqtt.connect.assert_called_once()
        _, kwargs = self.mock_mqtt.connect.call_args
        assert kwargs['broker'] == 'localhost'
        assert kwargs['port'] == 1883
        assert kwargs['client_id'] == 'test_client'
    
    def test_mqtt_disconnect_api(self):
        """测试MQTT断开连接API"""
        # 模拟断开返回 (success, message)
        self.mock_mqtt.disconnect.return_value = (True, "已断开连接")
        
        # 发送POST请求
        response = self.client.post('/api/mqtt/disconnect')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['success'] is True
        assert data['message'] == "已断开连接"
        
        # 验证调用
        self.mock_mqtt.disconnect.assert_called_once()
    
    def test_mqtt_status_api(self):
        """测试获取MQTT状态API"""
        # 模拟MQTT已连接 (get_status 返回 dict)
        self.mock_mqtt.get_status.return_value = {'connected': True, 'broker': 'localhost'}
        
        # 发送GET请求
        response = self.client.get('/api/mqtt/status')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['connected'] is True
    
    def test_mqtt_publish_api(self):
        """测试MQTT发布消息API"""
        # 模拟发布成功
        self.mock_mqtt.publish.return_value = (True, "消息发布成功")
        
        # 发送POST请求
        response = self.client.post('/api/mqtt/publish',
                                  data=json.dumps({
                                      'topic': 'test/topic',
                                      'message': 'test message'
                                  }),
                                  content_type='application/json')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['success'] is True
        
        # 验证调用
        self.mock_mqtt.publish.assert_called_once_with('test/topic', 'test message')
    
    def test_mqtt_subscribe_api(self):
        """测试MQTT订阅主题API"""
        # 模拟订阅成功
        self.mock_mqtt.subscribe.return_value = (True, "主题订阅成功")
        
        # 发送POST请求 (endpoint 使用 'topics' 字段)
        response = self.client.post('/api/mqtt/subscribe',
                                  data=json.dumps({
                                      'topics': ['test/topic']
                                  }),
                                  content_type='application/json')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['success'] is True
        
        # 验证调用
        self.mock_mqtt.subscribe.assert_called_once_with(['test/topic'])
    
    def test_get_serial_data_api(self):
        """测试获取串口数据API"""
        # 发送GET请求
        response = self.client.get('/api/data/serial')
        
        # 验证响应 (endpoint 返回 {'data': [...]})
        assert response.status_code == 200
        data = json.loads(response.data)
        assert 'data' in data
        assert isinstance(data['data'], list)
    
    def test_get_mqtt_data_api(self):
        """测试获取MQTT数据API"""
        # 发送GET请求
        response = self.client.get('/api/data/mqtt')
        
        # 验证响应 (endpoint 返回 {'data': [...]})
        assert response.status_code == 200
        data = json.loads(response.data)
        assert 'data' in data
        assert isinstance(data['data'], list)
    
    def test_forward_config_api(self):
        """测试设置转发配置API"""
        # 发送POST请求
        response = self.client.post('/api/forward/config',
                                  data=json.dumps({
                                      'serial_to_mqtt': True,
                                      'mqtt_to_serial': False,
                                      'publish_topic': 'serial/data'
                                  }),
                                  content_type='application/json')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['success'] is True
        assert data['message'] == "转发配置已更新"
    
    def test_get_forward_status_api(self):
        """测试获取转发状态API"""
        # 发送GET请求
        response = self.client.get('/api/forward/status')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert 'serial_to_mqtt' in data
        assert 'mqtt_to_serial' in data
        assert 'publish_topic' in data
    
    def test_health_check_api(self):
        """测试健康检查API"""
        # 发送GET请求
        response = self.client.get('/api/health')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data['status'] == 'healthy'
        assert 'timestamp' in data
        assert 'services' in data
    
    def test_socketio_data_namespace(self):
        """测试数据命名空间的WebSocket连接"""
        # 连接到数据命名空间
        self.socketio_client.connect(namespace='/data')
        
        # 验证连接成功
        received = self.socketio_client.get_received('/data')
        # 应该收到历史数据事件
        events = [event['name'] for event in received]
        assert 'serial_data_history' in events or 'mqtt_data_history' in events
        
        # 断开连接
        self.socketio_client.disconnect(namespace='/data')
    
    def test_socketio_status_namespace(self):
        """测试状态命名空间的WebSocket连接"""
        # 连接到状态命名空间
        self.socketio_client.connect(namespace='/status')
        
        # 验证连接成功
        received = self.socketio_client.get_received('/status')
        # 应该收到当前状态事件
        events = [event['name'] for event in received]
        assert 'serial_status' in events
        assert 'mqtt_status' in events
        assert 'forward_status' in events
        
        # 断开连接
        self.socketio_client.disconnect(namespace='/status')


    def test_switch_screen_skips_offline_panels(self):
        """测试屏幕切换跳过离线终端：只落到在线面板"""
        import app as app_module
        from app import switch_screen_panel, panel_config, screen_lock

        orig_config = dict(panel_config)
        orig_index = app_module.screen_current_panel_index

        def _current_addr():
            with screen_lock:
                panels = app_module.get_sorted_panels()
                idx = app_module.screen_current_panel_index
                return panels[idx][1]['address'] if panels else None

        try:
            panel_config.clear()
            panel_config.update({
                'PANEL_1': {'address': 1, 'position': 1, 'status': 'offline'},
                'PANEL_2': {'address': 2, 'position': 2, 'status': 'online'},
                'PANEL_3': {'address': 3, 'position': 3, 'status': 'offline'},
                'PANEL_4': {'address': 4, 'position': 4, 'status': 'online'},
            })

            with patch.object(app_module, 'refresh_screen_panel', return_value=(True, 'ok')):
                # 当前显示 PANEL_1(离线)，下一终端 -> 跳过 -> PANEL_2(在线)
                with screen_lock:
                    app_module.screen_current_panel_index = 0
                assert switch_screen_panel(1)[0]
                assert _current_addr() == 2

                # PANEL_2(在线) -> 下一终端 -> 跳过 PANEL_3(离线) -> PANEL_4(在线)
                assert switch_screen_panel(1)[0]
                assert _current_addr() == 4

                # PANEL_4(在线) -> 下一终端 -> 环绕跳过 PANEL_1(离线) -> PANEL_2(在线)
                assert switch_screen_panel(1)[0]
                assert _current_addr() == 2

                # PANEL_2(在线) -> 上一终端 -> 环绕跳过 PANEL_1(离线) -> PANEL_4(在线)
                assert switch_screen_panel(-1)[0]
                assert _current_addr() == 4
        finally:
            panel_config.clear()
            panel_config.update(orig_config)
            with screen_lock:
                app_module.screen_current_panel_index = orig_index

    def test_switch_screen_all_offline_falls_back(self):
        """测试全部终端离线时切换退化为相邻切换，不卡死"""
        import app as app_module
        from app import switch_screen_panel, panel_config, screen_lock

        orig_config = dict(panel_config)
        orig_index = app_module.screen_current_panel_index

        def _current_panel():
            with screen_lock:
                panels = app_module.get_sorted_panels()
                idx = app_module.screen_current_panel_index
                return panels[idx][0] if panels else None

        try:
            panel_config.clear()
            panel_config.update({
                'PANEL_1': {'address': 1, 'position': 1, 'status': 'offline'},
                'PANEL_2': {'address': 2, 'position': 2, 'status': 'offline'},
                'PANEL_3': {'address': 3, 'position': 3, 'status': 'offline'},
            })
            with patch.object(app_module, 'refresh_screen_panel', return_value=(True, 'ok')):
                with screen_lock:
                    app_module.screen_current_panel_index = 0
                assert switch_screen_panel(1)[0]
                assert _current_panel() == 'PANEL_2'
        finally:
            panel_config.clear()
            panel_config.update(orig_config)
            with screen_lock:
                app_module.screen_current_panel_index = orig_index


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