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()
        
        # 模拟初始状态
        self.mock_serial.get_status.return_value = False
        self.mock_mqtt.get_status.return_value = False
    
    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', 9600)
    
    def test_serial_disconnect_api(self):
        """测试串口断开连接API"""
        # 发送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"""
        # 模拟串口已连接
        self.mock_serial.get_status.return_value = True
        
        # 发送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请求
        response = self.client.post('/api/serial/send',
                                  data=json.dumps({
                                      'data': '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连接成功"
        
        # 验证调用
        self.mock_mqtt.connect.assert_called_once_with(mqtt_config)
    
    def test_mqtt_disconnect_api(self):
        """测试MQTT断开连接API"""
        # 发送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'] == "MQTT已断开"
        
        # 验证调用
        self.mock_mqtt.disconnect.assert_called_once()
    
    def test_mqtt_status_api(self):
        """测试获取MQTT状态API"""
        # 模拟MQTT已连接
        self.mock_mqtt.get_status.return_value = True
        
        # 发送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请求
        response = self.client.post('/api/mqtt/subscribe',
                                  data=json.dumps({
                                      'topic': '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')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert isinstance(data, list)
    
    def test_get_mqtt_data_api(self):
        """测试获取MQTT数据API"""
        # 发送GET请求
        response = self.client.get('/api/data/mqtt')
        
        # 验证响应
        assert response.status_code == 200
        data = json.loads(response.data)
        assert isinstance(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')


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