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 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());
|
}
|
|
private PushIngestAdapter ingestPushSource()
|
{
|
TDeviceAdapter source = new TDeviceAdapter();
|
source.setAdapterCode(ADAPTER);
|
source.setAdapterType(PushIngestAdapter.ADAPTER_TYPE);
|
source.setStatus("0");
|
source.setTimezoneId("Asia/Shanghai");
|
source.setSourceConfig("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\"}");
|
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(),
|
java.util.Map.of("tokenEnv", "BATTERY_INGEST_TOKEN")), 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; }
|
}
|
}
|
