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.ArrayList;
|
import java.util.Date;
|
import java.util.HexFormat;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import org.junit.jupiter.api.Test;
|
import org.mockito.ArgumentCaptor;
|
import com.battery.device.api.TestDataObservationAdapter;
|
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
|
import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
|
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
|
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
|
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
|
import com.battery.device.api.model.AdapterCapabilities;
|
import com.battery.device.api.model.ChannelIdentity;
|
import com.battery.device.api.model.DetectedChannel;
|
import com.battery.device.api.model.DeviceObservation;
|
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.TTestMapper;
|
import com.battery.business.service.ITTestService;
|
|
/**
|
* 逐行接收的归属边界：水位由"开始接收"与"继续接收"这一刻来源里的最新一行决定，
|
* 只取严格晚于水位的数据行；同秒多行都保留；重复同内容幂等；同身份不同内容不覆盖。
|
*/
|
class VendorSampleReceptionServiceTest
|
{
|
private static final Instant START = Instant.parse("2026-09-28T02:00:00Z");
|
private static final ChannelIdentity CHANNEL = new ChannelIdentity("neware76", 21, 36, 1, 2);
|
private static final String IDENTITY = "neware76:21:36:1:2";
|
|
@Test
|
void firstPollReadsTheWatermarkWithoutWritingDeviceHistory()
|
{
|
Fixture fixture = fixture();
|
fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
|
|
fixture.service.captureReceivedData();
|
|
assertEquals(1, fixture.adapter.watermarkReads);
|
assertTrue(fixture.adapter.sampleReads.isEmpty());
|
verify(fixture.mapper, never()).insertRealtimeRecord(any());
|
}
|
|
@Test
|
void everyRowStrictlyAfterTheWatermarkBecomesATestFactIncludingSameSecondRows()
|
{
|
Fixture fixture = fixture();
|
fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
|
fixture.service.captureReceivedData();
|
|
// 同秒两行都落库：槽位留空，来源行身份把它们区分开。
|
fixture.adapter.samples = List.of(sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"),
|
sample(7438L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"));
|
fixture.clock.advance(Duration.ofSeconds(60));
|
fixture.service.captureReceivedData();
|
|
ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(fixture.mapper, times(2)).insertRealtimeRecord(rows.capture());
|
assertEquals(List.of("neware:440:7437", "neware:440:7438"),
|
rows.getAllValues().stream().map(TTestRealtimeRecord::getSourceRecordId).toList());
|
assertTrue(rows.getAllValues().stream().allMatch(row -> row.getCaptureSlotUtc() == null));
|
assertEquals(Instant.parse("2026-09-28T01:55:56Z"),
|
rows.getAllValues().get(0).getCabinetDeviceTimeUtc().toInstant());
|
assertEquals("2026-09-28 09:55:56", rows.getAllValues().get(0).getCabinetDeviceTimeRaw());
|
assertEquals(IDENTITY, rows.getAllValues().get(0).getSourceIdentity());
|
}
|
|
@Test
|
void resumingReceptionReadsANewWatermarkSoThePauseGapIsNotBackfilled()
|
{
|
Fixture fixture = fixture();
|
fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
|
fixture.service.captureReceivedData();
|
|
fixture.adapter.samples = List.of(sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"));
|
fixture.clock.advance(Duration.ofSeconds(60));
|
fixture.service.captureReceivedData();
|
assertEquals(List.of(new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L)),
|
fixture.adapter.sampleReads);
|
|
// 暂停后继续：接收边界推进，水位重新读取；暂停期间产生的行不再进本测试。
|
fixture.test.setReceptionBoundaryAtUtc(Date.from(START.plusSeconds(600)));
|
fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T02:05:56Z"), 7500L);
|
fixture.adapter.samples = List.of();
|
fixture.clock.advance(Duration.ofSeconds(60));
|
fixture.service.captureReceivedData();
|
|
assertEquals(2, fixture.adapter.watermarkReads);
|
// 取到新水位后，只追加新水位之后的行；暂停期间的空档不会被补回来。
|
fixture.clock.advance(Duration.ofSeconds(60));
|
fixture.service.captureReceivedData();
|
assertEquals(List.of(new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L),
|
new SampleWatermark(Instant.parse("2026-09-28T02:05:56Z"), 7500L)), fixture.adapter.sampleReads);
|
}
|
|
@Test
|
void repeatedIdenticalRowIsIgnoredAndConflictingContentIsNotOverwritten()
|
{
|
Fixture fixture = fixture();
|
fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
|
fixture.service.captureReceivedData();
|
|
CabinetRealtime sample = sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56");
|
fixture.adapter.samples = List.of(sample);
|
fixture.clock.advance(Duration.ofSeconds(60));
|
fixture.service.captureReceivedData();
|
ArgumentCaptor<TTestRealtimeRecord> first = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(fixture.mapper).insertRealtimeRecord(first.capture());
|
|
// 同身份同内容：幂等忽略，不再插入。
|
when(fixture.mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"))
|
.thenReturn(HexFormat.of().formatHex(first.getValue().getContentSha256()));
|
fixture.adapter.watermark = new SampleWatermark(sample.deviceTime(), sample.sequenceId());
|
fixture.clock.advance(Duration.ofSeconds(60));
|
fixture.service.captureReceivedData();
|
verify(fixture.mapper, times(1)).insertRealtimeRecord(any());
|
|
// 同身份不同内容：保留原事实，拒绝覆盖。
|
when(fixture.mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"))
|
.thenReturn("0000000000000000000000000000000000000000000000000000000000000000");
|
fixture.clock.advance(Duration.ofSeconds(60));
|
fixture.service.captureReceivedData();
|
verify(fixture.mapper, times(1)).insertRealtimeRecord(any());
|
}
|
|
@Test
|
void sourceWithoutWatermarkKeepsThePollingSlotPath()
|
{
|
Fixture fixture = fixture();
|
// 不提供水位的来源（例如开发模拟器）继续按轮询槽位保存柜端事实。
|
fixture.adapter.watermark = null;
|
fixture.adapter.latest = sample(9L, "2026-09-28T02:00:30Z", "2026-09-28 10:00:30");
|
|
fixture.service.captureReceivedData();
|
|
assertEquals(1, fixture.adapter.watermarkReads);
|
assertTrue(fixture.adapter.sampleReads.isEmpty());
|
ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(fixture.mapper).insertRealtimeRecord(rows.capture());
|
assertTrue(rows.getValue().getCaptureSlotUtc() != null);
|
}
|
|
private static CabinetRealtime sample(long sequenceId, String deviceTime, String raw)
|
{
|
return new CabinetRealtime(sequenceId, Instant.parse(deviceTime), -1.1889, 1.3733, 1.398611, 0.030648,
|
3, 12, null, "2", Instant.parse("2026-09-28T02:01:00Z"), raw, "Asia/Hong_Kong", 300L,
|
"neware:440:" + sequenceId);
|
}
|
|
private static Fixture fixture()
|
{
|
TTestMapper mapper = mock(TTestMapper.class);
|
ITTestService tests = mock(ITTestService.class);
|
AdaptiveAdapter adapter = new AdaptiveAdapter();
|
TTest test = receivingTest();
|
when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
|
when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
|
when(mapper.selectCollectorCode(9L)).thenReturn("collector-9");
|
when(tests.selectTestById(1L)).thenReturn(test);
|
TestClock clock = new TestClock(START);
|
return new Fixture(mapper, tests, adapter, test, clock,
|
new TestDataReceptionServiceImpl(mapper, tests, adapter, clock));
|
}
|
|
private static TTest receivingTest()
|
{
|
TTestChannelBinding binding = new TTestChannelBinding();
|
binding.setBindingId(2L);
|
binding.setChannelId(3L);
|
binding.setAdapterCode("neware76");
|
binding.setAdapterTimezoneId("Asia/Hong_Kong");
|
binding.setDevtype(21);
|
binding.setDevid(36);
|
binding.setSubdevid(1);
|
binding.setChlid(2);
|
TTestBattery battery = new TTestBattery();
|
battery.setTestBatteryId(4L);
|
battery.setCollectorId(9L);
|
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 record Fixture(TTestMapper mapper, ITTestService tests, AdaptiveAdapter adapter, TTest test,
|
TestClock clock, TestDataReceptionServiceImpl service) { }
|
|
/** 只暴露水位与逐行读取的测试替身，并记录被问到的水位与逐行读取次数。 */
|
private static final class AdaptiveAdapter implements TestDataObservationAdapter
|
{
|
private SampleWatermark watermark;
|
private List<CabinetRealtime> samples = new ArrayList<>();
|
private CabinetRealtime latest;
|
private int watermarkReads;
|
private final List<SampleWatermark> sampleReads = new ArrayList<>();
|
|
@Override public String adapterCode() { return "neware76"; }
|
@Override public AdapterCapabilities capabilities() { return AdapterCapabilities.observeOnly(); }
|
@Override public List<DetectedChannel> discoverChannels() { return List.of(); }
|
@Override public Map<ChannelIdentity, DeviceObservation> observe(Set<ChannelIdentity> channels)
|
{ return Map.of(); }
|
@Override public CabinetRealtime latestCabinet(ChannelIdentity channel) { return latest; }
|
@Override public CollectorRealtime latestCollector(String adapterCode, String collectorCode) { return null; }
|
@Override public List<StepRecord> stepRecords(ChannelIdentity channel) { return List.of(); }
|
@Override public List<CycleRecord> cycleRecords(ChannelIdentity channel) { return List.of(); }
|
@Override public long persistenceIntervalMillis(ChannelIdentity channel) { return 60000L; }
|
|
@Override public SampleWatermark currentWatermark(ChannelIdentity channel)
|
{
|
watermarkReads++;
|
return watermark;
|
}
|
|
@Override public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark since)
|
{
|
sampleReads.add(since);
|
return List.copyOf(samples);
|
}
|
}
|
|
/** 可推进的测试时钟：接收节奏与厂商写库节奏都按它推进。 */
|
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; }
|
}
|
}
|
