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.SampleWatermark;
|
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}]}
|
""";
|
|
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 collectorAndStepCycleObservationsStayEmptyWithoutDeclaredAuxiliaryMetrics()
|
{
|
ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
|
ingest(BATCH);
|
PushIngestAdapter adapter = adapter(NOW);
|
|
assertNull(adapter.latestCollector(ADAPTER, "collector-1"));
|
assertTrue(adapter.stepRecords(channel()).isEmpty());
|
assertTrue(adapter.cycleRecords(channel()).isEmpty());
|
// 落地表在本地，取数不产生厂商负载，所以按展示节奏取数；厂商写库节奏仍是 60 秒。
|
assertEquals(5000L, adapter.persistenceIntervalMillis(channel()));
|
AdapterCapabilities capabilities = adapter.capabilities();
|
assertTrue(capabilities.discovery() && capabilities.realtimeObservation() && !capabilities.control());
|
}
|
|
/** 环境量与量纲确认后落到配置：列名、指标编码、单位、采集器通道都由声明给出，代码里没有厂商列名。 */
|
@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);
|
}
|
}
|
