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("discharge", latest.stepType());
|
assertEquals("放电", latest.stepName());
|
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());
|
// 工步名称用同一套词（甲方软件那一列「状态」显示的就是充电/放电/搁置）：该来源的方案里
|
// 只有「步_N」标号，这个名称是判定出的方向，不是设备给的名称。
|
assertEquals("充电", step.stepName());
|
// 容量与能量是工步累计，取该段末行的值；工步时间是工步内相对时间（秒）。
|
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("搁置", steps.get(1).stepName());
|
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.stepName());
|
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);
|
}
|
}
|
