package com.battery.business.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.jupiter.api.Test;
import com.battery.device.api.DeviceAdapterException;
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
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.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.mapper.TDeviceAdapterMapper;
import org.mockito.Mockito;

/**
 * 推式来源适配器的观测契约：落地行投影成统一观测（身份、时间、单位、幂等身份），
 * 发现候选带最近数据时间且心跳缺失即明确失败，水位按 (设备时间, 协议序号) 取严格更晚的行。
 * 原始行由接收端点解析，所以这里只喂落地行，适配器里不应出现任何厂商列名。
 */
class PushIngestAdapterContractTest
{
    private static final String ADAPTER = "j22";
    private static final String TOKEN = "j22-token";
    /** 前两行同一个绝对秒（08:03:20 UTC）但协议序号不同；第三行晚一分钟。 */
    private static final String BATCH = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"record_time":"134349692000000000","Data_Point":420,"Step_Time":540.0,"Step_Index":4,
               "Cycle_Index":1,"Voltage":1.26778,"eCurrent":0.5,"ChargeCapacity":0.12,"ChargeEnergy":0.3},
              {"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":5,
               "Cycle_Index":1,"Voltage":1.2702,"eCurrent":-0.4,"DischargeCapacity":0.05,"DischargeEnergy":0.1}]}
            """;
    /**
     * 两个循环的采样行：循环 1 是充电、搁置、放电三段（都结束了），循环 2 的第一段还在进行。
     * 容量列是工步累计：充电段从 0.10 涨到 0.12，放电段从 0.05 涨到 0.09。
     */
    private static final String TWO_CYCLES = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"record_time":"134349692000000000","Data_Point":420,"Step_Time":540.0,"Step_Index":1,"Cycle_Index":1,
               "Voltage":1.20,"eCurrent":1.0,"ChargeCapacity":0.10,"ChargeEnergy":0.30},
              {"record_time":"134349698000000000","Data_Point":421,"Step_Time":600.0,"Step_Index":1,"Cycle_Index":1,
               "Voltage":1.25,"eCurrent":1.0,"ChargeCapacity":0.12,"ChargeEnergy":0.36},
              {"record_time":"134349704000000000","Data_Point":422,"Step_Time":1200.0,"Step_Index":2,"Cycle_Index":1,
               "Voltage":1.25,"eCurrent":0.0,"ChargeCapacity":0.12,"DischargeCapacity":0.0},
              {"record_time":"134349764000000000","Data_Point":423,"Step_Time":1800.0,"Step_Index":3,"Cycle_Index":1,
               "Voltage":1.24,"eCurrent":-1.0,"ChargeCapacity":0.12,"DischargeCapacity":0.05,"DischargeEnergy":0.15},
              {"record_time":"134349824000000000","Data_Point":424,"Step_Time":3600.0,"Step_Index":3,"Cycle_Index":1,
               "Voltage":1.20,"eCurrent":-1.0,"ChargeCapacity":0.12,"DischargeCapacity":0.09,"DischargeEnergy":0.27},
              {"record_time":"134349830000000000","Data_Point":425,"Step_Time":60.0,"Step_Index":1,"Cycle_Index":2,
               "Voltage":1.18,"eCurrent":2.0,"ChargeCapacity":0.01,"ChargeEnergy":0.02}]}
            """;
    /** 第一段的电流正负混用：方向判不出来，明细照留但不进循环的任何一侧。 */
    private static final String MIXED_SIGN = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"record_time":"134349692000000000","Data_Point":420,"Step_Index":1,"Cycle_Index":1,
               "Voltage":1.20,"eCurrent":1.0,"ChargeCapacity":0.10},
              {"record_time":"134349698000000000","Data_Point":421,"Step_Index":1,"Cycle_Index":1,
               "Voltage":1.20,"eCurrent":-1.0,"ChargeCapacity":0.10,"DischargeCapacity":0.02},
              {"record_time":"134349704000000000","Data_Point":422,"Step_Index":2,"Cycle_Index":1,
               "Voltage":1.25,"eCurrent":1.0,"ChargeCapacity":0.12,"ChargeEnergy":0.36},
              {"record_time":"134349830000000000","Data_Point":425,"Step_Index":1,"Cycle_Index":2,
               "Voltage":1.18,"eCurrent":2.0,"ChargeCapacity":0.01}]}
            """;

    private final FakeIngestLanding landing = new FakeIngestLanding();
    private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
    private static final Instant NOW = Instant.parse("2026-09-27T09:00:00Z");
    /**
     * 环境量的确认结论落在这里：厂商列名只在配置里出现一次，代码里没有任何厂商列名。
     * 22 所的列语义与量纲仍未现场确认，所以生产配置里默认没有这一项。
     */
    private static final Map<String, String> AUXILIARY_METRICS = Map.of("auxiliaryMetrics",
            "[{\"vendorChannel\":26,\"channelCode\":\"S-01\",\"column\":\"Tempbox\","
                    + "\"metricCode\":\"TEMPERATURE\",\"unit\":\"℃\"}]");

    @Test
    void topologyPushMakesChannelsCandidatesWithTheirOwnLatestDataTime()
    {
        // 拓扑行里可以带 channelNo：来源自己软件里显示的通道号（现场核对：22 所界面号 = 表名号 + 1）
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26,\"channelNo\":27},{\"channel\":18}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        List<DetectedChannel> channels = adapter.discoverChannels();

        assertEquals(List.of(18, 26), channels.stream().map(channel -> channel.identity().chlid()).toList());
        DetectedChannel active = channels.stream().filter(channel -> channel.identity().chlid() == 26).findFirst()
                .orElseThrow();
        // 最近数据时间来自该通道自己的采样行，不是拓扑上报时间。
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), active.lastDataAt());
        assertTrue(active.available());
        // 候选的机器地址取实例配置：登记时会被带进检测柜的 IP 字段。
        assertEquals("127.0.0.1", active.serverIp());
        // 来源自己的通道叫法随候选一起交出（界面与登记用它显示），身份仍是表名里的号。
        assertEquals("27", active.sourceDisplayNo());
        assertEquals(26, active.identity().chlid());
        // 登记范围由操作员决定：没有采样行的通道照样出现在候选里，只是没有最近数据时间。
        assertNull(channels.stream().filter(channel -> channel.identity().chlid() == 18).findFirst()
                .orElseThrow().lastDataAt());
        // 来源没报叫法的通道保持为空，界面回退到「单元-通道」。
        assertNull(channels.stream().filter(channel -> channel.identity().chlid() == 18).findFirst()
                .orElseThrow().sourceDisplayNo());
    }

    /** 心跳缺失或过期即来源不可达：发现路径明确失败，业务侧据此把已登记设备置为离线并撤下旧快照。 */
    @Test
    void missingOrStaleHeartbeatFailsDiscoveryInsteadOfReturningStaleValues()
    {
        // 还没收到过任何推送：这是"取数程序没起来"这一条可见的失败，不是沉默的空清单。
        DeviceAdapterException missing = assertThrows(DeviceAdapterException.class, () -> adapter(NOW).discoverChannels());
        assertEquals("SOURCE_OFFLINE", missing.failureCode());

        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        // 收到推送即刷新心跳；容忍窗口是三个轮询周期，一轮没推不算离线。
        assertEquals(1, adapter(NOW.plus(Duration.ofSeconds(179))).discoverChannels().size());
        // 连丢三轮才算离线，且不必等候选过期。
        DeviceAdapterException stale = assertThrows(DeviceAdapterException.class,
                () -> adapter(NOW.plus(Duration.ofSeconds(181))).discoverChannels());
        assertEquals("SOURCE_OFFLINE", stale.failureCode());
    }

    @Test
    void samplesAreProjectedWithRawTimeUnscaledUnitsAndSourceRowIdentity()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        CabinetRealtime latest = adapter.latestCabinet(channel());

        // 时间按厂商格式换算为 UTC，同时保留原始值与来源时区；单位是物理单位，不做缩放。
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), latest.deviceTime());
        assertEquals("134349698000000000", latest.deviceTimeRaw());
        assertEquals("Asia/Shanghai", latest.timezoneId());
        assertEquals(1.2702, latest.voltageV(), 1e-9);
        assertEquals(-0.4, latest.currentA(), 1e-9);
        assertEquals(600L, latest.relativeTimeSeconds());
        assertEquals(1, latest.cycleNo());
        assertEquals(5, latest.stepNo());
        // 放电容量的取法按电流方向，与直连来源同一口径；来源行身份供幂等与冲突判定。
        assertEquals(0.05, latest.capacityAh(), 1e-9);
        assertEquals("push:DB_20260927_155209/Ws26_IV_0:422", latest.sourceRecordId());

        // 实时快照：通道工作状态在读落地表路线下不可观测（ADR-0063）。
        Map<ChannelIdentity, RealtimeSnapshot> snapshots = adapter.inquire(Set.of(channel()));
        assertEquals(DeviceWorkStatus.UNOBSERVABLE, snapshots.get(channel()).workStatus());
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), snapshots.get(channel()).deviceTime());
    }

    @Test
    void watermarkAndSamplesSinceKeepSameSecondRowsDistinctAndSkipNothingBeforeTheBoundary()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        // 水位是落地表里该通道最新一行：设备绝对时间与协议序号。
        assertEquals(new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 422L),
                adapter.currentWatermark(channel()));

        // 只取严格晚于水位的行：同一绝对秒内的多行靠协议序号各自成立。
        List<CabinetRealtime> rows = adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T07:53:20Z"), 420L));
        assertEquals(List.of(421L, 422L), rows.stream().map(CabinetRealtime::sequenceId).toList());
        assertEquals(0, adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 422L)).size());
        // 同一绝对秒内序号更大的行仍然成立：水位是二元组而不是秒。
        assertEquals(List.of(422L), adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 421L)).stream()
                .map(CabinetRealtime::sequenceId).toList());
        // 调用方尚未取得水位时不把来源里的历史并入本次接收。
        assertTrue(adapter.samplesSince(channel(), null).isEmpty());
        // 水位为空的二元组表示边界时刻来源里什么也没有：之后出现的行全部属于本次接收。
        assertEquals(3, adapter.samplesSince(channel(), new SampleWatermark(null, 0L)).size());
    }

    @Test
    void emptySourceHasNoWatermarkAndNoRealtimeObservation()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        PushIngestAdapter adapter = adapter(NOW);

        assertEquals(new SampleWatermark(null, 0L), adapter.currentWatermark(channel()));
        assertNull(adapter.latestCabinet(channel()));
        assertTrue(adapter.inquire(Set.of(channel())).isEmpty());
    }

    /** 未声明辅助测量列就不交出测量值：确认之前不按列名推断语义，也不写零占位。 */
    @Test
    void collectorObservationsStayEmptyWithoutDeclaredAuxiliaryMetrics()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        assertNull(adapter.latestCollector(ADAPTER, "collector-1"));
        // 落地表在本地，取数不产生厂商负载，所以按展示节奏取数；厂商写库节奏仍是 60 秒。
        assertEquals(5000L, adapter.persistenceIntervalMillis(channel()));
        // 工步与循环的重算跟着来源的写库节奏走，不跟实时行的展示节奏：源头最快 60 秒才产出一行。
        assertEquals(60000L, adapter.aggregateIntervalMillis(channel()));
        AdapterCapabilities capabilities = adapter.capabilities();
        assertTrue(capabilities.discovery() && capabilities.realtimeObservation() && !capabilities.control());
    }

    /**
     * 来源没有工步层，工步与循环由采样行分段推导（ADR-0067）：已结束的段各交出一条工步记录，
     * 取值来自该段末行，方向由该段自己的电流符号判定；最新一段仍在进行，不交出去。
     */
    @Test
    void closedSampleSegmentsBecomeStepRecordsWhileTheNewestSegmentStaysUnpublished()
    {
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        List<StepRecord> steps = adapter.stepRecords(channel());

        // 三行落地行分成两段：(循环 1, 工步 4) 已结束，(循环 1, 工步 5) 还在进行。
        assertEquals(1, steps.size());
        StepRecord step = steps.get(0);
        assertEquals(1, step.cycleNo());
        assertEquals(4, step.stepNo());
        // 方向取自该段电流符号：段内两行恒正即充电。
        assertEquals("charge", step.stepType());
        // 容量与能量是工步累计，取该段末行的值；工步时间是工步内相对时间（秒）。
        assertEquals(0.121, step.capacityAh(), 1e-9);
        assertEquals(0.31, step.energyWh(), 1e-9);
        assertEquals(600.0, step.measurements().get("stepDuration").doubleValue(), 1e-9);
        assertEquals(1.26778, step.measurements().get("startVoltage").doubleValue(), 1e-9);
        // 设备没给的列不写零占位：该段只有充电侧。
        assertFalse(step.measurements().containsKey("dischargeCapacity"));
        assertEquals(1.2701, step.endVoltageV(), 1e-9);
        assertEquals(0.5, step.endCurrentA(), 1e-9);
        assertEquals(Instant.parse("2026-09-27T07:53:20Z"), step.startedAt());
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), step.endedAt());
        // 接收时间取我们收到该段末行的时刻（与推式实时路径同一口径），设备原始时间照留。
        assertEquals(NOW, step.serverReceivedAt());
        assertEquals("134349698000000000", step.deviceTimeRaw());
        // 来源行身份含来源库与段标识：同一次测试重算多少次都得到同一个身份。
        assertEquals("push:DB_20260927_155209/Ws26_IV_0:step:1:4", step.sourceRecordId());
        // 进行中的那一段所属的循环整体不交：半个循环不是事实。
        assertTrue(adapter.cycleRecords(channel()).isEmpty());
    }

    /**
     * 循环由已结束的工步按方向求和（搁置段两向都不增长，贡献 0），效率是放电容量与充电容量之比；
     * 推导值带来源类别，界面与导出据此标注它是聚合出来的。
     */
    @Test
    void cyclesAggregateClosedStepsByDirectionAndSkipTheRunningCycle()
    {
        ingest(TWO_CYCLES);
        PushIngestAdapter adapter = adapter(NOW);

        List<StepRecord> steps = adapter.stepRecords(channel());
        List<CycleRecord> cycles = adapter.cycleRecords(channel());

        // 循环 1 的三段都已结束（后面已经出现循环 2 的第一段）；循环 2 的第一段仍在进行。
        assertEquals(List.of("1:1:charge", "1:2:rest", "1:3:discharge"), steps.stream()
                .map(step -> step.cycleNo() + ":" + step.stepNo() + ":" + step.stepType()).toList());
        // 搁置段自身的累计是 0（该段什么都没充放），不是缺失。
        assertEquals(0.0, steps.get(1).capacityAh(), 1e-9);
        assertEquals(0.0, steps.get(1).energyWh(), 1e-9);
        assertEquals(List.of(1), cycles.stream().map(CycleRecord::cycleNo).toList());

        CycleRecord cycle = cycles.get(0);
        assertEquals(0.12, cycle.chargeCapacityAh(), 1e-9);
        assertEquals(0.09, cycle.dischargeCapacityAh(), 1e-9);
        assertEquals(0.36, cycle.chargeEnergyWh(), 1e-9);
        assertEquals(0.27, cycle.dischargeEnergyWh(), 1e-9);
        // 效率是百分数：0.09 / 0.12。
        assertEquals(75.0, cycle.efficiency(), 1e-9);
        assertEquals(CycleRecord.SOURCE_AGGREGATED, cycle.sourceKind());
        assertEquals("push:DB_20260927_155209/Ws26_IV_0:cycle:1", cycle.sourceRecordId());
    }

    /** 方向判不出来时（段内电流正负混用）明细照留、单项容量留空，也不进循环的任何一侧。 */
    @Test
    void segmentsWithMixedCurrentSignsStayOutOfTheCycleAggregate()
    {
        ingest(MIXED_SIGN);
        PushIngestAdapter adapter = adapter(NOW);

        List<StepRecord> steps = adapter.stepRecords(channel());
        List<CycleRecord> cycles = adapter.cycleRecords(channel());

        StepRecord undecided = steps.get(0);
        assertEquals(1, undecided.cycleNo());
        assertEquals(1, undecided.stepNo());
        // 方向不唯一时不猜：类型留空、单项容量与能量留空。
        assertNull(undecided.stepType());
        assertNull(undecided.capacityAh());
        assertNull(undecided.energyWh());
        // 四组容量能量作为设备事实照原样保留，供核对。
        assertEquals(0.02, undecided.measurements().get("dischargeCapacity").doubleValue(), 1e-9);

        // 该段不进任何一侧：循环 1 的放电侧算出来是 0，而不是把这一段里的 0.02 算进去。
        assertEquals(1, cycles.size());
        assertEquals(0.12, cycles.get(0).chargeCapacityAh(), 1e-9);
        assertEquals(0.0, cycles.get(0).dischargeCapacityAh(), 1e-9);
    }

    /** 环境量与量纲确认后落到配置：列名、指标编码、单位、采集器通道都由声明给出，代码里没有厂商列名。 */
    @Test
    void declaredAuxiliaryMetricIsProjectedWithItsOwnSourceIdentityValueUnitAndDeviceTime()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest("""
                {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
                  {"record_time":"134349692000000000","Data_Point":420,"Voltage":1.26778,"eCurrent":0.5,"Tempbox":26.5}]}
                """);
        PushIngestAdapter adapter = adapter(NOW, AUXILIARY_METRICS);

        var collector = adapter.latestCollector(ADAPTER, "collector-1");

        assertTrue(collector.online());
        var measurement = collector.measurements().get(0);
        assertEquals("S-01", measurement.channelCode());
        assertEquals("TEMPERATURE", measurement.metricCode());
        assertEquals(0, new java.math.BigDecimal("26.5").compareTo(measurement.value()));
        assertEquals("℃", measurement.unit());
        // 设备时间与来源身份来自这一行自己：不与检测柜记录共用时间，也不伪造配对。
        assertEquals(Instant.parse("2026-09-27T07:53:20Z"), measurement.deviceTime());
        assertEquals("134349692000000000", measurement.deviceTimeRaw());
    }

    /** 落地行里没有这一列（设备没给）就不写这一项，也不写零占位。 */
    @Test
    void declaredAuxiliaryMetricAbsentFromTheRowIsNotFabricated()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW, AUXILIARY_METRICS);

        var collector = adapter.latestCollector(ADAPTER, "collector-1");

        assertTrue(collector.online());
        assertTrue(collector.measurements().isEmpty());
    }

    /** 来源离线（心跳超时）时不交出任何测量值：离线不伪造测量值，业务侧据此也不落采集器行。 */
    @Test
    void staleHeartbeatReportsCollectorSourceOfflineWithoutMeasurements()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest("""
                {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
                  {"record_time":"134349692000000000","Data_Point":420,"Voltage":1.26778,"eCurrent":0.5,"Tempbox":26.5}]}
                """);
        PushIngestAdapter adapter = adapter(NOW.plus(Duration.ofMinutes(10)), AUXILIARY_METRICS);

        var collector = adapter.latestCollector(ADAPTER, "collector-1");

        assertFalse(collector.online());
        assertTrue(collector.measurements().isEmpty());
    }

    @Test
    void channelIdentityFromAnotherInstanceFailsInsteadOfFallingBack()
    {
        PushIngestAdapter adapter = adapter(NOW);

        DeviceAdapterException failure = assertThrows(DeviceAdapterException.class,
                () -> adapter.currentWatermark(new ChannelIdentity("neware76", 21, 36, 1, 2)));
        assertTrue(failure.getMessage().contains("neware76"));
        assertFalse(adapter.adapterCode().isEmpty());
    }

    private ChannelIdentity channel()
    {
        return new ChannelIdentity(ADAPTER, 0, 0, 1, 26);
    }

    private PushIngestAdapter adapter(Instant now)
    {
        return adapter(now, Map.of("tokenEnv", "BATTERY_INGEST_TOKEN"));
    }

    private PushIngestAdapter adapter(Instant now, Map<String, String> sourceOptions)
    {
        return new PushIngestAdapter(config(sourceOptions), landing, Clock.fixed(now, ZoneOffset.UTC));
    }

    private static AdapterInstanceConfig config(Map<String, String> sourceOptions)
    {
        return new AdapterInstanceConfig(ADAPTER, PushIngestAdapter.ADAPTER_TYPE, "127.0.0.1", 8080, 60000,
                "Asia/Shanghai", true, AdapterCapabilities.observeOnly(), sourceOptions);
    }

    private void ingest(String body)
    {
        TDeviceAdapter source = new TDeviceAdapter();
        source.setAdapterCode(ADAPTER);
        source.setAdapterType(PushIngestAdapter.ADAPTER_TYPE);
        source.setStatus("0");
        source.setTimezoneId("Asia/Shanghai");
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(source);
        new DeviceIngestServiceImpl(adapters, landing, Clock.fixed(NOW, ZoneOffset.UTC), key -> TOKEN)
                .ingest(ADAPTER, TOKEN, body);
    }
}