|
|
@@ -0,0 +1,214 @@
|
|
|
1
|
+package com.water.data_engine.service;
|
|
|
2
|
+
|
|
|
3
|
+import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
4
|
+import com.water.data_engine.mapper.CollectRecordMapper;
|
|
|
5
|
+import com.water.data_engine.mapper.CollectTaskMapper;
|
|
|
6
|
+import com.water.data_engine.mapper.DataSourceMapper;
|
|
|
7
|
+import org.junit.jupiter.api.BeforeEach;
|
|
|
8
|
+import org.junit.jupiter.api.DisplayName;
|
|
|
9
|
+import org.junit.jupiter.api.Test;
|
|
|
10
|
+import org.junit.jupiter.api.extension.ExtendWith;
|
|
|
11
|
+import org.mockito.Mock;
|
|
|
12
|
+import org.mockito.junit.jupiter.MockitoExtension;
|
|
|
13
|
+import org.springframework.jdbc.core.JdbcTemplate;
|
|
|
14
|
+import org.springframework.kafka.core.KafkaTemplate;
|
|
|
15
|
+import org.springframework.messaging.simp.SimpMessagingTemplate;
|
|
|
16
|
+
|
|
|
17
|
+import java.util.HashMap;
|
|
|
18
|
+import java.util.List;
|
|
|
19
|
+import java.util.Map;
|
|
|
20
|
+
|
|
|
21
|
+import static org.junit.jupiter.api.Assertions.*;
|
|
|
22
|
+import static org.mockito.ArgumentMatchers.*;
|
|
|
23
|
+import static org.mockito.Mockito.*;
|
|
|
24
|
+
|
|
|
25
|
+/**
|
|
|
26
|
+ * Kafka 消费者测试
|
|
|
27
|
+ */
|
|
|
28
|
+@ExtendWith(MockitoExtension.class)
|
|
|
29
|
+class KafkaConsumerTest {
|
|
|
30
|
+
|
|
|
31
|
+ @Mock
|
|
|
32
|
+ private KafkaTemplate<String, String> kafkaTemplate;
|
|
|
33
|
+
|
|
|
34
|
+ @Mock
|
|
|
35
|
+ private JdbcTemplate jdbcTemplate;
|
|
|
36
|
+
|
|
|
37
|
+ @Mock
|
|
|
38
|
+ private DataSourceMapper dataSourceMapper;
|
|
|
39
|
+
|
|
|
40
|
+ @Mock
|
|
|
41
|
+ private CollectTaskMapper collectTaskMapper;
|
|
|
42
|
+
|
|
|
43
|
+ @Mock
|
|
|
44
|
+ private CollectRecordMapper collectRecordMapper;
|
|
|
45
|
+
|
|
|
46
|
+ @Mock
|
|
|
47
|
+ private SimpMessagingTemplate wsMessagingTemplate;
|
|
|
48
|
+
|
|
|
49
|
+ private DataCollectService collectService;
|
|
|
50
|
+
|
|
|
51
|
+ private ObjectMapper objectMapper;
|
|
|
52
|
+
|
|
|
53
|
+ @BeforeEach
|
|
|
54
|
+ void setUp() {
|
|
|
55
|
+ collectService = new DataCollectService(
|
|
|
56
|
+ kafkaTemplate, jdbcTemplate, dataSourceMapper,
|
|
|
57
|
+ collectTaskMapper, collectRecordMapper, wsMessagingTemplate
|
|
|
58
|
+ );
|
|
|
59
|
+ objectMapper = new ObjectMapper();
|
|
|
60
|
+ }
|
|
|
61
|
+
|
|
|
62
|
+ @Test
|
|
|
63
|
+ @DisplayName("Kafka消费-IoT原始数据")
|
|
|
64
|
+ void testConsumeIotRaw() {
|
|
|
65
|
+ // Given
|
|
|
66
|
+ String deviceSn = "FM001";
|
|
|
67
|
+ String message = buildIotTelemetryMessage(deviceSn);
|
|
|
68
|
+
|
|
|
69
|
+ // When
|
|
|
70
|
+ collectService.consumeIotRaw(message);
|
|
|
71
|
+
|
|
|
72
|
+ // Then
|
|
|
73
|
+ verify(jdbcTemplate, times(3)).update(
|
|
|
74
|
+ eq("INSERT INTO water_iot.iot_telemetry (ts, device_sn, metric_key, metric_value, quality) VALUES (NOW, ?, ?, ?, 1)"),
|
|
|
75
|
+ eq(deviceSn),
|
|
|
76
|
+ anyString(),
|
|
|
77
|
+ any()
|
|
|
78
|
+ );
|
|
|
79
|
+ }
|
|
|
80
|
+
|
|
|
81
|
+ @Test
|
|
|
82
|
+ @DisplayName("Kafka消费-水质数据")
|
|
|
83
|
+ void testConsumeQualityData() {
|
|
|
84
|
+ // Given
|
|
|
85
|
+ String testPoint = "水厂出口";
|
|
|
86
|
+ String message = buildQualityDataMessage(testPoint);
|
|
|
87
|
+
|
|
|
88
|
+ // When
|
|
|
89
|
+ collectService.consumeQualityData(message);
|
|
|
90
|
+
|
|
|
91
|
+ // Then
|
|
|
92
|
+ verify(jdbcTemplate).update(
|
|
|
93
|
+ eq("INSERT INTO water_quality_record (test_type, test_point, point_type, area, " +
|
|
|
94
|
+ "turbidity, ph, residual_chlorine, is_qualified, created_at) " +
|
|
|
95
|
+ "VALUES (?, ?, ?, ?, ?, ?, ?, ?, NOW())"),
|
|
|
96
|
+ any(),
|
|
|
97
|
+ eq(testPoint),
|
|
|
98
|
+ any(),
|
|
|
99
|
+ any(),
|
|
|
100
|
+ any(),
|
|
|
101
|
+ any(),
|
|
|
102
|
+ any()
|
|
|
103
|
+ );
|
|
|
104
|
+ }
|
|
|
105
|
+
|
|
|
106
|
+ @Test
|
|
|
107
|
+ @DisplayName("数据验证-合格数据")
|
|
|
108
|
+ void testDataValidation_ValidData() {
|
|
|
109
|
+ Map<String, Object> validData = new HashMap<>();
|
|
|
110
|
+ validData.put("deviceSn", "FM001");
|
|
|
111
|
+ validData.put("timestamp", System.currentTimeMillis());
|
|
|
112
|
+ validData.put("metrics", List.of(
|
|
|
113
|
+ Map.of("key", "LL", "value", 12.5),
|
|
|
114
|
+ Map.of("key", "YL", "value", 0.35),
|
|
|
115
|
+ Map.of("key", "PH", "value", 7.2)
|
|
|
116
|
+ ));
|
|
|
117
|
+
|
|
|
118
|
+ // 使用反射访问私有方法
|
|
|
119
|
+ boolean result = collectService.validateData("iot", validData);
|
|
|
120
|
+ assertTrue(result, "合格数据应该通过验证");
|
|
|
121
|
+ }
|
|
|
122
|
+
|
|
|
123
|
+ @Test
|
|
|
124
|
+ @DisplayName("数据验证-无效设备编号")
|
|
|
125
|
+ void testDataValidation_InvalidDeviceSn() {
|
|
|
126
|
+ Map<String, Object> invalidData = new HashMap<>();
|
|
|
127
|
+ invalidData.put("deviceSn", "INVALID_DEVICE"); // 超过20个字符
|
|
|
128
|
+ invalidData.put("timestamp", System.currentTimeMillis());
|
|
|
129
|
+ invalidData.put("metrics", List.of(
|
|
|
130
|
+ Map.of("key", "LL", "value", 12.5)
|
|
|
131
|
+ ));
|
|
|
132
|
+
|
|
|
133
|
+ boolean result = collectService.validateData("iot", invalidData);
|
|
|
134
|
+ assertFalse(result, "无效设备编号应该无法通过验证");
|
|
|
135
|
+ }
|
|
|
136
|
+
|
|
|
137
|
+ @Test
|
|
|
138
|
+ @DisplayName("数据验证-数值超出范围")
|
|
|
139
|
+ void testDataValidation_InvalidValue() {
|
|
|
140
|
+ Map<String, Object> invalidData = new HashMap<>();
|
|
|
141
|
+ invalidData.put("deviceSn", "FM001");
|
|
|
142
|
+ invalidData.put("timestamp", System.currentTimeMillis());
|
|
|
143
|
+ invalidData.put("metrics", List.of(
|
|
|
144
|
+ Map.of("key", "LL", "value", 999999) // 流量超出合理范围
|
|
|
145
|
+ ));
|
|
|
146
|
+
|
|
|
147
|
+ boolean result = collectService.validateData("iot", invalidData);
|
|
|
148
|
+ assertFalse(result, "超出范围的数值应该无法通过验证");
|
|
|
149
|
+ }
|
|
|
150
|
+
|
|
|
151
|
+ @Test
|
|
|
152
|
+ @DisplayName("Topic路由测试")
|
|
|
153
|
+ void testRouteTopic() {
|
|
|
154
|
+ assertEquals("iot.raw.generic", collectService.routeTopic("iot"));
|
|
|
155
|
+ assertEquals("iot.raw.generic", collectService.routeTopic("mqtt"));
|
|
|
156
|
+ assertEquals("data.quality", collectService.routeTopic("quality"));
|
|
|
157
|
+ assertEquals("data.manual", collectService.routeTopic("manual"));
|
|
|
158
|
+ assertEquals("data.api", collectService.routeTopic("api"));
|
|
|
159
|
+ assertEquals("data.raw", collectService.routeTopic("unknown"));
|
|
|
160
|
+ }
|
|
|
161
|
+
|
|
|
162
|
+ /**
|
|
|
163
|
+ * 构建IoT遥测数据消息
|
|
|
164
|
+ */
|
|
|
165
|
+ private String buildIotTelemetryMessage(String deviceSn) {
|
|
|
166
|
+ Map<String, Object> data = new HashMap<>();
|
|
|
167
|
+ data.put("deviceSn", deviceSn);
|
|
|
168
|
+ data.put("timestamp", System.currentTimeMillis());
|
|
|
169
|
+ data.put("metrics", List.of(
|
|
|
170
|
+ Map.of("key", "LL", "value", 12.5),
|
|
|
171
|
+ Map.of("key", "YL", "value", 0.35),
|
|
|
172
|
+ Map.of("key", "PH", "value", 7.2)
|
|
|
173
|
+ ));
|
|
|
174
|
+
|
|
|
175
|
+ Map<String, Object> envelope = new HashMap<>();
|
|
|
176
|
+ envelope.put("sourceType", "iot");
|
|
|
177
|
+ envelope.put("sourceId", deviceSn);
|
|
|
178
|
+ envelope.put("timestamp", System.currentTimeMillis());
|
|
|
179
|
+ envelope.put("data", data);
|
|
|
180
|
+
|
|
|
181
|
+ try {
|
|
|
182
|
+ return objectMapper.writeValueAsString(envelope);
|
|
|
183
|
+ } catch (Exception e) {
|
|
|
184
|
+ throw new RuntimeException("构建测试消息失败", e);
|
|
|
185
|
+ }
|
|
|
186
|
+ }
|
|
|
187
|
+
|
|
|
188
|
+ /**
|
|
|
189
|
+ * 构建水质数据消息
|
|
|
190
|
+ */
|
|
|
191
|
+ private String buildQualityDataMessage(String testPoint) {
|
|
|
192
|
+ Map<String, Object> data = new HashMap<>();
|
|
|
193
|
+ data.put("testType", "常规检测");
|
|
|
194
|
+ data.put("testPoint", testPoint);
|
|
|
195
|
+ data.put("pointType", "出厂水");
|
|
|
196
|
+ data.put("area", "主城区");
|
|
|
197
|
+ data.put("turbidity", 0.5);
|
|
|
198
|
+ data.put("ph", 7.2);
|
|
|
199
|
+ data.put("residualChlorine", 0.3);
|
|
|
200
|
+ data.put("isQualified", true);
|
|
|
201
|
+
|
|
|
202
|
+ Map<String, Object> envelope = new HashMap<>();
|
|
|
203
|
+ envelope.put("sourceType", "quality");
|
|
|
204
|
+ envelope.put("sourceId", "WQ001");
|
|
|
205
|
+ envelope.put("timestamp", System.currentTimeMillis());
|
|
|
206
|
+ envelope.put("data", data);
|
|
|
207
|
+
|
|
|
208
|
+ try {
|
|
|
209
|
+ return objectMapper.writeValueAsString(envelope);
|
|
|
210
|
+ } catch (Exception e) {
|
|
|
211
|
+ throw new RuntimeException("构建测试消息失败", e);
|
|
|
212
|
+ }
|
|
|
213
|
+ }
|
|
|
214
|
+}
|
