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') if __name__ == "__main__": pytest.main(["-v", __file__])