diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index 649ad1a7..10d6506d 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -225,15 +225,23 @@ JT808 真实转发链路已验证： 仓库提供 `tools/jt808_e2e_smoke.py`，用于重复验证 TCP `808` 到 TDengine history 的闭环： ```bash +export TDENGINE_REST_URL='http://:6041/rest/sql/vehicle_ts' +export TDENGINE_USERNAME='root' +export TDENGINE_PASSWORD='' + python3 tools/jt808_e2e_smoke.py \ + --connection-mode session \ --start-phone 13079962000 \ --frames 3 \ --expect-history-count 3 \ --verify-pagination \ - --archive-root "$PROJECT_ROOT/data/archive-jt808" + --archive-root "$PROJECT_ROOT/data/archive-jt808" \ + --tdengine-rest-url "$TDENGINE_REST_URL" \ + --tdengine-username "$TDENGINE_USERNAME" \ + --tdengine-password "$TDENGINE_PASSWORD" ``` -默认 `--connection-mode per-frame`，适合稳定 smoke。需要暴露长连接连续帧边界时可切换 `--connection-mode session`。脚本会输出发送数量、history 可见数量、分页验证结果和 raw archive 检查结果。 +默认 `--connection-mode per-frame`，适合单帧连接 smoke。生产验收建议使用 `--connection-mode session` 覆盖同一连接内连续位置帧。脚本会输出发送数量、history 可见数量、分页验证结果、raw archive 检查结果和 `tdengineRawFrames`。`tdengineRawFrames` 表示本次可见位置记录中的唯一 `rawUri` 有多少个已确认写入 TDengine `raw_frames`。 - 合成 0x0100 注册帧终端号 `13079969999` 已验证：TCP `808` 返回 `0x8100` 注册 ACK，`raw_frames.metadata_json` 包含 `jt808.register.province`、`jt808.register.city`、`jt808.register.maker`、`jt808.register.deviceType`、`jt808.register.deviceId`、`jt808.register.plateColor`、`jt808.register.plate`，对应 raw archive 文件存在。 复查 SQL： @@ -277,13 +285,29 @@ SELECT COUNT(*) FROM raw_frames WHERE protocol = 'GB32960' AND vin = 'LTEST202606290001'; SELECT COUNT(*) FROM telemetry_fields WHERE protocol = 'GB32960' AND vin = 'LTEST202606290001'; -SELECT event_time, vin, raw_uri, raw_id +SELECT event_time, vin, raw_uri, frame_id FROM raw_frames WHERE protocol = 'GB32960' AND vin = 'LTEST202606290001' ORDER BY event_time DESC LIMIT 5; ``` +也可以直接运行仓库 smoke 工具验证 TCP `32960`、history records、TDengine 字段、TDengine `raw_frames` 和 raw archive： + +```bash +export TDENGINE_REST_URL='http://:6041/rest/sql/vehicle_ts' +export TDENGINE_USERNAME='root' +export TDENGINE_PASSWORD='' + +python3 tools/gb32960_e2e_smoke.py \ + --archive-root "$PROJECT_ROOT/data/archive-jt808" \ + --tdengine-rest-url "$TDENGINE_REST_URL" \ + --tdengine-username "$TDENGINE_USERNAME" \ + --tdengine-password "$TDENGINE_PASSWORD" +``` + +预期输出包含 `records >= 1`、`fieldCounts` 中关键字段均大于 0、`tdengineRawFrames` 等于本次可见唯一 `rawUri` 数量、`archiveChecked >= 1`。 + 复查 API： ```bash diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java index 42ad5d33..9322d071 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java @@ -30,7 +30,7 @@ public final class TdengineEnvelopeRows { metadata.putAll(envelope.getMetadataMap()); metadata.putAll(raw.getMetadataMap()); return Optional.of(new TdengineRawFrameRow( - instant(envelope.getEventTimeMs()), + rawFrameTs(envelope, metadata), raw.getFrameId(), instant(envelope.getIngestTimeMs()), raw.getMessageId(), @@ -142,6 +142,24 @@ public final class TdengineEnvelopeRows { return Instant.ofEpochMilli(epochMillis); } + private static Instant rawFrameTs(VehicleEnvelope envelope, Map metadata) { + String rawArchiveEventId = firstNonBlank(metadata.get("rawArchiveEventId"), envelope.getEventId()); + if (rawArchiveEventId == null || rawArchiveEventId.isBlank()) { + return instant(envelope.getEventTimeMs()); + } + try { + long sequenceId = Long.parseLong(rawArchiveEventId); + long baseMillis = sequenceId / 1000L; + long sameMillisecondOffset = sequenceId % 1000L; + if (Math.abs(baseMillis - envelope.getEventTimeMs()) > 60_000L) { + return instant(envelope.getEventTimeMs()); + } + return Instant.ofEpochMilli(baseMillis + sameMillisecondOffset); + } catch (NumberFormatException ignored) { + return instant(envelope.getEventTimeMs()); + } + } + private static String protocol(VehicleEnvelope envelope) { return firstNonBlank(envelope.getSource(), "UNKNOWN"); } diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java index e2a8152c..e70e40b9 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java @@ -51,6 +51,20 @@ class TdengineEnvelopeRowsTest { assertThat(row.metadataJson()).contains("\"channel\":\"tcp-808\"", "\"auth\":\"passed\""); } + @Test + void expandsRawFrameTsWithRawArchiveEventIdSequenceToAvoidSameMillisecondOverwrite() { + VehicleEnvelope first = rawFrameEnvelope("1782729152494000"); + VehicleEnvelope second = rawFrameEnvelope("1782729152494001"); + + TdengineRawFrameRow firstRow = TdengineEnvelopeRows.rawFrame(first).orElseThrow(); + TdengineRawFrameRow secondRow = TdengineEnvelopeRows.rawFrame(second).orElseThrow(); + + assertThat(firstRow.ts()).isEqualTo(Instant.ofEpochMilli(1_782_729_152_494L)); + assertThat(secondRow.ts()).isEqualTo(Instant.ofEpochMilli(1_782_729_152_495L)); + assertThat(firstRow.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_729_152_494L)); + assertThat(secondRow.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_729_152_494L)); + } + @Test void mapsLocationEnvelopeToVehicleLocationRow() { VehicleEnvelope envelope = VehicleEnvelope.newBuilder() @@ -190,4 +204,25 @@ class TdengineEnvelopeRowsTest { assertThat(TdengineEnvelopeRows.telemetryFields(envelope).getFirst().vehicleKey()) .isEqualTo("jt808:13079963291"); } + + private static VehicleEnvelope rawFrameEnvelope(String rawArchiveEventId) { + return VehicleEnvelope.newBuilder() + .setEventId(rawArchiveEventId) + .setVin("unknown") + .setSource("JT808") + .setEventTimeMs(1_782_729_152_494L) + .setIngestTimeMs(1_782_729_152_539L) + .putMetadata("phone", "13079973200") + .putMetadata("rawArchiveEventId", rawArchiveEventId) + .setRawFrameFact(RawFrameFactPayload.newBuilder() + .setFrameId("rf-" + rawArchiveEventId) + .setPhone("13079973200") + .setMessageId(0x0200) + .setRawUri("archive://2026/06/29/JT808/unknown/" + rawArchiveEventId + ".bin") + .setChecksum("sha256:" + rawArchiveEventId) + .setRawSizeBytes(41) + .setParseStatus(ParseStatusProto.PARSE_STATUS_NOT_PARSED) + .putMetadata("rawArchiveEventId", rawArchiveEventId)) + .build(); + } } diff --git a/tools/gb32960_e2e_smoke.py b/tools/gb32960_e2e_smoke.py index 91aac0de..35819dd5 100644 --- a/tools/gb32960_e2e_smoke.py +++ b/tools/gb32960_e2e_smoke.py @@ -6,6 +6,7 @@ from __future__ import annotations import argparse import datetime as dt import json +import os import pathlib import socket import time @@ -14,6 +15,11 @@ import urllib.request from dataclasses import dataclass from typing import Any +try: + from tools import tdengine_smoke +except ModuleNotFoundError: + import tdengine_smoke + SHANGHAI = dt.timezone(dt.timedelta(hours=8)) DEFAULT_SAMPLE = ( @@ -144,6 +150,25 @@ def wait_for_fields( return last_items +def wait_for_tdengine_raw_frames( + rest_url: str, + username: str, + password: str, + sql: str, + minimum: int, + timeout_seconds: float, + request_timeout: float, +) -> int: + deadline = time.monotonic() + timeout_seconds + last_count = 0 + while time.monotonic() < deadline: + last_count = tdengine_smoke.query_count(rest_url, username, password, sql, request_timeout) + if last_count >= minimum: + return last_count + time.sleep(1) + return last_count + + def parse_field_keys(value: str) -> list[str]: return [item.strip() for item in value.split(",") if item.strip()] @@ -172,6 +197,50 @@ def check_archive(root: pathlib.Path, records: list[dict[str, Any]], limit: int) return checked +def raw_frame_count_sql(info: FrameInfo, records: list[dict[str, Any]]) -> str: + sqls = raw_frame_count_sqls(info, records) + if not sqls: + raise ValueError("no raw archive uri found in gb32960 records") + return sqls[0] + + +def raw_frame_count_sqls(info: FrameInfo, records: list[dict[str, Any]]) -> list[str]: + seen: set[str] = set() + sqls: list[str] = [] + for record in records: + uri = record.get("rawArchiveUri") or "" + if not uri or uri in seen: + continue + seen.add(uri) + sqls.append(tdengine_smoke.raw_frame_count_sql( + protocol="GB32960", + vehicle_key=info.vin, + vin=info.vin, + raw_uri=uri, + )) + return sqls + + +def verify_tdengine_raw_frames( + rest_url: str, + username: str, + password: str, + sqls: list[str], + timeout_seconds: float, + request_timeout: float, +) -> int: + if not sqls: + raise RuntimeError("no raw archive uri available for tdengine raw_frames verification") + verified = 0 + for sql in sqls: + count = wait_for_tdengine_raw_frames( + rest_url, username, password, sql, 1, timeout_seconds, request_timeout) + if count < 1: + raise RuntimeError(f"tdengine raw_frames row not visible for query: {sql}") + verified += 1 + return verified + + def run(args: argparse.Namespace) -> dict[str, Any]: frame = read_hex_frame(pathlib.Path(args.frame_hex_file)) info = parse_frame_info(frame) @@ -212,6 +281,17 @@ def run(args: argparse.Namespace) -> dict[str, Any]: if args.archive_root: archive_checked = check_archive(pathlib.Path(args.archive_root), records, args.archive_check_limit) + tdengine_raw_frames = None + if args.tdengine_rest_url: + tdengine_raw_frames = verify_tdengine_raw_frames( + args.tdengine_rest_url, + args.tdengine_username, + args.tdengine_password, + raw_frame_count_sqls(info, records), + args.history_timeout, + args.http_timeout, + ) + return { "tcpHost": args.tcp_host, "tcpPort": args.tcp_port, @@ -221,6 +301,7 @@ def run(args: argparse.Namespace) -> dict[str, Any]: "ackBytesHex": ack.hex(), "records": len(records), "fieldCounts": field_counts, + "tdengineRawFrames": tdengine_raw_frames, "archiveChecked": archive_checked, "firstRawUri": first_raw_uri(records), } @@ -245,6 +326,9 @@ def parser() -> argparse.ArgumentParser: p.add_argument("--expect-record-count", type=int, default=1) p.add_argument("--archive-root", default="") p.add_argument("--archive-check-limit", type=int, default=1) + p.add_argument("--tdengine-rest-url", default=os.environ.get("TDENGINE_REST_URL", "")) + p.add_argument("--tdengine-username", default=os.environ.get("TDENGINE_USERNAME", "root")) + p.add_argument("--tdengine-password", default=os.environ.get("TDENGINE_PASSWORD", "taosdata")) p.add_argument("--require-ack", action="store_true", default=True) p.add_argument("--tcp-timeout", type=float, default=5) p.add_argument("--http-timeout", type=float, default=5) diff --git a/tools/jt808_e2e_smoke.py b/tools/jt808_e2e_smoke.py index f15f02b8..0ca2990b 100755 --- a/tools/jt808_e2e_smoke.py +++ b/tools/jt808_e2e_smoke.py @@ -6,6 +6,7 @@ from __future__ import annotations import argparse import datetime as dt import json +import os import pathlib import socket import time @@ -13,6 +14,11 @@ import urllib.parse import urllib.request from typing import Any +try: + from tools import tdengine_smoke +except ModuleNotFoundError: + import tdengine_smoke + def bcd(value: str) -> bytes: digits = "".join(ch for ch in value if ch.isdigit()) @@ -247,6 +253,69 @@ def archive_path(root: pathlib.Path, uri: str) -> pathlib.Path: return root / uri[len(prefix):] +def raw_frame_count_sql(phone: str, items: list[dict[str, Any]]) -> str: + sqls = raw_frame_count_sqls(phone, items) + if not sqls: + raise ValueError("no raw uri found in jt808 location items") + return sqls[0] + + +def raw_frame_count_sqls(phone: str, items: list[dict[str, Any]]) -> list[str]: + seen: set[str] = set() + sqls: list[str] = [] + for item in items: + raw_uri = item.get("rawUri") or "" + if not raw_uri or raw_uri in seen: + continue + seen.add(raw_uri) + sqls.append(tdengine_smoke.raw_frame_count_sql( + protocol="JT808", + vehicle_key=item.get("vehicleKey") or ("jt808:" + phone), + phone=phone, + raw_uri=raw_uri, + )) + return sqls + + +def verify_tdengine_raw_frames( + rest_url: str, + username: str, + password: str, + sqls: list[str], + timeout_seconds: float, + request_timeout: float, +) -> int: + if not sqls: + raise RuntimeError("no raw uri available for tdengine raw_frames verification") + verified = 0 + for sql in sqls: + count = wait_for_tdengine_raw_frames( + rest_url, username, password, sql, 1, timeout_seconds, request_timeout) + if count < 1: + raise RuntimeError(f"tdengine raw_frames row not visible for query: {sql}") + verified += 1 + return verified + + +def wait_for_tdengine_raw_frames( + rest_url: str, + username: str, + password: str, + sql: str, + minimum: int, + timeout_seconds: float, + request_timeout: float, +) -> int: + deadline = time.monotonic() + timeout_seconds + last_count = 0 + while time.monotonic() < deadline: + last_count = tdengine_smoke.query_count(rest_url, username, password, sql, request_timeout) + if last_count >= minimum: + return last_count + time.sleep(1) + return last_count + + def wait_for_locations( history_base_url: str, phone: str, @@ -285,6 +354,13 @@ def generated_phone(start_phone: str, offset: int) -> str: return str(int(start_phone) + offset) +def expected_history_count(args: argparse.Namespace) -> int: + minimum = min(args.frames, args.expect_history_count) + if args.verify_pagination and args.frames > 1: + minimum = max(minimum, 2) + return minimum + + def run(args: argparse.Namespace) -> dict[str, Any]: event_time = parse_event_time(args.event_time) date_from = (event_time - dt.timedelta(minutes=5)).isoformat() @@ -339,12 +415,12 @@ def run(args: argparse.Namespace) -> dict[str, Any]: first_phone, date_from, date_to, - min(args.frames, args.expect_history_count), + expected_history_count(args), args.history_timeout, args.http_timeout, ) items = page.get("items") or [] - if len(items) < min(args.frames, args.expect_history_count): + if len(items) < expected_history_count(args): raise RuntimeError(f"history rows not visible for {first_phone}: got {len(items)}") archive_checked = 0 @@ -356,6 +432,17 @@ def run(args: argparse.Namespace) -> dict[str, Any]: raise RuntimeError(f"raw archive missing: {path}") archive_checked += 1 + tdengine_raw_frames = None + if args.tdengine_rest_url: + tdengine_raw_frames = verify_tdengine_raw_frames( + args.tdengine_rest_url, + args.tdengine_username, + args.tdengine_password, + raw_frame_count_sqls(first_phone, items), + args.history_timeout, + args.http_timeout, + ) + pagination_ok = None if args.verify_pagination and args.frames > 1: pagination_ok = verify_pagination(args.history_base_url, first_phone, date_from, date_to, args.http_timeout) @@ -372,6 +459,7 @@ def run(args: argparse.Namespace) -> dict[str, Any]: "sendElapsedSeconds": round(elapsed, 3), "sendLocationsPerSecond": round(sent_locations / elapsed, 2), "historyRowsForFirstPhone": len(items), + "tdengineRawFrames": tdengine_raw_frames, "archiveChecked": archive_checked, "paginationOk": pagination_ok, "firstFrameId": items[0].get("frameId") if items else None, @@ -396,6 +484,9 @@ def parser() -> argparse.ArgumentParser: p.add_argument("--verify-pagination", action="store_true") p.add_argument("--archive-root", default="") p.add_argument("--archive-check-limit", type=int, default=3) + p.add_argument("--tdengine-rest-url", default=os.environ.get("TDENGINE_REST_URL", "")) + p.add_argument("--tdengine-username", default=os.environ.get("TDENGINE_USERNAME", "root")) + p.add_argument("--tdengine-password", default=os.environ.get("TDENGINE_PASSWORD", "taosdata")) p.add_argument("--require-register-ack", action="store_true", default=True) p.add_argument("--tcp-timeout", type=float, default=2) p.add_argument("--http-timeout", type=float, default=5) diff --git a/tools/tdengine_smoke.py b/tools/tdengine_smoke.py new file mode 100644 index 00000000..06ad0524 --- /dev/null +++ b/tools/tdengine_smoke.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +"""Small TDengine REST helpers for local smoke verifiers.""" + +from __future__ import annotations + +import base64 +import json +import urllib.request +from typing import Any + + +def sql_literal(value: str) -> str: + return "'" + (value or "").replace("'", "''") + "'" + + +def raw_frame_count_sql( + *, + protocol: str, + vehicle_key: str = "", + vin: str = "", + phone: str = "", + raw_uri: str = "", +) -> str: + if not protocol or not protocol.strip(): + raise ValueError("protocol is required") + filters = ["protocol = " + sql_literal(protocol.strip().upper())] + optional_filters = { + "vehicle_key": vehicle_key, + "vin": vin, + "phone": phone, + "raw_uri": raw_uri, + } + present = False + for column, value in optional_filters.items(): + if value and value.strip(): + filters.append(column + " = " + sql_literal(value.strip())) + present = True + if not present: + raise ValueError("at least one identity or raw_uri filter is required") + return "SELECT COUNT(*) FROM raw_frames WHERE " + " AND ".join(filters) + + +def count_from_response(response: dict[str, Any]) -> int: + if response.get("code") != 0: + raise RuntimeError(f"tdengine query failed: {response}") + data = response.get("data") or [] + if not data or not data[0]: + return 0 + return int(data[0][0]) + + +def query_count(rest_url: str, username: str, password: str, sql: str, timeout: float) -> int: + request = urllib.request.Request(rest_url, data=sql.encode("utf-8"), method="POST") + if username: + token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii") + request.add_header("Authorization", "Basic " + token) + request.add_header("Content-Type", "text/plain; charset=utf-8") + request.add_header("Accept", "application/json") + with urllib.request.urlopen(request, timeout=timeout) as response: + payload = json.loads(response.read().decode("utf-8")) + return count_from_response(payload) diff --git a/tools/test_gb32960_e2e_smoke.py b/tools/test_gb32960_e2e_smoke.py index a1e76118..66a16e0b 100644 --- a/tools/test_gb32960_e2e_smoke.py +++ b/tools/test_gb32960_e2e_smoke.py @@ -58,6 +58,32 @@ class Gb32960E2ESmokeTest(unittest.TestCase): self.assertEqual(uri, "archive://2026/06/29/GB32960/LTEST/frame.bin") + def test_raw_frame_count_sql_filters_by_vin_and_raw_uri(self): + info = smoke.FrameInfo("LTEST202606290001", 0x02, None) + sql = smoke.raw_frame_count_sql(info, [ + {"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST202606290001/frame.bin"}, + ]) + + self.assertEqual( + sql, + "SELECT COUNT(*) FROM raw_frames WHERE protocol = 'GB32960' " + "AND vehicle_key = 'LTEST202606290001' " + "AND vin = 'LTEST202606290001' " + "AND raw_uri = 'archive://2026/06/29/GB32960/LTEST202606290001/frame.bin'", + ) + + def test_raw_frame_count_sqls_include_each_unique_record_raw_uri(self): + info = smoke.FrameInfo("LTEST202606290001", 0x02, None) + sqls = smoke.raw_frame_count_sqls(info, [ + {"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST202606290001/frame-1.bin"}, + {"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST202606290001/frame-2.bin"}, + {"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST202606290001/frame-1.bin"}, + ]) + + self.assertEqual(len(sqls), 2) + self.assertTrue(sqls[0].endswith("frame-1.bin'")) + self.assertTrue(sqls[1].endswith("frame-2.bin'")) + def test_check_archive_ignores_duplicate_uris(self): with tempfile.TemporaryDirectory() as temp_dir: root = pathlib.Path(temp_dir) diff --git a/tools/test_jt808_e2e_smoke.py b/tools/test_jt808_e2e_smoke.py index 34318789..c01deaca 100644 --- a/tools/test_jt808_e2e_smoke.py +++ b/tools/test_jt808_e2e_smoke.py @@ -1,4 +1,5 @@ import pathlib +import types import unittest from tools import jt808_e2e_smoke as smoke @@ -33,6 +34,47 @@ class Jt808E2ESmokeTest(unittest.TestCase): self.assertEqual(path, pathlib.Path("/tmp/archive/2026/06/29/JT808/unknown/frame.bin")) + def test_raw_frame_count_sql_filters_by_vehicle_key_phone_and_raw_uri(self): + sql = smoke.raw_frame_count_sql("13079961234", [ + { + "vehicleKey": "jt808:13079961234", + "rawUri": "archive://2026/06/29/JT808/13079961234/frame.bin", + }, + ]) + + self.assertEqual( + sql, + "SELECT COUNT(*) FROM raw_frames WHERE protocol = 'JT808' " + "AND vehicle_key = 'jt808:13079961234' " + "AND phone = '13079961234' " + "AND raw_uri = 'archive://2026/06/29/JT808/13079961234/frame.bin'", + ) + + def test_raw_frame_count_sqls_include_each_unique_visible_raw_uri(self): + sqls = smoke.raw_frame_count_sqls("13079961234", [ + { + "vehicleKey": "jt808:13079961234", + "rawUri": "archive://2026/06/29/JT808/13079961234/frame-1.bin", + }, + { + "vehicleKey": "jt808:13079961234", + "rawUri": "archive://2026/06/29/JT808/13079961234/frame-2.bin", + }, + { + "vehicleKey": "jt808:13079961234", + "rawUri": "archive://2026/06/29/JT808/13079961234/frame-1.bin", + }, + ]) + + self.assertEqual(len(sqls), 2) + self.assertTrue(sqls[0].endswith("frame-1.bin'")) + self.assertTrue(sqls[1].endswith("frame-2.bin'")) + + def test_expected_history_count_waits_for_pagination_sample(self): + args = types.SimpleNamespace(frames=3, expect_history_count=1, verify_pagination=True) + + self.assertEqual(smoke.expected_history_count(args), 2) + if __name__ == "__main__": unittest.main() diff --git a/tools/test_tdengine_smoke.py b/tools/test_tdengine_smoke.py new file mode 100644 index 00000000..4b89ff9c --- /dev/null +++ b/tools/test_tdengine_smoke.py @@ -0,0 +1,35 @@ +import unittest + +from tools import tdengine_smoke + + +class TdengineSmokeTest(unittest.TestCase): + def test_sql_literal_escapes_single_quotes(self): + self.assertEqual(tdengine_smoke.sql_literal("VIN'001"), "'VIN''001'") + + def test_raw_frame_count_sql_filters_by_protocol_vehicle_and_raw_uri(self): + sql = tdengine_smoke.raw_frame_count_sql( + protocol="JT808", + vehicle_key="jt808:13079962000", + raw_uri="archive://2026/06/29/JT808/frame.bin", + ) + + self.assertEqual( + sql, + "SELECT COUNT(*) FROM raw_frames WHERE protocol = 'JT808' " + "AND vehicle_key = 'jt808:13079962000' " + "AND raw_uri = 'archive://2026/06/29/JT808/frame.bin'", + ) + + def test_raw_frame_count_sql_requires_at_least_one_identity_or_raw_uri_filter(self): + with self.assertRaises(ValueError): + tdengine_smoke.raw_frame_count_sql(protocol="GB32960") + + def test_count_from_response_reads_first_cell(self): + response = {"code": 0, "data": [[3]], "rows": 1} + + self.assertEqual(tdengine_smoke.count_from_response(response), 3) + + +if __name__ == "__main__": + unittest.main()