✨ feat(iot): 添加 IotDeviceMessageServiceImpl 的单元测试，增强消息处理的可靠性

YunaiV
1 parent d668af19
Showing 5 changed files with 216 additions and 4 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImplTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/property/IotDevicePropertyServiceImplTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleActionTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotMqttDataRuleActionTest.java
@@ -97,11 +97,11 @@ public class IotWebSocketDataRuleAction extends
         }
     }
-    // 加锁原因：多线程并发发送消息时，多个线程可能同时检测到连接断开并尝试重连，
-    // 使用锁 + 双重检查保证同一服务器地址只有一个线程执行重连操作
     /**
      * 使用锁进行重连，保证同一服务器地址的重连操作线程安全
      *
+     * 加锁原因：多线程并发发送消息时，多个线程可能同时检测到连接断开并尝试重连，使用锁 + 双重检查保证同一服务器地址只有一个线程执行重连操作
+     *
      * @param webSocketClient WebSocket 客户端
      * @param config          配置信息
      */
+package cn.iocoder.yudao.module.iot.service.device.message;
+
+import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
+import cn.iocoder.yudao.module.iot.dal.tdengine.IotDeviceMessageMapper;
+import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
+import cn.iocoder.yudao.module.iot.service.ota.IotOtaTaskRecordService;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+
+import java.util.HashMap;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+/**
+ * {@link IotDeviceMessageServiceImpl} 的单元测试
+ *
+ * @author 芋道源码
+ */
+public class IotDeviceMessageServiceImplTest extends BaseMockitoUnitTest {
+
+    @InjectMocks
+    private IotDeviceMessageServiceImpl service;
+
+    @Mock
+    private IotDeviceService deviceService;
+    @Mock
+    private IotDevicePropertyService devicePropertyService;
+    @Mock
+    private IotOtaTaskRecordService otaTaskRecordService;
+    @Mock
+    private IotDeviceMessageMapper deviceMessageMapper;
+    @Mock
+    private IotDeviceMessageProducer deviceMessageProducer;
+
+    @Test
+    public void testCreateDeviceLogAsync_tsFallback_whenNull() {
+        // 准备：构造一条 ts 为 null 的消息
+        IotDeviceMessage message = buildMessage();
+        long before = System.currentTimeMillis();
+
+        // 调用
+        service.createDeviceLogAsync(message);
+        long after = System.currentTimeMillis();
+
+        // 断言：mapper.insert 接收到的 messageDO 已经被填上 ts，值在 [before, after] 区间
+        ArgumentCaptor<IotDeviceMessageDO> captor = ArgumentCaptor.forClass(IotDeviceMessageDO.class);
+        verify(deviceMessageMapper).insert(captor.capture());
+        Long actualTs = captor.getValue().getTs();
+        assertNotNull(actualTs, "ts 不应为空");
+        assertTrue(actualTs >= before && actualTs <= after,
+                "ts 应在调用前后区间内； 实际 = " + actualTs);
+    }
+
+    @Test
+    public void testCreateDeviceLogAsync_swallowMapperException() {
+        // 准备：mapper.insert 抛异常，验证 @Async 方法内部 try/catch 兜底，不向上抛
+        IotDeviceMessage message = buildMessage();
+        doThrow(new RuntimeException("DB unavailable")).when(deviceMessageMapper).insert(any());
+
+        // 调用 & 断言
+        assertDoesNotThrow(() -> service.createDeviceLogAsync(message));
+        verify(deviceMessageMapper).insert(any(IotDeviceMessageDO.class));
+    }
+
+    /** 构造一条最简属性上报消息 */
+    private IotDeviceMessage buildMessage() {
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setId("msg-1");
+        message.setDeviceId(2L);
+        message.setMethod(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod());
+        message.setParams(new HashMap<>());
+        return message;
+    }
+
+}
@@ -86,7 +86,7 @@ public class IotDevicePropertyServiceImplTest extends BaseMockitoUnitTest {
         service.saveDeviceProperty(device, message);
         // 断言：没有合法属性，不会写入 TDengine 与 Redis
-        verify(devicePropertyMapper, never()).insert(any(), any(), anyLong());
+        verify(devicePropertyMapper, never()).insert(any(), any(), anyLong(), anyLong());
         verify(deviceDataRedisDAO, never()).putAll(anyLong(), any());
     }
@@ -187,7 +187,7 @@ public class IotDevicePropertyServiceImplTest extends BaseMockitoUnitTest {
     @SuppressWarnings("unchecked")
     private Map<String, Object> captureMapperInsertProperties() {
         ArgumentCaptor<Map<String, Object>> captor = ArgumentCaptor.forClass(Map.class);
-        verify(devicePropertyMapper).insert(any(IotDeviceDO.class), captor.capture(), anyLong());
+        verify(devicePropertyMapper).insert(any(IotDeviceDO.class), captor.capture(), anyLong(), anyLong());
         return captor.getValue();
     }
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
+
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
+import com.zaxxer.hikari.HikariDataSource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import javax.sql.DataSource;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.*;
+
+/**
+ * {@link IotDatabaseDataRuleAction} 的单元测试
+ *
+ * @author HUIHUI
+ */
+class IotDatabaseDataRuleActionTest {
+
+    private IotDatabaseDataRuleAction databaseDataRuleAction;
+
+    @Mock
+    private JdbcTemplate jdbcTemplate;
+
+    @BeforeEach
+    public void setUp() {
+        MockitoAnnotations.openMocks(this);
+        databaseDataRuleAction = new IotDatabaseDataRuleAction();
+    }
+
+    @Test
+    public void testGetType() {
+        // 调用 & 断言：返回 Database 类型枚举值
+        assertEquals(IotDataSinkTypeEnum.DATABASE.getType(), databaseDataRuleAction.getType());
+    }
+
+    @Test
+    public void testCloseProducer_whenHikari() throws Exception {
+        // 准备：底层是 HikariDataSource
+        HikariDataSource hikari = mock(HikariDataSource.class);
+        when(jdbcTemplate.getDataSource()).thenReturn(hikari);
+
+        // 调用
+        databaseDataRuleAction.closeProducer(jdbcTemplate);
+
+        // 断言：HikariDataSource 被关闭
+        verify(hikari, times(1)).close();
+    }
+
+    @Test
+    public void testCloseProducer_whenNotHikari() throws Exception {
+        // 准备：底层不是 HikariDataSource，避免误调 close
+        DataSource other = mock(DataSource.class);
+        when(jdbcTemplate.getDataSource()).thenReturn(other);
+
+        // 调用 & 断言：不抛异常，且不会尝试关闭非 Hikari 数据源
+        assertDoesNotThrow(() -> databaseDataRuleAction.closeProducer(jdbcTemplate));
+        verifyNoInteractions(other);
+    }
+
+}
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
+
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.*;
+
+/**
+ * {@link IotMqttDataRuleAction} 的单元测试
+ *
+ * @author HUIHUI
+ */
+class IotMqttDataRuleActionTest {
+
+    private IotMqttDataRuleAction mqttDataRuleAction;
+
+    @Mock
+    private MqttClient mqttClient;
+
+    @BeforeEach
+    public void setUp() {
+        MockitoAnnotations.openMocks(this);
+        mqttDataRuleAction = new IotMqttDataRuleAction();
+    }
+
+    @Test
+    public void testGetType() {
+        // 调用 & 断言：返回 MQTT 类型枚举值
+        assertEquals(IotDataSinkTypeEnum.MQTT.getType(), mqttDataRuleAction.getType());
+    }
+
+    @Test
+    public void testCloseProducer_whenConnected() throws Exception {
+        // 准备：连接中状态
+        when(mqttClient.isConnected()).thenReturn(true);
+
+        // 调用
+        mqttDataRuleAction.closeProducer(mqttClient);
+
+        // 断言：先 disconnect 再 close
+        verify(mqttClient, times(1)).disconnect();
+        verify(mqttClient, times(1)).close();
+    }
+
+    @Test
+    public void testCloseProducer_whenAlreadyDisconnected() throws Exception {
+        // 准备：已断开状态
+        when(mqttClient.isConnected()).thenReturn(false);
+
+        // 调用
+        mqttDataRuleAction.closeProducer(mqttClient);
+
+        // 断言：跳过 disconnect，仅 close
+        verify(mqttClient, never()).disconnect();
+        verify(mqttClient, times(1)).close();
+    }
+
+}