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