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; /** Ticket08 通过接收入口验证三类持久化事实。 */ @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("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("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("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("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 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(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("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 saved = ArgumentCaptor.forClass(TTestRealtimeRecord.class); verify(fixture.mapper).insertRealtimeRecord(saved.capture()); assertNull(saved.getValue().getCabinetSequenceId()); 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()))); 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 step = ArgumentCaptor.forClass(TTestStepRecord.class); ArgumentCaptor cycle = ArgumentCaptor.forClass(TTestCycleRecord.class); verify(fixture.mapper).insertStepRecord(step.capture()); verify(fixture.mapper).insertCycleRecord(cycle.capture()); assertEquals("record:STEP-1", step.getValue().getDedupeKey()); 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("collector-9")).thenReturn(collectorSnapshot(false, List.of())); when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1); fixture.service().captureReceivedData(); ArgumentCaptor 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("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"); } 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 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); } } }