|
|
@@ -0,0 +1,317 @@
|
|
|
+package com.zksy.web.websocket;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
+import com.zksy.base.domain.EquipmentStatus;
|
|
|
+import com.zksy.base.service.EquipmentStatusService;
|
|
|
+import com.zksy.base.service.BusinessWorkOrderService;
|
|
|
+import com.zksy.framework.web.service.TokenService;
|
|
|
+import org.springframework.data.redis.connection.Message;
|
|
|
+import org.junit.jupiter.api.BeforeEach;
|
|
|
+import org.junit.jupiter.api.Test;
|
|
|
+import org.mockito.ArgumentCaptor;
|
|
|
+import org.springframework.test.util.ReflectionTestUtils;
|
|
|
+import org.springframework.web.socket.CloseStatus;
|
|
|
+import org.springframework.web.socket.TextMessage;
|
|
|
+import org.springframework.web.socket.WebSocketSession;
|
|
|
+
|
|
|
+import java.util.HashMap;
|
|
|
+import java.nio.charset.StandardCharsets;
|
|
|
+import java.time.LocalDateTime;
|
|
|
+
|
|
|
+import static org.junit.jupiter.api.Assertions.assertEquals;
|
|
|
+import static org.junit.jupiter.api.Assertions.assertNotNull;
|
|
|
+import static org.junit.jupiter.api.Assertions.assertTrue;
|
|
|
+import static org.mockito.Mockito.mock;
|
|
|
+import static org.mockito.Mockito.verify;
|
|
|
+import static org.mockito.Mockito.when;
|
|
|
+
|
|
|
+class DeviceStatusCockpitWebSocketHandlerTest {
|
|
|
+ private static final String DEVICE_CODE = "TC-PS-WS-202609020012";
|
|
|
+
|
|
|
+ private DeviceStatusWebSocketHandler handler;
|
|
|
+ private WebSocketSession session;
|
|
|
+ private EquipmentStatusService equipmentStatusService;
|
|
|
+ private SubscriptionManager subscriptionManager;
|
|
|
+
|
|
|
+ @BeforeEach
|
|
|
+ void setUp() {
|
|
|
+ handler = new DeviceStatusWebSocketHandler();
|
|
|
+ session = mock(WebSocketSession.class);
|
|
|
+ HashMap<String, Object> attributes = new HashMap<>();
|
|
|
+ org.mockito.Mockito.doReturn("ws-session-1").when(session).getId();
|
|
|
+ org.mockito.Mockito.doReturn(true).when(session).isOpen();
|
|
|
+ org.mockito.Mockito.doReturn(attributes).when(session).getAttributes();
|
|
|
+
|
|
|
+ equipmentStatusService = mock(EquipmentStatusService.class);
|
|
|
+ subscriptionManager = new SubscriptionManager();
|
|
|
+ ReflectionTestUtils.setField(handler, "equipmentStatusService", equipmentStatusService);
|
|
|
+ ReflectionTestUtils.setField(handler, "subscriptionManager", subscriptionManager);
|
|
|
+ ReflectionTestUtils.setField(handler, "tokenService", mock(TokenService.class));
|
|
|
+ ReflectionTestUtils.setField(handler, "businessWorkOrderService", mock(BusinessWorkOrderService.class));
|
|
|
+ ReflectionTestUtils.setField(handler, "workOrderMaxItems", 100);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void cockpitSubscribeAllReturnsExplicitScopeSubscription() throws Exception {
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+
|
|
|
+ ArgumentCaptor<TextMessage> response = ArgumentCaptor.forClass(TextMessage.class);
|
|
|
+ verify(session).sendMessage(response.capture());
|
|
|
+ JSONObject payload = JSONObject.parseObject(response.getValue().getPayload());
|
|
|
+ assertEquals("cockpitSubscribed", payload.getString("type"));
|
|
|
+ assertEquals("all", payload.getString("scope"));
|
|
|
+ assertTrue(payload.containsKey("serverUpdatedAt"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void cockpitAllScopeReceivesChangedDeviceDelta() throws Exception {
|
|
|
+ handler.afterConnectionEstablished(session);
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+
|
|
|
+ EquipmentStatus status = new EquipmentStatus();
|
|
|
+ status.setCurrentStatus(1);
|
|
|
+ status.setAlarmStatus(1);
|
|
|
+ status.setOnlineStatus(1);
|
|
|
+ status.setStatusUpdateTime(LocalDateTime.parse("2026-09-02T09:59:50"));
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(status);
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(DEVICE_CODE)).thenReturn(true);
|
|
|
+ DeviceStatusRedisListener listener = new DeviceStatusRedisListener();
|
|
|
+ ReflectionTestUtils.setField(listener, "subscriptionManager", subscriptionManager);
|
|
|
+ ReflectionTestUtils.setField(listener, "equipmentStatusService", equipmentStatusService);
|
|
|
+ ReflectionTestUtils.setField(listener, "webSocketHandler", handler);
|
|
|
+ ReflectionTestUtils.setField(listener, "deltaTracker", new DeviceStatusDeltaTracker());
|
|
|
+ listener.onMessage(message("{\"deviceCode\":\"" + DEVICE_CODE + "\","
|
|
|
+ + "\"serverUpdatedAt\":\"2026-09-02T10:00:00+08:00\"}"), null);
|
|
|
+
|
|
|
+ ArgumentCaptor<TextMessage> messages = ArgumentCaptor.forClass(TextMessage.class);
|
|
|
+ verify(session, org.mockito.Mockito.times(2)).sendMessage(messages.capture());
|
|
|
+ JSONObject delta = messages.getAllValues().stream()
|
|
|
+ .map(item -> JSONObject.parseObject(item.getPayload()))
|
|
|
+ .filter(item -> "cockpitDeviceStatusDelta".equals(item.getString("type")))
|
|
|
+ .findFirst().orElse(null);
|
|
|
+ assertNotNull(delta);
|
|
|
+ assertEquals(DEVICE_CODE, delta.getString("deviceCode"));
|
|
|
+ assertEquals(DEVICE_CODE, delta.getString("device_code"));
|
|
|
+ assertEquals("ALARM", delta.getJSONObject("data").getString("status"));
|
|
|
+ assertEquals("2026-09-02T09:59:50+08:00", delta.getJSONObject("data").getString("statusTime"));
|
|
|
+ assertEquals("2026-09-02T10:00:00+08:00", delta.getJSONObject("data").getString("serverUpdatedAt"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void existingDeviceCodeSubscriptionStillReceivesLegacyPush() throws Exception {
|
|
|
+ handler.afterConnectionEstablished(session);
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"subscribe\",\"deviceCodes\":[\"" + DEVICE_CODE + "\"]}"));
|
|
|
+
|
|
|
+ EquipmentStatus status = new EquipmentStatus();
|
|
|
+ status.setCurrentStatus(1);
|
|
|
+ status.setAlarmStatus(1);
|
|
|
+ status.setOnlineStatus(1);
|
|
|
+ status.setStatusUpdateTime(LocalDateTime.parse("2026-09-02T09:59:50"));
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(status);
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(DEVICE_CODE)).thenReturn(true);
|
|
|
+ DeviceStatusRedisListener listener = newListener();
|
|
|
+ listener.onMessage(message("{\"deviceCode\":\"" + DEVICE_CODE + "\","
|
|
|
+ + "\"serverUpdatedAt\":\"2026-09-02T10:00:00+08:00\"}"), null);
|
|
|
+
|
|
|
+ ArgumentCaptor<TextMessage> messages = ArgumentCaptor.forClass(TextMessage.class);
|
|
|
+ verify(session, org.mockito.Mockito.times(2)).sendMessage(messages.capture());
|
|
|
+ JSONObject push = messages.getAllValues().stream()
|
|
|
+ .map(item -> JSONObject.parseObject(item.getPayload()))
|
|
|
+ .filter(item -> "deviceStatusPush".equals(item.getString("type")))
|
|
|
+ .findFirst().orElse(null);
|
|
|
+ assertNotNull(push);
|
|
|
+ assertEquals(DEVICE_CODE, push.getString("deviceCode"));
|
|
|
+ assertEquals(DEVICE_CODE, push.getJSONObject("data").getString("deviceCode"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void cockpitSubscriptionAcceptsLegacyRedisDeviceCodeAlias() throws Exception {
|
|
|
+ handler.afterConnectionEstablished(session);
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(alarmStatus());
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(DEVICE_CODE)).thenReturn(true);
|
|
|
+ DeviceStatusRedisListener listener = newListener();
|
|
|
+ listener.onMessage(message("{\"device_code\":\"" + DEVICE_CODE + "\","
|
|
|
+ + "\"serverUpdatedAt\":\"2026-09-02T10:00:00+08:00\"}"), null);
|
|
|
+
|
|
|
+ ArgumentCaptor<TextMessage> messages = ArgumentCaptor.forClass(TextMessage.class);
|
|
|
+ verify(session, org.mockito.Mockito.times(2)).sendMessage(messages.capture());
|
|
|
+ JSONObject delta = messages.getAllValues().stream()
|
|
|
+ .map(item -> JSONObject.parseObject(item.getPayload()))
|
|
|
+ .filter(item -> "cockpitDeviceStatusDelta".equals(item.getString("type")))
|
|
|
+ .findFirst().orElse(null);
|
|
|
+ assertNotNull(delta);
|
|
|
+ assertEquals(DEVICE_CODE, delta.getString("deviceCode"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void cockpitAllScopeIgnoresUnchangedDeviceStates() throws Exception {
|
|
|
+ handler.afterConnectionEstablished(session);
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(alarmStatus());
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(DEVICE_CODE)).thenReturn(true);
|
|
|
+ DeviceStatusRedisListener listener = newListener();
|
|
|
+ listener.onMessage(message(event(DEVICE_CODE, "2026-09-02T10:00:00+08:00")), null);
|
|
|
+ listener.onMessage(message(event(DEVICE_CODE, "2026-09-02T10:00:01+08:00")), null);
|
|
|
+
|
|
|
+ verify(session, org.mockito.Mockito.times(2)).sendMessage(ArgumentCaptor.forClass(TextMessage.class).capture());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void cockpitAllScopeIgnoresOldOutOfOrderServerUpdates() throws Exception {
|
|
|
+ handler.afterConnectionEstablished(session);
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(alarmStatus());
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(DEVICE_CODE)).thenReturn(true);
|
|
|
+ DeviceStatusRedisListener listener = newListener();
|
|
|
+ listener.onMessage(message(event(DEVICE_CODE, "2026-09-02T10:00:05+08:00")), null);
|
|
|
+
|
|
|
+ EquipmentStatus recovered = alarmStatus();
|
|
|
+ recovered.setAlarmStatus(0);
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(recovered);
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(DEVICE_CODE)).thenReturn(true);
|
|
|
+ listener.onMessage(message(event(DEVICE_CODE, "2026-09-02T10:00:00+08:00")), null);
|
|
|
+
|
|
|
+ verify(session, org.mockito.Mockito.times(2)).sendMessage(ArgumentCaptor.forClass(TextMessage.class).capture());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void unknownDeviceReturnsExplicitCockpitFallback() throws Exception {
|
|
|
+ handler.afterConnectionEstablished(session);
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+
|
|
|
+ String unknownCode = "TC-PS-WS-202609099999";
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(unknownCode)).thenReturn(false);
|
|
|
+ DeviceStatusRedisListener listener = newListener();
|
|
|
+ listener.onMessage(message(event(unknownCode, "2026-09-02T10:00:00+08:00")), null);
|
|
|
+
|
|
|
+ ArgumentCaptor<TextMessage> messages = ArgumentCaptor.forClass(TextMessage.class);
|
|
|
+ verify(session, org.mockito.Mockito.times(2)).sendMessage(messages.capture());
|
|
|
+ JSONObject delta = messages.getAllValues().stream()
|
|
|
+ .map(item -> JSONObject.parseObject(item.getPayload()))
|
|
|
+ .filter(item -> "cockpitDeviceStatusDelta".equals(item.getString("type")))
|
|
|
+ .findFirst().orElse(null);
|
|
|
+ assertNotNull(delta);
|
|
|
+ assertEquals(unknownCode, delta.getString("deviceCode"));
|
|
|
+ assertEquals("EMPTY", delta.getJSONObject("data").getString("status"));
|
|
|
+ assertEquals("DEVICE_NOT_FOUND", delta.getJSONObject("data").getString("reason"));
|
|
|
+ assertEquals("2026-09-02T10:00:00+08:00", delta.getJSONObject("data").getString("serverUpdatedAt"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void missingPointStatusReturnsExplicitNoStatusFallback() throws Exception {
|
|
|
+ handler.afterConnectionEstablished(session);
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(DEVICE_CODE)).thenReturn(true);
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(null);
|
|
|
+ DeviceStatusRedisListener listener = newListener();
|
|
|
+ listener.onMessage(message(event(DEVICE_CODE, "2026-09-02T10:00:00+08:00")), null);
|
|
|
+
|
|
|
+ ArgumentCaptor<TextMessage> messages = ArgumentCaptor.forClass(TextMessage.class);
|
|
|
+ verify(session, org.mockito.Mockito.times(2)).sendMessage(messages.capture());
|
|
|
+ JSONObject delta = messages.getAllValues().stream()
|
|
|
+ .map(item -> JSONObject.parseObject(item.getPayload()))
|
|
|
+ .filter(item -> "cockpitDeviceStatusDelta".equals(item.getString("type")))
|
|
|
+ .findFirst().orElse(null);
|
|
|
+ assertNotNull(delta);
|
|
|
+ assertEquals("EMPTY", delta.getJSONObject("data").getString("status"));
|
|
|
+ assertEquals("NO_STATUS", delta.getJSONObject("data").getString("reason"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void pingAnswersWithPong() throws Exception {
|
|
|
+ handler.handleTextMessage(session, new TextMessage("{\"type\":\"ping\"}"));
|
|
|
+
|
|
|
+ ArgumentCaptor<TextMessage> response = ArgumentCaptor.forClass(TextMessage.class);
|
|
|
+ verify(session).sendMessage(response.capture());
|
|
|
+ JSONObject payload = JSONObject.parseObject(response.getValue().getPayload());
|
|
|
+ assertEquals("pong", payload.getString("type"));
|
|
|
+ assertTrue(payload.getLongValue("timestamp") > 0);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void disconnectCleansUpSubscriptionAndReconnectRecoversIt() throws Exception {
|
|
|
+ handler.afterConnectionEstablished(session);
|
|
|
+ handler.handleTextMessage(session, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+ handler.afterConnectionClosed(session, CloseStatus.NORMAL);
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(alarmStatus());
|
|
|
+ when(equipmentStatusService.existsByDeviceCode(DEVICE_CODE)).thenReturn(true);
|
|
|
+ DeviceStatusRedisListener listener = newListener();
|
|
|
+ listener.onMessage(message(event(DEVICE_CODE, "2026-09-02T10:00:00+08:00")), null);
|
|
|
+
|
|
|
+ WebSocketSession reconnected = mock(WebSocketSession.class);
|
|
|
+ HashMap<String, Object> attributes = new HashMap<>();
|
|
|
+ org.mockito.Mockito.doReturn("ws-session-2").when(reconnected).getId();
|
|
|
+ org.mockito.Mockito.doReturn(true).when(reconnected).isOpen();
|
|
|
+ org.mockito.Mockito.doReturn(attributes).when(reconnected).getAttributes();
|
|
|
+ handler.afterConnectionEstablished(reconnected);
|
|
|
+ handler.handleTextMessage(reconnected, new TextMessage(
|
|
|
+ "{\"type\":\"cockpitSubscribe\",\"scope\":\"all\"}"));
|
|
|
+
|
|
|
+ when(equipmentStatusService.getByDeviceCode(DEVICE_CODE)).thenReturn(recoveredNormal());
|
|
|
+ listener.onMessage(message(event(DEVICE_CODE, "2026-09-02T10:00:01+08:00")), null);
|
|
|
+
|
|
|
+ ArgumentCaptor<TextMessage> messages = ArgumentCaptor.forClass(TextMessage.class);
|
|
|
+ verify(reconnected, org.mockito.Mockito.times(2)).sendMessage(messages.capture());
|
|
|
+ long deltas = messages.getAllValues().stream()
|
|
|
+ .filter(item -> JSONObject.parseObject(item.getPayload()).getString("type")
|
|
|
+ .equals("cockpitDeviceStatusDelta"))
|
|
|
+ .count();
|
|
|
+ assertEquals(1, deltas);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static Message message(String payload) {
|
|
|
+ return new Message() {
|
|
|
+ @Override
|
|
|
+ public byte[] getBody() {
|
|
|
+ return payload.getBytes(StandardCharsets.UTF_8);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public byte[] getChannel() {
|
|
|
+ return DeviceStatusRedisListener.CHANNEL.getBytes(StandardCharsets.UTF_8);
|
|
|
+ }
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ private DeviceStatusRedisListener newListener() {
|
|
|
+ DeviceStatusRedisListener listener = new DeviceStatusRedisListener();
|
|
|
+ ReflectionTestUtils.setField(listener, "subscriptionManager", subscriptionManager);
|
|
|
+ ReflectionTestUtils.setField(listener, "equipmentStatusService", equipmentStatusService);
|
|
|
+ ReflectionTestUtils.setField(listener, "webSocketHandler", handler);
|
|
|
+ ReflectionTestUtils.setField(listener, "deltaTracker", new DeviceStatusDeltaTracker());
|
|
|
+ return listener;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static String event(String deviceCode, String serverUpdatedAt) {
|
|
|
+ return "{\"deviceCode\":\"" + deviceCode + "\",\"serverUpdatedAt\":\"" + serverUpdatedAt + "\"}";
|
|
|
+ }
|
|
|
+
|
|
|
+ private EquipmentStatus alarmStatus() {
|
|
|
+ EquipmentStatus status = new EquipmentStatus();
|
|
|
+ status.setCurrentStatus(1);
|
|
|
+ status.setAlarmStatus(1);
|
|
|
+ status.setOnlineStatus(1);
|
|
|
+ status.setStatusUpdateTime(LocalDateTime.parse("2026-09-02T09:59:50"));
|
|
|
+ return status;
|
|
|
+ }
|
|
|
+
|
|
|
+ private EquipmentStatus recoveredNormal() {
|
|
|
+ EquipmentStatus status = alarmStatus();
|
|
|
+ status.setAlarmStatus(0);
|
|
|
+ return status;
|
|
|
+ }
|
|
|
+}
|
