对接第 6 台新威上位机（7.6 客户端，走库路线），并修两处对接中暴露的问题

yeg

2 天以前 1de35281a699b8d2ca90777f63c7e032c3c9dc3d

 ruoyi-admin/src/test/java/com/battery/business/service/impl/TestDataReceptionServiceImplTest.java

@@ -40,7 +40,7 @@
import com.battery.business.mapper.TTestMapper;
import com.battery.business.service.ITTestService;

/** Ticket08 通过接收入口验证三类持久化事实。 */
/** 通过接收入口验证三类持久化事实、实时预警和数据归属边界。 */
@ExtendWith(OutputCaptureExtension.class)
class TestDataReceptionServiceImplTest
{
@@ -48,11 +48,84 @@
    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("collector-9")).thenReturn(collectorSnapshot(true, List.of(
        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"),
@@ -72,6 +145,7 @@
        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());
@@ -85,7 +159,7 @@
    {
        Fixture fixture = fixture();
        when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenThrow(new IllegalArgumentException("offline"));
        when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true,
        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);

@@ -94,6 +168,7 @@
        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());
@@ -110,9 +185,12 @@
        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())));
                "STEP-1", 1, 2, "恒流充电", "cc", started, ended, 3.4, 2.6, 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)));
                "CYCLE-1-1-1", 1, started, ended, 2.5, 2.4, 8.4, 8.0, 96d, ended.toString(), "UTC", ended,
                CycleRecord.SOURCE_DEVICE)));
        when(fixture.mapper.insertStepRecord(any())).thenReturn(1);
        when(fixture.mapper.insertCycleRecord(any())).thenReturn(1);

@@ -123,10 +201,36 @@
        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(CycleRecord.SOURCE_DEVICE, cycle.getValue().getSourceKind());
        // 工步结束时的设备读数：电压与电流一起落库。
        assertEquals(new BigDecimal("3.4"), step.getValue().getVoltageV());
        assertEquals(new BigDecimal("2.6"), step.getValue().getCurrentA());
        assertEquals(Date.from(ended), step.getValue().getServerReceivedAtUtc());
        assertEquals(Date.from(ended), cycle.getValue().getServerReceivedAtUtc());
    }

    /** 接收调度每秒运行，但来源按自己的写库节奏产出工步：不按来源节奏限流会每秒重查厂商库。 */
    @Test void readsTheStepLayerOncePerSourceCadenceInsteadOfEverySchedulerTick()
    {
        Fixture fixture = fixture();
        when(fixture.simulator.stepRecords(any(ChannelIdentity.class))).thenReturn(List.of());
        when(fixture.simulator.cycleRecords(any(ChannelIdentity.class))).thenReturn(List.of());

        TestDataReceptionServiceImpl service = fixture.service();
        service.captureReceivedData();
        service.captureReceivedData();
        service.captureReceivedData();

        verify(fixture.simulator, times(1)).stepRecords(any(ChannelIdentity.class));
        verify(fixture.simulator, times(1)).cycleRecords(any(ChannelIdentity.class));
    }

    @Test
@@ -134,7 +238,7 @@
    {
        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.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();
@@ -153,7 +257,7 @@
    {
        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.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");

@@ -169,19 +273,54 @@
        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())));
                pausedAt.minusSeconds(20), pausedAt, 3d, 1.5d, 1d, 2d, 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)));
                pausedAt.minusSeconds(20), pausedAt, 1, 1, 2, 2, 100d, pausedAt.toString(), "UTC", pausedAt,
                CycleRecord.SOURCE_DEVICE)));
        fixture.service().captureReceivedData();
        verify(fixture.mapper, never()).insertStepRecord(any());
        verify(fixture.mapper, never()).insertCycleRecord(any());
    }

    /** 循环统计的来源类别随记录落库：聚合值必须能与设备或文件提供的值区分。 */
    @Test void persistsCycleSourceKindAndKeepsMissingMetricsEmpty()
    {
        Fixture fixture = fixture();
        when(fixture.simulator.cycleRecords(any(ChannelIdentity.class))).thenReturn(List.of(
                new CycleRecord("neware:442:cycle:3", 3, CAPTURED_AT.minusSeconds(600), CAPTURED_AT, 2.375, 1.383,
                        4.059, 1.715, null, CAPTURED_AT.toString(), "Asia/Hong_Kong", CAPTURED_AT,
                        CycleRecord.SOURCE_AGGREGATED)));
        when(fixture.simulator.stepRecords(any(ChannelIdentity.class))).thenReturn(List.of(new StepRecord(
                "neware:442:step:3:7", 3, 7, null, "7", CAPTURED_AT.minusSeconds(60), CAPTURED_AT, 1.3, 0.4, null, null,
                CAPTURED_AT.toString(), "Asia/Hong_Kong", CAPTURED_AT,
                java.util.Map.of("chargeCapacity", new BigDecimal("13.888888888889")))));
        when(fixture.mapper.insertCycleRecord(any())).thenReturn(1);
        when(fixture.mapper.insertStepRecord(any())).thenReturn(1);

        fixture.service().captureReceivedData();

        ArgumentCaptor<TTestCycleRecord> cycle = ArgumentCaptor.forClass(TTestCycleRecord.class);
        verify(fixture.mapper).insertCycleRecord(cycle.capture());
        assertEquals(CycleRecord.SOURCE_AGGREGATED, cycle.getValue().getSourceKind());
        // 没有可用充电容量就没有效率：空值不是零。
        assertNull(cycle.getValue().getEfficiency());
        assertEquals(new BigDecimal("2.375"), cycle.getValue().getChargeCapacityAh());

        ArgumentCaptor<TTestStepRecord> step = ArgumentCaptor.forClass(TTestStepRecord.class);
        verify(fixture.mapper).insertStepRecord(step.capture());
        // 方向未确认的工步没有单项容量与能量：落库保持空值，四组明细仍在扩展里。
        assertNull(step.getValue().getCapacityAh());
        assertNull(step.getValue().getEnergyWh());
        assertTrue(step.getValue().getExtensionMeasurementsJson().contains("chargeCapacity"));
    }

    @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())));
                CAPTURED_AT.minusSeconds(10), CAPTURED_AT, 3d, 1.5d, 1d, 2d, 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());
@@ -251,7 +390,7 @@
    {
        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");
                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,