From 2214abe9c535d163cfa4bbc06c15ba5c8bfa13e5 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 01:36:14 +0800 Subject: [PATCH] test: include realtime in production smoke --- docs/operations/current-ecs-deployment.md | 3 +- tools/go_native_prod_smoke.py | 84 ++++++++++++++++++++++- tools/test_go_native_prod_smoke.py | 58 ++++++++++++++++ 3 files changed, 141 insertions(+), 4 deletions(-) diff --git a/docs/operations/current-ecs-deployment.md b/docs/operations/current-ecs-deployment.md index 743d16b5..b165cd98 100644 --- a/docs/operations/current-ecs-deployment.md +++ b/docs/operations/current-ecs-deployment.md @@ -142,7 +142,7 @@ curl -sS 'http://115.29.187.205:20210/api/stats/daily-metrics?dateFrom=2026-07-0 python3 tools/go_native_prod_smoke.py --date 2026-07-02 --timeout 8 ``` -该脚本通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询，JT808 的位置/里程点查询，以及 GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计。任一检查不达标时退出码为非 0。 +该脚本通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询，JT808 的位置/里程点查询，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。任一检查不达标时退出码为非 0。 ## 部署命令 @@ -188,3 +188,4 @@ journalctl -u lingniu-go-realtime-api --since '5 minutes ago' --no-pager - `realtime-api` 正在监听 `20210`。 - GB32960、JT808、YUTONG_MQTT 的东八区 RAW 查询均可命中生产数据。 - `vehicle_daily_metric` 可查到 `daily_mileage_km` 和 `daily_total_mileage_km`。 +- Redis realtime 可查到 GB32960、JT808、YUTONG_MQTT 的在线状态和协议快照。 diff --git a/tools/go_native_prod_smoke.py b/tools/go_native_prod_smoke.py index c8e3fba9..e9ba8cab 100755 --- a/tools/go_native_prod_smoke.py +++ b/tools/go_native_prod_smoke.py @@ -23,6 +23,7 @@ class CheckSpec: path: str params: dict[str, Any] minimum: int + kind: str = "total" @dataclass(frozen=True) @@ -43,7 +44,10 @@ def shanghai_day_window(now: dt.datetime | None = None) -> tuple[str, str]: def api_url(base_url: str, path: str, params: dict[str, Any]) -> str: - return base_url.rstrip("/") + path + "?" + urllib.parse.urlencode(params) + url = base_url.rstrip("/") + path + if not params: + return url + return url + "?" + urllib.parse.urlencode(params) def query_json(base_url: str, path: str, params: dict[str, Any], timeout: float) -> dict[str, Any]: @@ -63,6 +67,31 @@ def check_total(name: str, payload: dict[str, Any], minimum: int) -> Check: return Check(name, "fail", count, minimum, f"count={count}; expected >= {minimum}; sample={sample}") +def check_realtime(name: str, payload: dict[str, Any]) -> Check: + if payload.get("online") is False: + return Check(name, "fail", 0, 1, "online=false") + identifier = payload.get("vin") or payload.get("vehicle_key") or "" + protocols = payload.get("protocols") or [] + protocol = payload.get("protocol") or "" + fields = payload.get("fields") or {} + message = json.dumps( + { + key: value + for key, value in { + "identifier": identifier, + "protocol": protocol, + "protocols": protocols, + "field_count": len(fields) if isinstance(fields, dict) else 0, + "online": payload.get("online"), + }.items() + if value not in (None, "", []) + }, + ensure_ascii=False, + sort_keys=True, + ) + return Check(name, "pass", 1, 1, message) + + def sample_summary(items: list[Any]) -> str: if not items: return "{}" @@ -161,17 +190,65 @@ def build_check_specs( ] +def vehicle_identifier(item: dict[str, Any]) -> str: + for key in ("vin", "vehicle_key"): + value = str(item.get(key) or "").strip() + if value and value.lower() != "unknown": + return value + return "" + + +def build_realtime_specs(payloads: dict[str, dict[str, Any]]) -> list[CheckSpec]: + configs = [ + ("gb32960", "gb32960.raw", "GB32960"), + ("jt808", "jt808.raw", "JT808"), + ("yutong_mqtt", "yutong_mqtt.raw", "YUTONG_MQTT"), + ] + specs: list[CheckSpec] = [] + for prefix, source_name, protocol in configs: + items = payloads.get(source_name, {}).get("items") or [] + if not items or not isinstance(items[0], dict): + continue + identifier = vehicle_identifier(items[0]) + if not identifier: + continue + encoded = urllib.parse.quote(identifier, safe="") + base_path = "/api/realtime/vehicles/" + encoded + specs.extend([ + CheckSpec(prefix + ".realtime", base_path, {}, 1, "realtime"), + CheckSpec(prefix + ".realtime_online", base_path + "/online", {}, 1, "realtime"), + CheckSpec(prefix + ".realtime_protocol", base_path + "/protocols/" + protocol, {}, 1, "realtime"), + ]) + return specs + + def run_checks(base_url: str, specs: list[CheckSpec], timeout: float) -> list[Check]: checks: list[Check] = [] for spec in specs: try: payload = query_json(base_url, spec.path, spec.params, timeout) - checks.append(check_total(spec.name, payload, spec.minimum)) + if spec.kind == "realtime": + checks.append(check_realtime(spec.name, payload)) + else: + checks.append(check_total(spec.name, payload, spec.minimum)) 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 +def query_payloads(base_url: str, specs: list[CheckSpec], timeout: float) -> tuple[list[Check], dict[str, dict[str, Any]]]: + checks: list[Check] = [] + payloads: dict[str, dict[str, Any]] = {} + for spec in specs: + try: + payload = query_json(base_url, spec.path, spec.params, timeout) + payloads[spec.name] = payload + checks.append(check_total(spec.name, payload, spec.minimum)) + except Exception as exc: + checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}")) + return checks, payloads + + def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", default=DEFAULT_BASE_URL) @@ -201,7 +278,8 @@ def main(argv: list[str]) -> int: min_history=args.min_history, min_stat=args.min_stat, ) - checks = run_checks(args.base_url, specs, args.timeout) + checks, payloads = query_payloads(args.base_url, specs, args.timeout) + checks.extend(run_checks(args.base_url, build_realtime_specs(payloads), args.timeout)) status = overall_status(checks) print(json.dumps({ "status": status, diff --git a/tools/test_go_native_prod_smoke.py b/tools/test_go_native_prod_smoke.py index 78543fe7..2a207589 100644 --- a/tools/test_go_native_prod_smoke.py +++ b/tools/test_go_native_prod_smoke.py @@ -78,6 +78,64 @@ class GoNativeProdSmokeTest(unittest.TestCase): self.assertIn("gb32960.daily_mileage", names) self.assertIn("jt808.daily_total_mileage", names) + def test_vehicle_identifier_prefers_vin_over_vehicle_key(self): + identifier = smoke.vehicle_identifier({ + "vin": "LB9A32A20R0LS1343", + "vehicle_key": "GB32960:ignored", + }) + + self.assertEqual(identifier, "LB9A32A20R0LS1343") + + def test_vehicle_identifier_falls_back_to_vehicle_key(self): + identifier = smoke.vehicle_identifier({ + "vin": "", + "vehicle_key": "JT808:013307811170", + }) + + self.assertEqual(identifier, "JT808:013307811170") + + def test_realtime_specs_are_built_from_raw_samples_with_vin(self): + payloads = { + "gb32960.raw": {"items": [{"vin": "LB9A32A20R0LS1343", "vehicle_key": "LB9A32A20R0LS1343"}]}, + "jt808.raw": {"items": [{"vin": "", "vehicle_key": "JT808:013307811170"}]}, + "yutong_mqtt.raw": {"items": [{"vin": "LMRKH9AC6R1004108"}]}, + } + + specs = smoke.build_realtime_specs(payloads) + + self.assertEqual( + [(spec.name, spec.path) for spec in specs], + [ + ("gb32960.realtime", "/api/realtime/vehicles/LB9A32A20R0LS1343"), + ("gb32960.realtime_online", "/api/realtime/vehicles/LB9A32A20R0LS1343/online"), + ("gb32960.realtime_protocol", "/api/realtime/vehicles/LB9A32A20R0LS1343/protocols/GB32960"), + ("jt808.realtime", "/api/realtime/vehicles/JT808%3A013307811170"), + ("jt808.realtime_online", "/api/realtime/vehicles/JT808%3A013307811170/online"), + ("jt808.realtime_protocol", "/api/realtime/vehicles/JT808%3A013307811170/protocols/JT808"), + ("yutong_mqtt.realtime", "/api/realtime/vehicles/LMRKH9AC6R1004108"), + ("yutong_mqtt.realtime_online", "/api/realtime/vehicles/LMRKH9AC6R1004108/online"), + ("yutong_mqtt.realtime_protocol", "/api/realtime/vehicles/LMRKH9AC6R1004108/protocols/YUTONG_MQTT"), + ], + ) + + def test_realtime_check_requires_online_true_when_field_exists(self): + check = smoke.check_realtime( + "gb32960.realtime_online", + {"vin": "LB9A32A20R0LS1343", "online": True, "protocols": ["GB32960"]}, + ) + + self.assertEqual(check.status, "pass") + self.assertEqual(check.count, 1) + + def test_realtime_check_fails_when_online_false(self): + check = smoke.check_realtime( + "gb32960.realtime_online", + {"vin": "LB9A32A20R0LS1343", "online": False}, + ) + + self.assertEqual(check.status, "fail") + self.assertIn("online=false", check.message) + if __name__ == "__main__": unittest.main()