package com.battery.business.service.impl;
|
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
|
import java.time.Clock;
|
import java.time.Instant;
|
import java.time.ZoneOffset;
|
import java.util.Date;
|
import org.junit.jupiter.api.Test;
|
import com.ruoyi.common.exception.ServiceException;
|
import com.battery.business.domain.TDeviceAdapter;
|
import com.battery.business.domain.TDeviceIngestRecord;
|
import com.battery.business.mapper.TDeviceAdapterMapper;
|
import com.battery.business.service.DeviceAdapterErrorCodes;
|
import com.battery.business.service.DeviceIngestErrorCodes;
|
import org.mockito.Mockito;
|
|
/**
|
* 推式来源接收端点的判定口径：凭据、来源状态与载荷不合法时**整批不落库**（取数程序因此不推进水位），
|
* 合法批次里厂商原始行原样落地、身份与时间由后端解析、重复推送幂等、同身份不同内容不覆盖。
|
*/
|
class DeviceIngestServiceTest
|
{
|
private static final String ADAPTER = "j22";
|
private static final String TOKEN = "j22-token";
|
/** 现场实测的一行：FILETIME 134349692000000000 等于 2026-09-27 07:53:20 UTC（北京 15:53）。 */
|
private static final String SAMPLE_BATCH = """
|
{"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
|
{"record_time":"134349692000000000","Data_Point":420,"Schedule_Index":2818575540,
|
"Test_Time":600.0,"Step_Time":540.0,"Step_Index":4,"Cycle_Index":1,
|
"Voltage":1.26778,"eCurrent":0.5,"ChargeCapacity":0.12,"DischargeCapacity":0.0,
|
"ChargeEnergy":0.3,"DischargeEnergy":0.0,"IR":0.01,"Power":0.63,"Tempbox":25.4}]}
|
""";
|
|
private final FakeIngestLanding landing = new FakeIngestLanding();
|
private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
|
private final Clock clock = Clock.fixed(Instant.parse("2026-09-27T09:00:00Z"), ZoneOffset.UTC);
|
private final DeviceIngestServiceImpl service =
|
new DeviceIngestServiceImpl(adapters, landing, clock, key -> TOKEN);
|
|
@Test
|
void unknownSourceIsRejectedWithoutWritingAnything()
|
{
|
var failure = assertThrows(ServiceException.class, () -> service.ingest("mystery", TOKEN, "{}"));
|
|
assertEquals(DeviceAdapterErrorCodes.NOT_FOUND, failure.getCode());
|
assertTrue(landing.rows().isEmpty());
|
}
|
|
@Test
|
void sourceThatIsNotAPushSourceOrIsDisabledIsRejectedWithoutWritingAnything()
|
{
|
Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("NEWARE_MYSQL", "0"));
|
var wrongType = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH));
|
assertEquals(DeviceAdapterErrorCodes.UNSUPPORTED_TYPE, wrongType.getCode());
|
|
Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "1"));
|
var disabled = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH));
|
assertEquals(DeviceIngestErrorCodes.SOURCE_DISABLED, disabled.getCode());
|
|
assertTrue(landing.rows().isEmpty());
|
}
|
|
/** 凭据不入库：实例配置里只有变量名，取值缺失同样按凭据错误拒绝（失败关闭）。 */
|
@Test
|
void wrongMissingOrUnresolvableTokenIsRejectedWithoutWritingAnything()
|
{
|
Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));
|
|
assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
|
assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, "wrong", SAMPLE_BATCH)).getCode());
|
assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
|
assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, null, SAMPLE_BATCH)).getCode());
|
|
DeviceIngestServiceImpl withoutCredential = new DeviceIngestServiceImpl(adapters, landing, clock,
|
key -> { throw new IllegalStateException("缺少来源凭据：" + key); });
|
assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
|
assertThrows(ServiceException.class, () -> withoutCredential.ingest(ADAPTER, TOKEN, SAMPLE_BATCH))
|
.getCode());
|
|
assertTrue(landing.rows().isEmpty());
|
}
|
|
/** 请求体承载厂商原始行：不含已换算的数值，单位换算、时间换算、身份映射与幂等键都在后端完成。 */
|
@Test
|
void rawVendorRowLandsWithResolvedIdentityTimeAndUnscaledMeasurements()
|
{
|
Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));
|
|
var result = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH);
|
|
assertEquals(1, result.accepted());
|
TDeviceIngestRecord row = landing.rows().get(0);
|
// 身份映射：22 所没有设备类型与单元维度，一台电脑登记为一个柜，通道号就是厂商通道号。
|
assertEquals(0, row.getDevtype());
|
assertEquals(0, row.getDevid());
|
assertEquals(1, row.getSubdevid());
|
assertEquals(26, row.getChlid());
|
// 时间换算：FILETIME → UTC，同时保留设备给出的原始时间文本。
|
assertEquals(Instant.parse("2026-09-27T07:53:20Z"), row.getDeviceTimeUtc().toInstant());
|
assertEquals("134349692000000000", row.getDeviceTimeRaw());
|
// 幂等键：库/表 + Data_Point（22 所每个测试新建一个库，Data_Point 在每个测试内重新编号）。
|
assertEquals(420L, row.getRecordSequence());
|
assertEquals("DB_20260927_155209/Ws26_IV_0:420", row.getRecordKey());
|
assertEquals("DB_20260927_155209/Ws26_IV_0", row.getSourceRef());
|
// 测量值按物理单位直接落库，不做缩放：1.26778 V 就是 1.26778 V。
|
assertEquals(0, new java.math.BigDecimal("1.26778").compareTo(row.getVoltageV()));
|
assertEquals(0, new java.math.BigDecimal("0.5").compareTo(row.getCurrentA()));
|
assertEquals(0, new java.math.BigDecimal("0.12").compareTo(row.getChargeCapacityAh()));
|
assertEquals(1, row.getCycleNo());
|
assertEquals(4, row.getStepNo());
|
assertEquals(0, new java.math.BigDecimal("540.0").compareTo(row.getRelativeTimeSeconds()));
|
// 原始行的键值原样保留，可供回放与核对。
|
assertTrue(row.getPayloadJson().contains("Data_Point"));
|
assertTrue(row.getPayloadJson().contains("\"Schedule_Index\":2818575540"));
|
}
|
|
@Test
|
void repeatedPushIsIdempotentAndConflictingContentIsReportedWithoutOverwriting()
|
{
|
Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));
|
|
assertEquals(1, service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH).accepted());
|
// 同一批再推一次：同一条记录被幂等忽略，采样行仍是一行。
|
var repeat = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH);
|
assertEquals(0, repeat.accepted());
|
assertEquals(1, repeat.ignored());
|
assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());
|
|
// 同身份但内容不同：报冲突且不覆盖已有事实。
|
var conflict = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH.replace("1.26778", "1.99"));
|
assertEquals(1, conflict.conflicts());
|
assertEquals(0, conflict.accepted());
|
assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());
|
assertEquals(0, new java.math.BigDecimal("1.26778")
|
.compareTo(landing.samples(ADAPTER, 0, 0, 1, 26).get(0).getVoltageV()));
|
}
|
|
@Test
|
void topologyHeartbeatAndSamplesShareTheSameLandingTable()
|
{
|
Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));
|
|
service.ingest(ADAPTER, TOKEN, "{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26},{\"channel\":18}]}");
|
// 心跳只刷新存活事实，不产生数据行。
|
var heartbeat = service.ingest(ADAPTER, TOKEN, "{\"kind\":\"HEARTBEAT\"}");
|
assertEquals(0, heartbeat.accepted());
|
assertEquals(3, landing.rows().size());
|
assertEquals(clock.instant(), landing.selectHeartbeatAt(ADAPTER).toInstant());
|
|
// 收到采样行同样是"程序还在推"的证据：接受请求就会刷新心跳，不依赖心跳消息本身。
|
Date before = landing.selectHeartbeatAt(ADAPTER);
|
assertNotNull(before);
|
assertEquals(1, service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH).accepted());
|
assertEquals(4, landing.rows().size());
|
assertEquals(before, landing.selectHeartbeatAt(ADAPTER));
|
// 拓扑上报的通道清单是长期声明，重复上报幂等；采样行按事件逐行落库。
|
assertEquals(2, landing.selectTopologyChannels(ADAPTER).size());
|
assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());
|
}
|
|
@Test
|
void malformedBatchIsRejectedWithTheOffendingRowNamed()
|
{
|
Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));
|
|
for (String body : new String[] {
|
"{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26}",
|
"{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"rows\":[]}",
|
"{\"kind\":\"SAMPLE\",\"channel\":26,\"rows\":[]}",
|
"{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26,"
|
+ "\"rows\":[{\"Voltage\":1.2}]}",
|
"{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26,"
|
+ "\"rows\":[{\"Data_Point\":1}]}",
|
"{\"kind\":\"SOMETHING_ELSE\"}",
|
"not-json" })
|
{
|
var failure = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, body), body);
|
assertEquals(DeviceIngestErrorCodes.PAYLOAD_INVALID, failure.getCode(), body);
|
}
|
assertTrue(landing.rows().isEmpty());
|
}
|
|
private static TDeviceAdapter adapter(String type, String status)
|
{
|
TDeviceAdapter adapter = new TDeviceAdapter();
|
adapter.setAdapterId(7L);
|
adapter.setAdapterCode(ADAPTER);
|
adapter.setAdapterName("22 所那台");
|
adapter.setAdapterType(type);
|
adapter.setStatus(status);
|
adapter.setHost("127.0.0.1");
|
adapter.setPort(8080);
|
adapter.setTimezoneId("Asia/Shanghai");
|
adapter.setSourceConfig("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\"}");
|
return adapter;
|
}
|
}
|
