From 27f4ad346821e972b062acc0294490e6548bf885 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 02:00:50 +0800 Subject: [PATCH] test: verify raw parsed json in production smoke --- docs/operations/current-ecs-deployment.md | 4 +- .../go-vehicle-gateway-verification.md | 2 +- tools/go_native_prod_smoke.py | 48 +++++++++++++++++-- tools/test_go_native_prod_smoke.py | 34 +++++++++++++ 4 files changed, 82 insertions(+), 6 deletions(-) diff --git a/docs/operations/current-ecs-deployment.md b/docs/operations/current-ecs-deployment.md index da4b724b..e4aba876 100644 --- a/docs/operations/current-ecs-deployment.md +++ b/docs/operations/current-ecs-deployment.md @@ -169,7 +169,7 @@ python3 tools/go_native_prod_smoke.py --date 2026-07-02 --timeout 8 `go_systemd_prod_smoke.py` 通过 SSH 检查四个 Go systemd 服务是否 `active` 且 `enabled`，并确认 `808`、`32960` 由 `gateway` 监听、`20210` 由 `realtime-api` 监听，同时要求 `/opt/lingniu-go-native/spool/gateway` 没有待回放文件，避免旧 Java/Docker 进程占用生产端口或 Kafka spool 静默堆积。运行环境需要已配置 SSH 免密或已建立可用的 SSH 认证方式。 -`go_native_prod_smoke.py` 通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询，三类协议的位置和里程点查询，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。默认检查今天东八区数据时，还会要求三类 RAW 最新样本不超过 15 分钟，避免旧数据误判为接收正常；任一检查不达标时退出码为非 0。 +`go_native_prod_smoke.py` 通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询及结构化 `parsed_json`，三类协议的位置和里程点查询，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。默认检查今天东八区数据时，还会要求三类 RAW 最新样本不超过 15 分钟，避免旧数据误判为接收正常；任一检查不达标时退出码为非 0。 ## 部署命令 @@ -216,7 +216,7 @@ journalctl -u lingniu-go-realtime-api --since '5 minutes ago' --no-pager - `gateway` 正在监听 `32960` 和 `808`。 - `realtime-api` 正在监听 `20210`。 - gateway Kafka spool 当前无待回放文件。 -- GB32960、JT808、YUTONG_MQTT 的东八区 RAW 查询均可命中生产数据。 +- GB32960、JT808、YUTONG_MQTT 的东八区 RAW 查询均可命中生产数据，且最新 RAW 样本包含结构化 `parsed_json`。 - GB32960、JT808、YUTONG_MQTT 的最新 RAW 样本均在 15 分钟内。 - GB32960、JT808、YUTONG_MQTT 的 `vehicle_locations`、`vehicle_mileage_points` 查询均可命中生产数据。 - `vehicle_daily_metric` 可查到 `daily_mileage_km` 和 `daily_total_mileage_km`。 diff --git a/docs/operations/go-vehicle-gateway-verification.md b/docs/operations/go-vehicle-gateway-verification.md index 7212f5f8..5cb3998f 100644 --- a/docs/operations/go-vehicle-gateway-verification.md +++ b/docs/operations/go-vehicle-gateway-verification.md @@ -28,7 +28,7 @@ python3 tools/go_prod_acceptance.py --date 2026-07-02 - `go_systemd_prod_smoke.py`：应用 ECS systemd 服务 active/enabled、端口归属和 gateway spool - `go_kafka_prod_smoke.py`：Kafka topic 和消费组 lag -- `go_native_prod_smoke.py`：HTTP API、TDengine/MySQL/Redis 数据链路 +- `go_native_prod_smoke.py`：HTTP API、RAW `parsed_json`、TDengine/MySQL/Redis 数据链路 如需单独检查 systemd： diff --git a/tools/go_native_prod_smoke.py b/tools/go_native_prod_smoke.py index f53525e0..eeaeb572 100755 --- a/tools/go_native_prod_smoke.py +++ b/tools/go_native_prod_smoke.py @@ -25,6 +25,7 @@ class CheckSpec: minimum: int kind: str = "total" max_age_minutes: float | None = None + require_parsed_json: bool = False @dataclass(frozen=True) @@ -66,12 +67,14 @@ def check_total( minimum: int, max_age_minutes: float | None = None, now: dt.datetime | None = None, + require_parsed_json: bool = False, ) -> Check: count = int(payload.get("total") or 0) items = payload.get("items") or [] sample = sample_summary(items) age_message = latest_age_message(items, now) if count >= minimum: + parsed_message = "" if max_age_minutes is not None: latest_age = latest_age_minutes(items, now) if latest_age is None: @@ -85,7 +88,14 @@ def check_total( f"count={count}; latest_age_minutes={latest_age:.1f}; " f"expected <= {max_age_minutes:.1f}; sample={sample}", ) - return Check(name, "pass", count, minimum, f"count={count}; {age_message}; sample={sample}") + if require_parsed_json: + parsed_message = parsed_json_message(items) + if not parsed_message.startswith("parsed_json=ok"): + return Check(name, "fail", count, minimum, f"count={count}; {parsed_message}; sample={sample}") + details = [f"count={count}", age_message] + if parsed_message: + details.append(parsed_message) + return Check(name, "pass", count, minimum, "; ".join(details) + f"; sample={sample}") return Check(name, "fail", count, minimum, f"count={count}; expected >= {minimum}; {age_message}; sample={sample}") @@ -116,6 +126,23 @@ def latest_age_message(items: list[Any], now: dt.datetime | None = None) -> str: return f"latest_age_minutes={latest_age:.1f}" +def parsed_json_message(items: list[Any]) -> str: + if not items or not isinstance(items[0], dict): + return "missing parsed_json" + raw = items[0].get("parsed_json") + if isinstance(raw, dict) and raw: + return "parsed_json=ok" + if not isinstance(raw, str) or not raw.strip() or raw.strip().lower() == "null": + return "missing parsed_json" + try: + parsed = json.loads(raw) + except json.JSONDecodeError: + return "invalid parsed_json" + if isinstance(parsed, dict) and parsed: + return "parsed_json=ok" + return "invalid parsed_json" + + def check_realtime(name: str, payload: dict[str, Any]) -> Check: if payload.get("online") is False: return Check(name, "fail", 0, 1, "online=false") @@ -192,6 +219,7 @@ def build_check_specs( {**raw_params, "protocol": "GB32960"}, min_raw, max_age_minutes=max_raw_age_minutes, + require_parsed_json=True, ), CheckSpec( "jt808.raw", @@ -199,6 +227,7 @@ def build_check_specs( {**raw_params, "protocol": "JT808"}, min_raw, max_age_minutes=max_raw_age_minutes, + require_parsed_json=True, ), CheckSpec( "yutong_mqtt.raw", @@ -206,6 +235,7 @@ def build_check_specs( {**raw_params, "protocol": "YUTONG_MQTT"}, min_raw, max_age_minutes=max_raw_age_minutes, + require_parsed_json=True, ), CheckSpec("gb32960.locations", "/api/history/locations", gb32960_history_params, min_history), CheckSpec("gb32960.mileage_points", "/api/history/mileage-points", gb32960_history_params, min_history), @@ -304,7 +334,13 @@ def run_checks(base_url: str, specs: list[CheckSpec], timeout: float) -> list[Ch if spec.kind == "realtime": checks.append(check_realtime(spec.name, payload)) else: - checks.append(check_total(spec.name, payload, spec.minimum, spec.max_age_minutes)) + checks.append(check_total( + spec.name, + payload, + spec.minimum, + spec.max_age_minutes, + require_parsed_json=spec.require_parsed_json, + )) except Exception as exc: # pragma: no cover - exercised by live failures. checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}")) return checks @@ -317,7 +353,13 @@ def query_payloads(base_url: str, specs: list[CheckSpec], timeout: float) -> tup try: payload = query_json(base_url, spec.path, spec.params, timeout) payloads[spec.name] = payload - checks.append(check_total(spec.name, payload, spec.minimum, spec.max_age_minutes)) + checks.append(check_total( + spec.name, + payload, + spec.minimum, + spec.max_age_minutes, + require_parsed_json=spec.require_parsed_json, + )) except Exception as exc: checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}")) return checks, payloads diff --git a/tools/test_go_native_prod_smoke.py b/tools/test_go_native_prod_smoke.py index cefb09d2..a0afcee3 100644 --- a/tools/test_go_native_prod_smoke.py +++ b/tools/test_go_native_prod_smoke.py @@ -173,6 +173,40 @@ class GoNativeProdSmokeTest(unittest.TestCase): self.assertEqual(check.status, "pass") self.assertIn("latest_age_minutes=15.0", check.message) + def test_raw_check_requires_structured_parsed_json(self): + check = smoke.check_total( + "jt808.raw", + { + "total": 1, + "items": [{ + "ts": "2026-07-01 17:45:00", + "parsed_json": "", + }], + }, + minimum=1, + require_parsed_json=True, + ) + + self.assertEqual(check.status, "fail") + self.assertIn("missing parsed_json", check.message) + + def test_raw_check_accepts_structured_parsed_json(self): + check = smoke.check_total( + "gb32960.raw", + { + "total": 1, + "items": [{ + "ts": "2026-07-01 17:45:00", + "parsed_json": "{\"header\":{\"vin\":\"LTEST\"}}", + }], + }, + minimum=1, + require_parsed_json=True, + ) + + self.assertEqual(check.status, "pass") + self.assertIn("parsed_json=ok", check.message) + if __name__ == "__main__": unittest.main()