test(bootstrap): e2e verify gb32960 frame lands in archive store

Spring Boot Test 启用 archive sink 到 JUnit @TempDir，发一条 GB32960 实时上报帧： TCP → Gb32960 Netty → Dispatcher emitRawArchive → Disruptor 虚拟线程 → ArchiveEventSink.publish → LocalArchiveStore.put → 文件落盘断言： - 文件出现在 <tempDir>/yyyy/MM/dd/GB32960/<vin>/*.bin 路径下 - 文件内容字节级等于原始发送的 62 字节帧原始 IngestEndToEndTest 禁用 archive 只验业务事件路径；两个 E2E 互补，保证后续谁动 Dispatcher / AutoConfig / RawFrame 破坏 archive 扇出都会被 CI 立刻红线。 @DynamicPropertySource 把 @TempDir 的路径注入 lingniu.ingest.sink.archive.path，避免硬编码路径导致并行测试相互污染。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-20 16:26:04 +08:00
parent 1c1d400784
commit b14871ff1c
1 changed files with 174 additions and 0 deletions
--- a/bootstrap-all/src/test/java/com/lingniu/ingest/bootstrap/RawArchiveEndToEndTest.java
+++ b/bootstrap-all/src/test/java/com/lingniu/ingest/bootstrap/RawArchiveEndToEndTest.java
@@ -0,0 +1,174 @@
+package com.lingniu.ingest.bootstrap;
+
+import com.lingniu.ingest.codec.BccChecksum;
+import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+
+import java.io.ByteArrayOutputStream;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.time.Instant;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * 端到端验证原始报文冷存链路：
+ *
+ * <pre>
+ *   TCP → Gb32960 Netty → Dispatcher 发 RawArchive 事件
+ *       → Disruptor 虚拟线程 → ArchiveEventSink.publish
+ *       → LocalArchiveStore.put → 文件系统落盘
+ * </pre>
+ *
+ * <p>与 {@link IngestEndToEndTest} 互补：前者验证业务事件路径；这里验证 raw archive
+ * 路径。两个测试跑在独立 Spring 上下文里（archive 是否启用的配置不同）。
+ */
+@SpringBootTest(
+        classes = IngestApplication.class,
+        webEnvironment = SpringBootTest.WebEnvironment.NONE,
+        properties = {
+                "lingniu.ingest.gb32960.enabled=true",
+                "lingniu.ingest.gb32960.port=0",
+                "lingniu.ingest.jt808.enabled=false",
+                "lingniu.ingest.jt1078.enabled=false",
+                "lingniu.ingest.jsatl12.enabled=false",
+                "lingniu.ingest.mqtt.enabled=false",
+                "lingniu.ingest.xinda-push.enabled=false",
+                "lingniu.ingest.command-gateway.enabled=false",
+                "lingniu.ingest.sink.mq.enabled=false",
+                // 启用 archive sink，路径由 @DynamicPropertySource 注入
+                "lingniu.ingest.sink.archive.enabled=true",
+                "lingniu.ingest.sink.archive.type=local"
+        })
+class RawArchiveEndToEndTest {
+
+    @TempDir
+    static Path archiveRoot;
+
+    @DynamicPropertySource
+    static void properties(DynamicPropertyRegistry registry) {
+        registry.add("lingniu.ingest.sink.archive.path", archiveRoot::toString);
+    }
+
+    @Autowired
+    private Gb32960NettyServer server;
+
+    @Test
+    void gb32960FrameIsArchivedToDisk() throws Exception {
+        int port = server.getBoundPort();
+        assertThat(port).isGreaterThan(0);
+
+        String vin = "LTEST0000ARCH0001";
+        byte[] frame = buildRealtimeFrame(vin);
+
+        try (Socket socket = new Socket("127.0.0.1", port)) {
+            socket.getOutputStream().write(frame);
+            socket.getOutputStream().flush();
+        }
+
+        // 预期落盘目录：<root>/yyyy/MM/dd/GB32960/<vin>/
+        // ingestTime 在 Dispatcher 里用 receivedAt = Instant.now()，跨天边界理论上今天/昨天两个候选都可能命中，
+        // 为避免边界脆弱，扫描 root 下所有 .bin 文件匹配 vin。
+        Path file = awaitArchivedFile(archiveRoot, vin, 3000);
+        assertThat(file)
+                .as("expected an archived frame for vin=%s under %s within 3s", vin, archiveRoot)
+                .isNotNull();
+        assertThat(Files.readAllBytes(file)).containsExactly(frame);
+
+        // 额外：key 的日期分片应是 ingestTime 的 UTC 当天（允许跨天时两者都可接受）
+        String today = DateTimeFormatter.ofPattern("yyyy/MM/dd").withZone(ZoneOffset.UTC).format(Instant.now());
+        String yesterday = DateTimeFormatter.ofPattern("yyyy/MM/dd").withZone(ZoneOffset.UTC)
+                .format(Instant.now().minusSeconds(86400));
+        String rel = archiveRoot.relativize(file).toString().replace('\\', '/');
+        assertThat(rel).satisfiesAnyOf(
+                r -> assertThat(r).startsWith(today + "/GB32960/" + vin + "/"),
+                r -> assertThat(r).startsWith(yesterday + "/GB32960/" + vin + "/"));
+        assertThat(rel).endsWith(".bin");
+    }
+
+    private static Path awaitArchivedFile(Path root, String vin, long timeoutMs) throws Exception {
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        while (System.currentTimeMillis() < deadline) {
+            Path hit = findFirst(root, vin);
+            if (hit != null) return hit;
+            Thread.sleep(50);
+        }
+        return null;
+    }
+
+    private static Path findFirst(Path root, String vin) throws Exception {
+        if (!Files.exists(root)) return null;
+        try (Stream<Path> walk = Files.walk(root)) {
+            List<Path> hits = walk
+                    .filter(Files::isRegularFile)
+                    .filter(p -> p.toString().endsWith(".bin"))
+                    .filter(p -> p.toString().contains("/" + vin + "/") || p.toString().contains("\\" + vin + "\\"))
+                    .toList();
+            return hits.isEmpty() ? null : hits.get(0);
+        }
+    }
+
+    // ===== frame builder (inlined copy of IngestEndToEndTest#buildRealtimeFrame) =====
+
+    private static byte[] buildRealtimeFrame(String vin) {
+        ByteArrayOutputStream body = new ByteArrayOutputStream();
+        body.write(24); body.write(1); body.write(2); body.write(3); body.write(4); body.write(5);
+
+        body.write(0x01);
+        body.write(0x01); body.write(0x03); body.write(0x01);
+        writeU16(body, 523);
+        writeU32(body, 1234567);
+        writeU16(body, 6000);
+        writeU16(body, 1100);
+        body.write(70);
+        body.write(0x01);
+        body.write(0x0F);
+        writeU16(body, 500);
+        body.write(30);
+        body.write(0);
+
+        body.write(0x05);
+        body.write(0x00);
+        writeU32(body, 116_397_128L);
+        writeU32(body, 39_916_527L);
+
+        byte[] bodyBytes = body.toByteArray();
+
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        out.write(0x23); out.write(0x23);
+        out.write(0x02);
+        out.write(0xFE);
+        byte[] vinBytes = vin.getBytes(StandardCharsets.US_ASCII);
+        out.write(vinBytes, 0, 17);
+        out.write(0x01);
+        writeU16(out, bodyBytes.length);
+        out.write(bodyBytes, 0, bodyBytes.length);
+        byte[] almost = out.toByteArray();
+        byte bcc = BccChecksum.compute(almost, 2, almost.length - 2);
+        out.write(bcc & 0xFF);
+        return out.toByteArray();
+    }
+
+    private static void writeU16(ByteArrayOutputStream os, int v) {
+        os.write((v >> 8) & 0xFF);
+        os.write(v & 0xFF);
+    }
+
+    private static void writeU32(ByteArrayOutputStream os, long v) {
+        os.write((int) ((v >> 24) & 0xFF));
+        os.write((int) ((v >> 16) & 0xFF));
+        os.write((int) ((v >> 8) & 0xFF));
+        os.write((int) (v & 0xFF));
+    }
+}