package com.battery.business.service.impl;
|
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertNull;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
import static org.mockito.ArgumentMatchers.any;
|
import static org.mockito.ArgumentMatchers.eq;
|
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.math.BigDecimal;
|
import java.time.Clock;
|
import java.time.Instant;
|
import java.time.ZoneOffset;
|
import java.util.Date;
|
import java.util.HexFormat;
|
import java.util.List;
|
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.extension.ExtendWith;
|
import org.mockito.ArgumentCaptor;
|
import org.springframework.boot.test.system.CapturedOutput;
|
import org.springframework.boot.test.system.OutputCaptureExtension;
|
import com.battery.device.api.TestDataObservationAdapter;
|
import com.battery.device.api.model.ChannelIdentity;
|
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
|
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
|
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
|
import com.battery.device.api.TestDataObservationAdapter.CollectorMeasurement;
|
import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
|
import com.battery.business.domain.TTest;
|
import com.battery.business.domain.TTestBattery;
|
import com.battery.business.domain.TTestChannelBinding;
|
import com.battery.business.domain.TTestCollectorChannel;
|
import com.battery.business.domain.TTestCycleRecord;
|
import com.battery.business.domain.TTestRealtimeRecord;
|
import com.battery.business.domain.TTestStepRecord;
|
import com.battery.business.mapper.TTestMapper;
|
import com.battery.business.service.ITTestService;
|
|
/** 通过接收入口验证三类持久化事实、实时预警和数据归属边界。 */
|
@ExtendWith(OutputCaptureExtension.class)
|
class TestDataReceptionServiceImplTest
|
{
|
private static final Instant CAPTURED_AT = Instant.parse("2026-09-02T04:01:00Z");
|
private static final Clock CLOCK = Clock.fixed(CAPTURED_AT, ZoneOffset.UTC);
|
|
@Test
|
void detectsRealtimeTemperatureExceededBetweenPersistenceSlots()
|
{
|
Fixture fixture = fixture();
|
var threshold = new com.battery.business.domain.TTestWarningThreshold();
|
threshold.setThresholdId(11L); threshold.setMetricCode("TEMPERATURE");
|
threshold.setMinValue(new BigDecimal("8")); threshold.setMaxValue(new BigDecimal("33"));
|
threshold.setUnit("℃"); threshold.setSeverity("WARNING");
|
fixture.tests.selectTestById(1L).setThresholds(List.of(threshold));
|
Clock clock = mock(Clock.class);
|
when(clock.instant()).thenReturn(CAPTURED_AT);
|
var service = new TestDataReceptionServiceImpl(fixture.mapper, fixture.tests, fixture.simulator, clock);
|
when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
|
List.of(collector("S-01", 10L, "temperature", "25.1", "℃"))));
|
when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
|
service.captureReceivedData();
|
|
when(clock.instant()).thenReturn(CAPTURED_AT.plusSeconds(5));
|
when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
|
List.of(collector("S-01", 11L, "temperature", "38", "℃"))));
|
service.captureReceivedData();
|
|
verify(fixture.mapper).upsertWarningException(eq(1L), eq(11L), eq("COLLECTOR"), any(),
|
eq("temperature"), eq(new BigDecimal("38")), eq("℃"), any(), eq("WARNING"), any(Date.class));
|
verify(fixture.mapper, times(1)).insertRealtimeRecord(any());
|
}
|
|
@Test
|
void primaryOvertemperatureIsNotOverwrittenByNormalSecondaryTemperature()
|
{
|
Fixture fixture = fixture();
|
var threshold = new com.battery.business.domain.TTestWarningThreshold();
|
threshold.setThresholdId(11L); threshold.setMetricCode("TEMPERATURE");
|
threshold.setMinValue(new BigDecimal("8")); threshold.setMaxValue(new BigDecimal("33"));
|
threshold.setUnit("℃"); threshold.setSeverity("WARNING");
|
var test = fixture.tests.selectTestById(1L);
|
test.setThresholds(List.of(threshold));
|
test.getBatteries().get(0).setCollectorChannels(List.of(
|
mapping("PRIMARY_TEMPERATURE", "S-01"), mapping("SECONDARY_TEMPERATURE_2", "S-03"),
|
mapping("SECONDARY_TEMPERATURE_3", "S-05")));
|
when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true, List.of(
|
collector("S-01", 10L, "temperature", "38", "℃"),
|
collector("S-03", 10L, "temperature", "25.3", "℃"),
|
collector("S-05", 10L, "temperature", "25.5", "℃"))));
|
when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
|
|
fixture.service().captureReceivedData();
|
|
verify(fixture.mapper).upsertWarningException(eq(1L), eq(11L), eq("COLLECTOR"), any(),
|
eq("temperature"), eq(new BigDecimal("38")), eq("℃"), any(), eq("WARNING"), any(Date.class));
|
var row = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(fixture.mapper).insertRealtimeRecord(row.capture());
|
assertEquals(new BigDecimal("38"), row.getValue().getTemperature());
|
assertTrue(row.getValue().getExtensionMeasurementsJson().contains("SECONDARY_TEMPERATURE_3"));
|
}
|
|
@Test
|
void recordsOfflineSourcesInBackgroundEvenWithoutSuccessfulSamples() {
|
Fixture fixture = fixture();
|
when(fixture.mapper.selectOfflineTestSources(1L)).thenReturn(List.of(
|
new com.battery.business.domain.vo.OfflineTestSource("COLLECTOR", "9")));
|
var service = fixture.service();
|
|
service.captureReceivedData();
|
service.captureReceivedData();
|
|
verify(fixture.mapper).upsertOfflineException(1L, "COLLECTOR", "9", Date.from(CAPTURED_AT));
|
verify(fixture.mapper, never()).insertRealtimeRecord(any());
|
when(fixture.mapper.selectTestByIdForUpdate(1L)).thenReturn(null);
|
service.captureReceivedData();
|
verify(fixture.mapper, times(1)).selectOfflineTestSources(1L);
|
}
|
|
@Test
|
void savesOneCombinedRealtimeRowForOneBatteryPollingSlot()
|
{
|
Fixture fixture = fixture();
|
when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
|
when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true, List.of(
|
collector("S-01", 10L, "TEMPERATURE", "25.100", "℃"),
|
collector("S-02", 10L, "POSITIVE_POTENTIAL", "1.620", "V"),
|
collector("S-03", 10L, "NEGATIVE_POTENTIAL", "1.610", "V"),
|
collector("S-04", 10L, "AMBIENT_TEMPERATURE", "23.400", "℃"))));
|
when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
|
|
fixture.service().captureReceivedData();
|
|
ArgumentCaptor<TTestRealtimeRecord> saved = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(fixture.mapper).insertRealtimeRecord(saved.capture());
|
TTestRealtimeRecord row = saved.getValue();
|
assertEquals("2026-09-02 12:00:40.000", row.getCollectorDeviceTimeRaw());
|
assertEquals("2026-09-02 12:00:30.000", row.getCabinetDeviceTimeRaw());
|
assertEquals(1L, row.getTestId());
|
assertEquals(4L, row.getTestBatteryId());
|
assertEquals(2L, row.getBindingId());
|
assertEquals("dev-sim:22:1:1:1", row.getSourceIdentity());
|
assertEquals("cabinet:7:collector:10", row.getSourceRecordId());
|
assertEquals(Date.from(Instant.parse("2026-09-02T04:00:30Z")), row.getCabinetDeviceTimeUtc());
|
assertEquals(3752L, com.alibaba.fastjson2.JSON.parseObject(row.getExtensionMeasurementsJson()).getLong("relativeTimeSeconds"));
|
assertEquals(Date.from(Instant.parse("2026-09-02T04:00:40Z")), row.getCollectorDeviceTimeUtc());
|
assertEquals(new BigDecimal("25.100"), row.getTemperature());
|
assertEquals(new BigDecimal("1.620"), row.getPositivePotential());
|
assertEquals(new BigDecimal("1.610"), row.getNegativePotential());
|
assertTrue(row.getExtensionMeasurementsJson().contains("AMBIENT_TEMPERATURE"));
|
assertTrue(!row.getExtensionMeasurementsJson().contains("POSITIVE_POTENTIAL"));
|
}
|
|
@Test
|
void savesCollectorFieldsWhenCabinetSourceFails()
|
{
|
Fixture fixture = fixture();
|
when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenThrow(new IllegalArgumentException("offline"));
|
when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
|
List.of(collector("S-01", 11L, "TEMPERATURE", "26.2", "℃"))));
|
when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
|
|
fixture.service().captureReceivedData();
|
|
ArgumentCaptor<TTestRealtimeRecord> saved = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(fixture.mapper).insertRealtimeRecord(saved.capture());
|
assertNull(saved.getValue().getCabinetSequenceId());
|
assertNull(com.alibaba.fastjson2.JSON.parseObject(saved.getValue().getExtensionMeasurementsJson()).get("relativeTimeSeconds"));
|
assertNull(saved.getValue().getCabinetDevtype());
|
assertNull(saved.getValue().getCabinetAdapterTimezoneId());
|
assertNull(saved.getValue().getVoltageV());
|
assertEquals(11L, saved.getValue().getCollectorSequenceId());
|
assertEquals("collector:11", saved.getValue().getSourceRecordId());
|
assertEquals(new BigDecimal("26.2"), saved.getValue().getTemperature());
|
assertNull(saved.getValue().getPositivePotential());
|
}
|
|
@Test
|
void savesOnlyDeviceReturnedStepAndCycleRecords()
|
{
|
Fixture fixture = fixture();
|
Instant started = Instant.parse("2026-09-02T04:00:10Z");
|
Instant ended = Instant.parse("2026-09-02T04:00:50Z");
|
when(fixture.simulator.stepRecords(any(ChannelIdentity.class))).thenReturn(List.of(new StepRecord(
|
"STEP-1", 1, 2, "恒流充电", "cc", started, ended, 3.4, 2.5, 8.4, ended.toString(), "UTC", ended,
|
java.util.Map.of("stepDuration", new BigDecimal("40.125"), "startVoltage", new BigDecimal("3.1"),
|
"chargeMedianVoltage", new BigDecimal("3.25"), "dischargeMedianVoltage", BigDecimal.ZERO))));
|
when(fixture.simulator.cycleRecords(any(ChannelIdentity.class))).thenReturn(List.of(new CycleRecord(
|
"CYCLE-1-1-1", 1, started, ended, 2.5, 2.4, 8.4, 8.0, 96, ended.toString(), "UTC", ended)));
|
when(fixture.mapper.insertStepRecord(any())).thenReturn(1);
|
when(fixture.mapper.insertCycleRecord(any())).thenReturn(1);
|
|
fixture.service().captureReceivedData();
|
|
ArgumentCaptor<TTestStepRecord> step = ArgumentCaptor.forClass(TTestStepRecord.class);
|
ArgumentCaptor<TTestCycleRecord> cycle = ArgumentCaptor.forClass(TTestCycleRecord.class);
|
verify(fixture.mapper).insertStepRecord(step.capture());
|
verify(fixture.mapper).insertCycleRecord(cycle.capture());
|
assertEquals("record:STEP-1", step.getValue().getDedupeKey());
|
var metrics = com.alibaba.fastjson2.JSON.parseObject(step.getValue().getExtensionMeasurementsJson());
|
assertEquals(new BigDecimal("40.125"), metrics.getBigDecimal("stepDuration"));
|
assertEquals(new BigDecimal("3.1"), metrics.getBigDecimal("startVoltage"));
|
assertEquals(new BigDecimal("3.25"), metrics.getBigDecimal("chargeMedianVoltage"));
|
assertEquals(BigDecimal.ZERO, metrics.getBigDecimal("dischargeMedianVoltage"));
|
assertEquals("CYCLE-1-1-1", cycle.getValue().getSourceRecordId());
|
assertEquals("record:CYCLE-1-1-1", cycle.getValue().getDedupeKey());
|
assertEquals(Date.from(ended), step.getValue().getServerReceivedAtUtc());
|
assertEquals(Date.from(ended), cycle.getValue().getServerReceivedAtUtc());
|
}
|
|
@Test
|
void ignoresSameRealtimeContentOnRepeatedReception()
|
{
|
Fixture fixture = fixture();
|
when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
|
when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(false, List.of()));
|
when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
|
|
fixture.service().captureReceivedData();
|
ArgumentCaptor<TTestRealtimeRecord> saved = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
|
verify(fixture.mapper).insertRealtimeRecord(saved.capture());
|
when(fixture.mapper.selectRealtimeRecordContentHash(eq(1L), eq(4L), any(Date.class)))
|
.thenReturn(HexFormat.of().formatHex(saved.getValue().getContentSha256()));
|
|
fixture.service().captureReceivedData();
|
|
verify(fixture.mapper, times(1)).insertRealtimeRecord(any());
|
}
|
|
@Test
|
void rejectsConflictingRealtimeContentAndWritesApplicationLog(CapturedOutput output)
|
{
|
Fixture fixture = fixture();
|
when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
|
when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(false, List.of()));
|
when(fixture.mapper.selectRealtimeRecordContentHash(eq(1L), eq(4L), any(Date.class)))
|
.thenReturn("different-content");
|
|
fixture.service().captureReceivedData();
|
|
verify(fixture.mapper, never()).insertRealtimeRecord(any());
|
assertTrue(output.getOut().contains("实时采样内容冲突"));
|
}
|
|
@Test void resumedReceptionExcludesAggregatesObservedDuringPause()
|
{
|
Fixture fixture = fixture();
|
fixture.tests.selectTestById(1L).setReceptionBoundaryAtUtc(Date.from(CAPTURED_AT));
|
Instant pausedAt = CAPTURED_AT.minusSeconds(10);
|
when(fixture.simulator.stepRecords(any())).thenReturn(List.of(new StepRecord("paused", 1, 1, "工步", "cc",
|
pausedAt.minusSeconds(20), pausedAt, 3, 1, 2, pausedAt.toString(), "UTC", pausedAt, java.util.Map.of())));
|
when(fixture.simulator.cycleRecords(any())).thenReturn(List.of(new CycleRecord("paused", 1,
|
pausedAt.minusSeconds(20), pausedAt, 1, 1, 2, 2, 100, pausedAt.toString(), "UTC", pausedAt)));
|
fixture.service().captureReceivedData();
|
verify(fixture.mapper, never()).insertStepRecord(any());
|
verify(fixture.mapper, never()).insertCycleRecord(any());
|
}
|
|
@Test void deletedAggregateDoesNotReturnOnRepeatedObservation()
|
{
|
Fixture fixture = fixture();
|
when(fixture.simulator.stepRecords(any())).thenReturn(List.of(new StepRecord("deleted", 1, 1, "工步", "cc",
|
CAPTURED_AT.minusSeconds(10), CAPTURED_AT, 3, 1, 2, CAPTURED_AT.toString(), "UTC", CAPTURED_AT, java.util.Map.of())));
|
when(fixture.mapper.wasAggregateDeleted(1L, "STEP", 4L, "dev-sim:22:1:1:1", "record:deleted")).thenReturn(true);
|
fixture.service().captureReceivedData();
|
verify(fixture.mapper, never()).insertStepRecord(any());
|
}
|
|
@Test void deletedRealtimeSlotDoesNotReturnAfterServiceRestart()
|
{
|
Fixture fixture = fixture();
|
when(fixture.simulator.latestCabinet(any())).thenReturn(cabinetRecord());
|
when(fixture.mapper.wasRealtimeSlotDeleted(1L, 4L, CAPTURED_AT.toEpochMilli())).thenReturn(true);
|
fixture.service().captureReceivedData();
|
fixture.service().captureReceivedData();
|
verify(fixture.mapper, never()).insertRealtimeRecord(any());
|
}
|
|
private static Fixture fixture()
|
{
|
TTestMapper mapper = mock(TTestMapper.class);
|
ITTestService tests = mock(ITTestService.class);
|
TestDataObservationAdapter simulator = mock(TestDataObservationAdapter.class);
|
TTest test = receivingTest();
|
when(mapper.selectCollectorCode(9L)).thenReturn("collector-9");
|
when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
|
when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
|
when(tests.selectTestById(1L)).thenReturn(test);
|
when(simulator.persistenceIntervalMillis(any(ChannelIdentity.class))).thenReturn(60000L);
|
return new Fixture(mapper, tests, simulator);
|
}
|
|
private static TTest receivingTest()
|
{
|
TTestChannelBinding binding = new TTestChannelBinding();
|
binding.setBindingId(2L);
|
binding.setChannelId(3L);
|
binding.setAdapterCode("dev-sim");
|
binding.setAdapterTimezoneId("Asia/Hong_Kong");
|
binding.setDevtype(22);
|
binding.setDevid(1);
|
binding.setSubdevid(1);
|
binding.setChlid(1);
|
TTestBattery battery = new TTestBattery();
|
battery.setTestBatteryId(4L);
|
battery.setCollectorId(9L);
|
battery.setBinding(binding);
|
battery.setCollectorChannels(List.of(
|
mapping("TEMPERATURE", "S-01"), mapping("POSITIVE_POTENTIAL", "S-02"),
|
mapping("NEGATIVE_POTENTIAL", "S-03"), mapping("AMBIENT_TEMPERATURE", "S-04")));
|
TTest test = new TTest();
|
test.setTestId(1L);
|
test.setTestStatus("RUNNING");
|
test.setReceptionStatus("RECEIVING");
|
test.setStartedAtUtc(Date.from(Instant.parse("2026-09-02T04:00:00Z")));
|
test.setReceptionBoundaryAtUtc(Date.from(Instant.parse("2026-09-02T04:00:00Z")));
|
test.setBatteries(List.of(battery));
|
return test;
|
}
|
|
private static TTestCollectorChannel mapping(String metric, String channel)
|
{
|
TTestCollectorChannel mapping = new TTestCollectorChannel();
|
mapping.setMeasurementCode(metric);
|
mapping.setChannelCode(channel);
|
return mapping;
|
}
|
|
private static CabinetRealtime cabinetRecord()
|
{
|
Instant deviceTime = Instant.parse("2026-09-02T04:00:30Z");
|
return new CabinetRealtime(7L, deviceTime, 1, 3.2, 2.5, 8.4,
|
1, 2, "恒流充电", "cc", Instant.parse("2026-09-02T04:00:35Z"), "2026-09-02 12:00:30.000", "Asia/Hong_Kong", 3752L);
|
}
|
|
private static CollectorMeasurement collector(String channel, long sequenceId, String metric,
|
String value, String unit)
|
{
|
return new CollectorMeasurement(channel, sequenceId,
|
Instant.parse("2026-09-02T04:00:40Z"), "2026-09-02 12:00:40.000",
|
Instant.parse("2026-09-02T04:00:50Z"), metric, new BigDecimal(value), unit);
|
}
|
|
private static CollectorRealtime collectorSnapshot(boolean online,
|
List<CollectorMeasurement> measurements)
|
{
|
return new CollectorRealtime("collector-9", online, 1000, measurements, "Asia/Hong_Kong");
|
}
|
|
private record Fixture(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter simulator)
|
{
|
TestDataReceptionServiceImpl service()
|
{
|
return new TestDataReceptionServiceImpl(mapper, tests, simulator, CLOCK);
|
}
|
}
|
}
|
