package com.battery.business.runtime;
|
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertFalse;
|
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
import static org.junit.jupiter.api.Assertions.assertNull;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
|
import java.io.InputStream;
|
import java.nio.charset.StandardCharsets;
|
import java.sql.Timestamp;
|
import java.time.Instant;
|
import java.util.Date;
|
import java.util.HexFormat;
|
import java.util.List;
|
import java.util.UUID;
|
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
|
import org.junit.jupiter.api.AfterEach;
|
import org.junit.jupiter.api.BeforeEach;
|
import org.junit.jupiter.api.Test;
|
import org.mybatis.spring.SqlSessionFactoryBean;
|
import org.mybatis.spring.SqlSessionTemplate;
|
import org.springframework.core.io.ClassPathResource;
|
import org.springframework.jdbc.core.JdbcTemplate;
|
import org.springframework.jdbc.datasource.init.ScriptUtils;
|
import com.battery.business.domain.TDeviceIngestRecord;
|
import com.battery.business.domain.vo.TDeviceIngestChannel;
|
import com.battery.business.mapper.TDeviceIngestRecordMapper;
|
|
/**
|
* 推式来源落地表的迁移与落库语义（生产 Mapper + 迁移脚本 + H2 内存库）：
|
* 来源行身份让重复推送幂等、同身份不同内容不覆盖；心跳行按来源覆盖接收时间；
|
* 拓扑上报的通道清单带上自己的最近数据时间；水位按 (设备时间, 协议序号) 取严格更晚的行。
|
*/
|
public class PushIngestLandingMigrationContractTest
|
{
|
private static final String PATH = "db/migration-final/V24__push_ingest_landing_table.sql";
|
private static final String VENDOR_CHANNEL_NO_PATH = "db/migration-final/V25__ingest_vendor_channel_no.sql";
|
private static final String ADAPTER = "j22";
|
private static final Instant RECEIVED_AT = Instant.parse("2026-09-27T09:20:00Z");
|
|
private java.sql.Connection keepAlive;
|
private JdbcTemplate jdbc;
|
private TDeviceIngestRecordMapper mapper;
|
|
@BeforeEach void setUp() throws Exception
|
{
|
var source = new UnpooledDataSource("org.h2.Driver",
|
"jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_LOWER=TRUE", "sa", "");
|
keepAlive = source.getConnection();
|
jdbc = new JdbcTemplate(source);
|
// H2 无 MySQL 的 HEX()；测试中补一个等价别名，生产 SQL 仍在 MySQL 上验证。
|
jdbc.execute("create alias HEX for 'com.battery.business.runtime.PushIngestLandingMigrationContractTest.hexOf'");
|
// 按顺序跑落地表相关的迁移，使内存库结构与生产库一致（V24 建表、V25 加来源通道号列）。
|
ScriptUtils.executeSqlScript(keepAlive, new ClassPathResource(PATH));
|
ScriptUtils.executeSqlScript(keepAlive, new ClassPathResource(VENDOR_CHANNEL_NO_PATH));
|
|
SqlSessionFactoryBean factory = new SqlSessionFactoryBean();
|
factory.setDataSource(source);
|
factory.setTypeAliasesPackage("com.battery.business.domain");
|
factory.setMapperLocations(new ClassPathResource("mapper/battery/TDeviceIngestRecordMapper.xml"));
|
mapper = new SqlSessionTemplate(factory.getObject()).getMapper(TDeviceIngestRecordMapper.class);
|
}
|
|
@AfterEach void closeDatabase() throws Exception { if (keepAlive != null) keepAlive.close(); }
|
|
@Test
|
void migrationCreatesTheLandingTableWithoutWritingAnyData()
|
{
|
String migration = resource(PATH);
|
// 迁移只建表：不导入演示数据、不改写任何既有表。
|
assertFalse(migration.matches("(?s).*(INSERT|UPDATE|DELETE|DROP)\\b.*"));
|
assertTrue(migration.contains("UNIQUE KEY uk_ingest_record (adapter_code, payload_kind, record_key)"));
|
|
assertEquals(0, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));
|
// 落地表保留厂商原始载荷，且测量列可以缺列（设备没给就是空，不写零代替缺失）。
|
jdbc.update("insert into t_device_ingest_record(adapter_code, payload_kind, record_key, payload_json,"
|
+ " content_sha256, received_at_utc) values(?,?,?,?,?,?)", ADAPTER, "SAMPLE", "k",
|
"{\"Voltage\":1.2}", new byte[32], Timestamp.from(RECEIVED_AT));
|
assertEquals(1, jdbc.queryForObject(
|
"select count(*) from t_device_ingest_record where voltage_v is null", Integer.class));
|
}
|
|
/**
|
* 来源自己的通道叫法那条迁移：只加一个可空列，不写入任何数据；
|
* 老行读出来是空（界面回退到「单元-通道」），新行可以带上来源软件里显示的通道号。
|
*/
|
@Test
|
void vendorChannelNoMigrationOnlyAddsANullableColumn()
|
{
|
String migration = resource(VENDOR_CHANNEL_NO_PATH);
|
assertTrue(migration.contains("ADD COLUMN vendor_channel_no INT NULL"));
|
assertFalse(migration.matches("(?s).*(INSERT|UPDATE|DELETE|DROP)\\b.*"));
|
|
jdbc.update("insert into t_device_ingest_record(adapter_code, payload_kind, record_key, payload_json,"
|
+ " content_sha256, received_at_utc) values(?,?,?,?,?,?)", ADAPTER, "TOPOLOGY", "topology:26",
|
"{\"channel\":26}", new byte[32], Timestamp.from(RECEIVED_AT));
|
// 迁移在 setUp 里已按顺序执行过（V24 建表、V25 加列），这里只核对列的行为。
|
assertNull(jdbc.queryForObject(
|
"select vendor_channel_no from t_device_ingest_record where record_key = 'topology:26'",
|
Integer.class));
|
jdbc.update("update t_device_ingest_record set vendor_channel_no = ? where record_key = ?", 27,
|
"topology:26");
|
assertEquals(27, jdbc.queryForObject(
|
"select vendor_channel_no from t_device_ingest_record where record_key = 'topology:26'",
|
Integer.class));
|
// 展示列不参与身份：唯一键仍只由来源行身份决定。
|
assertEquals(1, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));
|
}
|
|
@Test
|
void repeatedPushIsIgnoredBySourceRowIdentityAndDifferentContentIsKept()
|
{
|
TDeviceIngestRecord first = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.2}");
|
assertEquals(1, mapper.insertIngestRecord(first));
|
|
// 同身份同内容：幂等忽略，库里仍是一行。
|
TDeviceIngestRecord repeat = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.2}");
|
assertEquals(0, mapper.insertIngestRecord(repeat));
|
assertEquals(1, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));
|
|
// 同身份不同内容：内容摘要不同，调用方据此报冲突；既有那一行不被覆盖。
|
TDeviceIngestRecord conflict = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}");
|
assertEquals(0, mapper.insertIngestRecord(conflict));
|
assertEquals(HexFormat.of().formatHex(first.getContentSha256()),
|
mapper.selectIngestRecordHashByKey(ADAPTER, "SAMPLE", first.getRecordKey()));
|
// H2 的 JSON 列会再引一层，所以只断内容；既有那一行没有被冲突推送覆盖。
|
assertTrue(jdbc.queryForObject("select payload_json from t_device_ingest_record", String.class)
|
.contains("Voltage"));
|
}
|
|
@Test
|
void heartbeatRowIsUpsertedPerSourceSoAHeartbeatNeverGrows()
|
{
|
mapper.upsertHeartbeat(heartbeat(RECEIVED_AT));
|
assertEquals(Timestamp.from(RECEIVED_AT).toInstant(), mapper.selectHeartbeatAt(ADAPTER).toInstant());
|
|
// 下一轮心跳只推进接收时间：它表示"取数程序还在推"，不是设备事实，没有保留历史的意义。
|
Instant later = RECEIVED_AT.plusSeconds(60);
|
mapper.upsertHeartbeat(heartbeat(later));
|
assertEquals(Timestamp.from(later).toInstant(), mapper.selectHeartbeatAt(ADAPTER).toInstant());
|
assertEquals(1, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));
|
// 从未收到推送的来源没有心跳事实。
|
assertNull(mapper.selectHeartbeatAt("other"));
|
}
|
|
@Test
|
void topologyChannelsCarryTheirOwnLatestSampleTimeAndNeverExpire()
|
{
|
mapper.insertIngestRecord(topology(26));
|
mapper.insertIngestRecord(topology(18));
|
// 18 号只有更早的采样行；26 号有两行。
|
mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:11", 11L, "{\"Voltage\":1.2}",
|
Instant.parse("2026-09-27T07:52:20Z")));
|
mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
|
Instant.parse("2026-09-27T07:53:20Z")));
|
mapper.insertIngestRecord(sample(18, "DB_20260923_161840/Ws18_IV_0:5", 5L, "{\"Voltage\":1.1}",
|
Instant.parse("2026-09-23T08:18:40Z")));
|
|
List<TDeviceIngestChannel> channels = mapper.selectTopologyChannels(ADAPTER);
|
assertEquals(List.of(18, 26), channels.stream().map(TDeviceIngestChannel::getChlid).toList());
|
assertEquals(Instant.parse("2026-09-23T08:18:40Z"), channels.get(0).getLastDataAt().toInstant());
|
assertEquals(Instant.parse("2026-09-27T07:53:20Z"), channels.get(1).getLastDataAt().toInstant());
|
// 没有采样行的通道不给"最近数据时间"：这个事实是观测到的，不按上报时间代替。
|
mapper.insertIngestRecord(topology(21));
|
assertNull(mapper.selectTopologyChannels(ADAPTER).stream()
|
.filter(channel -> channel.getChlid() == 21).findFirst().orElseThrow().getLastDataAt());
|
}
|
|
@Test
|
void samplesSinceReturnsOnlyRowsStrictlyAfterTheWatermarkIncludingSameSecondRows()
|
{
|
TDeviceIngestRecord before = sample(26, "DB_20260927_155209/Ws26_IV_0:11", 11L, "{\"Voltage\":1.2}",
|
Instant.parse("2026-09-27T07:52:20Z"));
|
TDeviceIngestRecord sameSecond = sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
|
Instant.parse("2026-09-27T07:53:20Z"));
|
TDeviceIngestRecord sameSecondLaterSequence = sample(26, "DB_20260927_155209/Ws26_IV_0:13", 13L,
|
"{\"Voltage\":1.4}", Instant.parse("2026-09-27T07:53:20Z"));
|
TDeviceIngestRecord after = sample(26, "DB_20260927_155209/Ws26_IV_0:14", 14L, "{\"Voltage\":1.5}",
|
Instant.parse("2026-09-27T07:54:20Z"));
|
for (TDeviceIngestRecord row : List.of(before, sameSecond, sameSecondLaterSequence, after))
|
mapper.insertIngestRecord(row);
|
|
// 水位落在一行上：那一行自己不算，严格更晚的才进本次接收。
|
List<TDeviceIngestRecord> rows = mapper.selectSamplesSince(ADAPTER, 0, 0, 1, 26,
|
Date.from(sameSecond.getDeviceTimeUtc().toInstant()), 12L, 500);
|
assertEquals(List.of(13L, 14L), rows.stream().map(TDeviceIngestRecord::getRecordSequence).toList());
|
|
// 同一绝对秒内的多行各自成立：水位为空表示边界时刻什么也没有，之后的行全部属于本次接收。
|
assertEquals(4, mapper.selectSamplesSince(ADAPTER, 0, 0, 1, 26, null, 0L, 500).size());
|
// 其他通道与来源的行不混进来。
|
assertTrue(mapper.selectSamplesSince("other", 0, 0, 1, 26, null, 0L, 500).isEmpty());
|
}
|
|
@Test
|
void latestSampleDrivesBothRealtimeObservationAndWatermark()
|
{
|
assertNull(mapper.selectLatestSample(ADAPTER, 0, 0, 1, 26));
|
mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
|
Instant.parse("2026-09-27T07:53:20Z")));
|
TDeviceIngestRecord latest = mapper.selectLatestSample(ADAPTER, 0, 0, 1, 26);
|
assertNotNull(latest);
|
assertEquals(12L, latest.getRecordSequence());
|
assertEquals("2026-09-27 15:53:20", latest.getDeviceTimeRaw());
|
assertEquals(Instant.parse("2026-09-27T07:53:20Z"), latest.getDeviceTimeUtc().toInstant());
|
// 原始行逐字留在落地表里，回放与核对解释逻辑都靠它（H2 的 JSON 列会再引一层，所以只断内容）。
|
assertTrue(latest.getPayloadJson().contains("Voltage"));
|
}
|
|
/**
|
* 工步与循环的分段推导读的就是这条查询（ADR-0067）：某一个来源库在该通道上的采样行，
|
* 按 (设备时间, 协议序号) 升序——推导依赖这个次序判断哪一段还在进行；其他来源库、
|
* 其他通道与其他来源的行都不混进来。
|
*/
|
@Test
|
void samplesOfOneSourceRefComeBackInWatermarkOrderAndNothingElseDoes()
|
{
|
mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
|
Instant.parse("2026-09-27T07:53:20Z")));
|
mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:11", 11L, "{\"Voltage\":1.2}",
|
Instant.parse("2026-09-27T07:52:20Z")));
|
// 换了来源库（厂商侧另一次测试）、另一个通道：都不属于本次推导范围。
|
TDeviceIngestRecord laterTest = sample(26, "DB_20260928_101010/Ws26_IV_0:1", 1L, "{\"Voltage\":1.4}",
|
Instant.parse("2026-09-28T02:10:10Z"));
|
laterTest.setSourceRef("DB_20260928_101010/Ws26_IV_0");
|
mapper.insertIngestRecord(laterTest);
|
mapper.insertIngestRecord(sample(18, "DB_20260923_161840/Ws18_IV_0:5", 5L, "{\"Voltage\":1.1}",
|
Instant.parse("2026-09-23T08:18:40Z")));
|
|
List<TDeviceIngestRecord> rows = mapper.selectSamplesOfSource(ADAPTER, 0, 0, 1, 26,
|
"DB_20260927_155209/Ws26_IV_0");
|
|
assertEquals(List.of(11L, 12L), rows.stream().map(TDeviceIngestRecord::getRecordSequence).toList());
|
assertTrue(mapper.selectSamplesOfSource(ADAPTER, 0, 0, 1, 26, "DB_20261001_000000/Ws26_IV_0").isEmpty());
|
}
|
|
/** 供 H2 别名使用的 HEX 等价实现；H2 没有 MySQL 的 HEX()。 */
|
public static String hexOf(byte[] value)
|
{
|
return value == null ? null : HexFormat.of().formatHex(value);
|
}
|
|
/** 内容摘要固定 32 字节：列是 BINARY(32)，短值会被补齐，比较时对不上。 */
|
private static byte[] digest(String content)
|
{
|
try
|
{
|
return java.security.MessageDigest.getInstance("SHA-256")
|
.digest(content.getBytes(StandardCharsets.UTF_8));
|
}
|
catch (Exception exception)
|
{
|
throw new IllegalStateException(exception);
|
}
|
}
|
|
private TDeviceIngestRecord sample(String recordKey, long sequence, String payload)
|
{
|
return sample(26, recordKey, sequence, payload, Instant.parse("2026-09-27T07:53:20Z"));
|
}
|
|
private TDeviceIngestRecord sample(int channel, String recordKey, long sequence, String payload)
|
{
|
return sample(channel, recordKey, sequence, payload, Instant.parse("2026-09-27T07:53:20Z"));
|
}
|
|
private TDeviceIngestRecord sample(int channel, String recordKey, long sequence, String payload,
|
Instant deviceTime)
|
{
|
TDeviceIngestRecord value = new TDeviceIngestRecord();
|
value.setAdapterCode(ADAPTER);
|
value.setPayloadKind(TDeviceIngestRecord.KIND_SAMPLE);
|
value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(channel);
|
value.setRecordSequence(sequence);
|
value.setRecordKey(recordKey);
|
value.setSourceRef("DB_20260927_155209/Ws26_IV_0");
|
value.setDeviceTimeUtc(Timestamp.from(deviceTime));
|
value.setDeviceTimeRaw("2026-09-27 15:53:20");
|
value.setTimezoneId("Asia/Shanghai");
|
value.setPayloadJson(payload);
|
value.setContentSha256(digest(payload));
|
value.setReceivedAtUtc(Timestamp.from(RECEIVED_AT));
|
return value;
|
}
|
|
private TDeviceIngestRecord topology(int channel)
|
{
|
TDeviceIngestRecord value = new TDeviceIngestRecord();
|
value.setAdapterCode(ADAPTER);
|
value.setPayloadKind(TDeviceIngestRecord.KIND_TOPOLOGY);
|
value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(channel);
|
value.setRecordSequence(0L);
|
value.setRecordKey("topology:" + channel);
|
value.setPayloadJson("{\"channel\":" + channel + "}");
|
value.setContentSha256(digest("topology:" + channel));
|
value.setReceivedAtUtc(Timestamp.from(RECEIVED_AT));
|
return value;
|
}
|
|
private TDeviceIngestRecord heartbeat(Instant receivedAt)
|
{
|
TDeviceIngestRecord value = new TDeviceIngestRecord();
|
value.setAdapterCode(ADAPTER);
|
value.setPayloadKind(TDeviceIngestRecord.KIND_HEARTBEAT);
|
value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(0);
|
value.setRecordSequence(0L);
|
value.setRecordKey("heartbeat");
|
value.setPayloadJson("{}");
|
value.setContentSha256(new byte[32]);
|
value.setReceivedAtUtc(Timestamp.from(receivedAt));
|
return value;
|
}
|
|
private static String resource(String path)
|
{
|
try (InputStream stream = PushIngestLandingMigrationContractTest.class.getClassLoader()
|
.getResourceAsStream(path))
|
{
|
if (stream == null) throw new IllegalStateException("缺少迁移文件：" + path);
|
return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
|
}
|
catch (java.io.IOException exception)
|
{
|
throw new IllegalStateException("无法读取迁移文件：" + path, exception);
|
}
|
}
|
}
|
