package com.battery.business.service.impl;
|
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
import static org.mockito.ArgumentMatchers.any;
|
import static org.mockito.Mockito.mock;
|
import static org.mockito.Mockito.never;
|
import static org.mockito.Mockito.times;
|
import static org.mockito.Mockito.verify;
|
import static org.mockito.Mockito.when;
|
|
import java.time.Clock;
|
import java.time.Duration;
|
import java.time.Instant;
|
import java.time.ZoneId;
|
import java.time.ZoneOffset;
|
import java.util.Date;
|
import java.util.List;
|
import org.junit.jupiter.api.Test;
|
import org.mockito.ArgumentCaptor;
|
import org.mockito.Mockito;
|
import com.battery.device.api.model.AdapterCapabilities;
|
import com.battery.device.api.registry.AdapterInstanceConfig;
|
import com.battery.business.domain.TDeviceAdapter;
|
import com.battery.business.domain.TDeviceIngestRecord;
|
import com.battery.business.domain.TTest;
|
import com.battery.business.domain.TTestBattery;
|
import com.battery.business.domain.TTestChannelBinding;
|
import com.battery.business.domain.TTestRealtimeRecord;
|
import com.battery.business.mapper.TDeviceAdapterMapper;
|
import com.battery.business.mapper.TTestMapper;
|
import com.battery.business.service.ITTestService;
|
|
/**
|
* 推式来源接进业务接收路径端到端：取数程序推来的厂商原始行 → 落地表 → 适配器读回统一观测 →
|
* 接收中测试按水位逐行保存。
|
*
|
* <p>用的是同一个观测契约，所以这里不需要任何推送专属的接收代码：落地行按事件落库、
|
* 轮询槽位为空（同秒多行因此各自成立），来源行身份承担幂等与冲突判定。
|
*/
|
class PushIngestReceptionContractTest
|
{
|
private static final String ADAPTER = "j22";
|
private static final Instant START = Instant.parse("2026-09-27T08:00:00Z");
|
/** 现场实测的一批：两行落在同一个绝对秒（08:03:20 UTC），协议序号不同。 */
|
private static final String BATCH = """
|
{"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
|
{"record_time":"134349698000000000","Data_Point":421,"Step_Time":600.0,"Step_Index":4,
|
"Cycle_Index":1,"Voltage":1.2701,"eCurrent":0.5,"ChargeCapacity":0.121,"ChargeEnergy":0.31},
|
{"record_time":"134349698000000000","Data_Point":422,"Step_Time":600.0,"Step_Index":4,
|
"Cycle_Index":1,"Voltage":1.2702,"eCurrent":0.5,"ChargeCapacity":0.122,"ChargeEnergy":0.32}]}
|
""";
|
/** 同一批里带上已声明的环境量列：辅助测量的值取自这一行自己，不做任何缩放。 */
|
private static final String AUXILIARY_BATCH = """
|
{"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
|
{"record_time":"134349698000000000","Data_Point":421,"Step_Time":600.0,"Step_Index":4,
|
"Cycle_Index":1,"Voltage":1.2701,"eCurrent":0.5,"ChargeCapacity":0.121,"ChargeEnergy":0.31,
|
"Tempbox":26.5},
|
{"record_time":"134349704000000000","Data_Point":422,"Step_Time":660.0,"Step_Index":4,
|
"Cycle_Index":1,"Voltage":1.2702,"eCurrent":0.5,"ChargeCapacity":0.122,"ChargeEnergy":0.32,
|
"Tempbox":26.5}]}
|
""";
|
|
private final FakeIngestLanding landing = new FakeIngestLanding();
|
private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
|
private final TestClock clock = new TestClock(START);
|
|
@Test
|
void pushedRawRowsBecomeTestFactsWithoutPollingSlots()
|
{
|
PushIngestAdapter adapter = ingestPushSource();
|
TTestMapper mapper = mock(TTestMapper.class);
|
ITTestService tests = mock(ITTestService.class);
|
TTest test = receivingTest();
|
when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
|
when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
|
when(tests.selectTestById(1L)).thenReturn(test);
|
TestDataReceptionServiceImpl service = new TestDataReceptionServiceImpl(mapper, tests, adapter, clock);
|
|
// 第一次接收只取水位：开始接收之前的行即使躺在落地表里也不进本测试。
|
landing.insertIngestRecord(boundaryRow("DB_20260927_155209/Ws26_IV_0:420", 420L));
|
service.captureReceivedData();
|
verify(mapper, never()).insertRealtimeRecord(any());
|
|
// 之后推来的行逐行落库：同秒两行各自成立，因为槽位为空（推送记录按事件落库，不占轮询槽位）。
|
ingest(BATCH);
|
clock.advance(Duration.ofSeconds(60));
|
service.captureReceivedData();
|
|
ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(mapper, times(2)).insertRealtimeRecord(rows.capture());
|
assertEquals(List.of("push:DB_20260927_155209/Ws26_IV_0:421",
|
"push:DB_20260927_155209/Ws26_IV_0:422"),
|
rows.getAllValues().stream().map(TTestRealtimeRecord::getSourceRecordId).toList());
|
assertTrue(rows.getAllValues().stream().allMatch(row -> row.getCaptureSlotUtc() == null));
|
assertEquals("j22:0:0:1:26", rows.getAllValues().get(0).getSourceIdentity());
|
assertEquals(Instant.parse("2026-09-27T08:03:20Z"),
|
rows.getAllValues().get(0).getCabinetDeviceTimeUtc().toInstant());
|
assertEquals("Asia/Shanghai", rows.getAllValues().get(0).getCabinetAdapterTimezoneId());
|
// 测量值按物理单位直接使用，不做缩放。
|
assertEquals(0, new java.math.BigDecimal("1.2701")
|
.compareTo(rows.getAllValues().get(0).getVoltageV()));
|
assertEquals(0, new java.math.BigDecimal("0.121")
|
.compareTo(rows.getAllValues().get(0).getCapacityAh()));
|
|
// 同一批再推一次不会重复落库：来源行身份让接收幂等。
|
ingest(BATCH);
|
clock.advance(Duration.ofSeconds(60));
|
service.captureReceivedData();
|
verify(mapper, times(2)).insertRealtimeRecord(any());
|
}
|
|
/**
|
* 采集器辅助测量按 ADR-0064 由同一来源提供：它作为扩展测量记录进入实时记录，
|
* **不跟检测柜记录伪造配对**——检测柜走各自的设备数据行（槽位为空），采集器走轮询槽位行（柜端字段为空），
|
* 两边各自保留自己的设备时间，绑定角色决定它进哪个指标。
|
*/
|
@Test
|
void collectorAuxiliaryMeasurementBecomesItsOwnRealtimeRowWithoutFakingAPairing()
|
{
|
PushIngestAdapter adapter = ingestPushSource("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\",\"auxiliaryMetrics\":"
|
+ "[{\"vendorChannel\":26,\"channelCode\":\"S-01\",\"column\":\"Tempbox\","
|
+ "\"metricCode\":\"TEMPERATURE\",\"unit\":\"℃\"}]}");
|
TTestMapper mapper = mock(TTestMapper.class);
|
ITTestService tests = mock(ITTestService.class);
|
TTest test = receivingTest();
|
TTestBattery battery = test.getBatteries().get(0);
|
battery.setCollectorId(7L);
|
com.battery.business.domain.TTestCollectorChannel collectorChannel =
|
new com.battery.business.domain.TTestCollectorChannel();
|
collectorChannel.setChannelCode("S-01");
|
collectorChannel.setMeasurementCode("PRIMARY_TEMPERATURE");
|
battery.setCollectorChannels(List.of(collectorChannel));
|
when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
|
when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
|
when(tests.selectTestById(1L)).thenReturn(test);
|
when(mapper.selectCollectorCode(7L)).thenReturn("j22-collector");
|
TestDataReceptionServiceImpl service = new TestDataReceptionServiceImpl(mapper, tests, adapter, clock);
|
|
landing.insertIngestRecord(boundaryRow("DB_20260927_155209/Ws26_IV_0:420", 420L));
|
service.captureReceivedData();
|
ingest(AUXILIARY_BATCH);
|
clock.advance(Duration.ofSeconds(60));
|
service.captureReceivedData();
|
|
ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(mapper, times(3)).insertRealtimeRecord(rows.capture());
|
List<TTestRealtimeRecord> cabinetRows = rows.getAllValues().subList(0, 2);
|
TTestRealtimeRecord collectorRow = rows.getAllValues().get(2);
|
// 检测柜事实不占轮询槽位，采集器事实占槽位：两条路径互不覆盖。
|
assertTrue(cabinetRows.stream().allMatch(row -> row.getCaptureSlotUtc() == null));
|
assertTrue(collectorRow.getCaptureSlotUtc() != null);
|
// 采集器行不携带检测柜设备时间与检测柜身份，来源行身份里也没有 cabinet 段。
|
assertEquals(null, collectorRow.getCabinetDeviceTimeUtc());
|
assertEquals(null, collectorRow.getCabinetChlid());
|
assertTrue(collectorRow.getSourceRecordId().indexOf("cabinet:") < 0);
|
// 测量值取自这一行自己的设备时间（该通道落地表里最新一行），指标编码与单位按配置声明，
|
// 固定指标落到自己的列。
|
assertEquals(Instant.parse("2026-09-27T08:13:20Z"), collectorRow.getCollectorDeviceTimeUtc().toInstant());
|
assertEquals("S-01", collectorRow.getCollectorChannelCode());
|
assertEquals(0, new java.math.BigDecimal("26.5").compareTo(collectorRow.getTemperature()));
|
assertEquals(Long.valueOf(7L), collectorRow.getCollectorId());
|
}
|
|
private PushIngestAdapter ingestPushSource()
|
{
|
return ingestPushSource("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\"}");
|
}
|
|
private PushIngestAdapter ingestPushSource(String sourceConfig)
|
{
|
TDeviceAdapter source = new TDeviceAdapter();
|
source.setAdapterCode(ADAPTER);
|
source.setAdapterType(PushIngestAdapter.ADAPTER_TYPE);
|
source.setStatus("0");
|
source.setTimezoneId("Asia/Shanghai");
|
source.setSourceConfig(sourceConfig);
|
when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(source);
|
ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
|
return new PushIngestAdapter(new AdapterInstanceConfig(ADAPTER, PushIngestAdapter.ADAPTER_TYPE,
|
"127.0.0.1", 8080, 60000, "Asia/Shanghai", true, AdapterCapabilities.observeOnly(),
|
com.alibaba.fastjson2.JSON.parseObject(sourceConfig)
|
.toJavaObject(new com.alibaba.fastjson2.TypeReference<java.util.Map<String, String>>() { })),
|
landing, clock);
|
}
|
|
private void ingest(String body)
|
{
|
new DeviceIngestServiceImpl(adapters, landing, clock, key -> "token").ingest(ADAPTER, "token", body);
|
}
|
|
/** 开始接收之前就躺在来源里的那一行：它只用来把水位定在边界时刻。 */
|
private TDeviceIngestRecord boundaryRow(String recordKey, long sequence)
|
{
|
TDeviceIngestRecord value = new TDeviceIngestRecord();
|
value.setAdapterCode(ADAPTER);
|
value.setPayloadKind(TDeviceIngestRecord.KIND_SAMPLE);
|
value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(26);
|
value.setRecordSequence(sequence);
|
value.setRecordKey(recordKey);
|
value.setSourceRef("DB_20260927_155209/Ws26_IV_0");
|
value.setDeviceTimeUtc(Date.from(START.minusSeconds(60)));
|
value.setDeviceTimeRaw("134349691400000000");
|
value.setTimezoneId("Asia/Shanghai");
|
value.setPayloadJson("{}");
|
value.setContentSha256(new byte[32]);
|
value.setReceivedAtUtc(Date.from(START.minusSeconds(60)));
|
return value;
|
}
|
|
private static TTest receivingTest()
|
{
|
TTestChannelBinding binding = new TTestChannelBinding();
|
binding.setBindingId(2L);
|
binding.setChannelId(3L);
|
binding.setAdapterCode(ADAPTER);
|
binding.setAdapterTimezoneId("Asia/Shanghai");
|
binding.setDevtype(0);
|
binding.setDevid(0);
|
binding.setSubdevid(1);
|
binding.setChlid(26);
|
TTestBattery battery = new TTestBattery();
|
battery.setTestBatteryId(4L);
|
battery.setBinding(binding);
|
battery.setCollectorChannels(List.of());
|
TTest test = new TTest();
|
test.setTestId(1L);
|
test.setTestStatus("RUNNING");
|
test.setReceptionStatus("RECEIVING");
|
test.setStartedAtUtc(Date.from(START));
|
test.setReceptionBoundaryAtUtc(Date.from(START));
|
test.setBatteries(List.of(battery));
|
return test;
|
}
|
|
/** 可推进的测试时钟：接收节奏与推送节奏都按它推进。 */
|
private static final class TestClock extends Clock
|
{
|
private Instant now;
|
|
private TestClock(Instant now) { this.now = now; }
|
|
private void advance(Duration duration) { now = now.plus(duration); }
|
|
@Override public ZoneId getZone() { return ZoneOffset.UTC; }
|
@Override public Clock withZone(ZoneId zone) { return this; }
|
@Override public Instant instant() { return now; }
|
}
|
}
|
