From 9bd48c57814bf1a1ed67d20affd44161d9c6529b Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 01:38:16 +0800 Subject: [PATCH] test: require fresh raw data in production smoke --- docs/operations/current-ecs-deployment.md | 3 +- tools/go_native_prod_smoke.py | 95 ++++++++++++++++++++--- tools/test_go_native_prod_smoke.py | 33 ++++++++ 3 files changed, 121 insertions(+), 10 deletions(-) diff --git a/docs/operations/current-ecs-deployment.md b/docs/operations/current-ecs-deployment.md index b165cd98..b476b1a9 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` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。任一检查不达标时退出码为非 0。 +该脚本通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询，JT808 的位置/里程点查询，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。默认检查今天东八区数据时，还会要求三类 RAW 最新样本不超过 15 分钟，避免旧数据误判为接收正常；任一检查不达标时退出码为非 0。 ## 部署命令 @@ -187,5 +187,6 @@ journalctl -u lingniu-go-realtime-api --since '5 minutes ago' --no-pager - `gateway` 正在监听 `32960` 和 `808`。 - `realtime-api` 正在监听 `20210`。 - GB32960、JT808、YUTONG_MQTT 的东八区 RAW 查询均可命中生产数据。 +- GB32960、JT808、YUTONG_MQTT 的最新 RAW 样本均在 15 分钟内。 - `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 e9ba8cab..b3c3fa01 100755 --- a/tools/go_native_prod_smoke.py +++ b/tools/go_native_prod_smoke.py @@ -24,6 +24,7 @@ class CheckSpec: params: dict[str, Any] minimum: int kind: str = "total" + max_age_minutes: float | None = None @dataclass(frozen=True) @@ -59,12 +60,60 @@ def query_json(base_url: str, path: str, params: dict[str, Any], timeout: float) return json.loads(response.read().decode("utf-8")) -def check_total(name: str, payload: dict[str, Any], minimum: int) -> Check: +def check_total( + name: str, + payload: dict[str, Any], + minimum: int, + max_age_minutes: float | None = None, + now: dt.datetime | None = None, +) -> Check: count = int(payload.get("total") or 0) - sample = sample_summary(payload.get("items") or []) + items = payload.get("items") or [] + sample = sample_summary(items) + age_message = latest_age_message(items, now) if count >= minimum: - return Check(name, "pass", count, minimum, f"count={count}; sample={sample}") - return Check(name, "fail", count, minimum, f"count={count}; expected >= {minimum}; sample={sample}") + if max_age_minutes is not None: + latest_age = latest_age_minutes(items, now) + if latest_age is None: + return Check(name, "fail", count, minimum, f"count={count}; missing latest ts; sample={sample}") + if latest_age > max_age_minutes: + return Check( + name, + "fail", + count, + minimum, + 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}") + return Check(name, "fail", count, minimum, f"count={count}; expected >= {minimum}; {age_message}; sample={sample}") + + +def parse_tdengine_utc_timestamp(value: str) -> dt.datetime: + value = str(value or "").strip() + for layout in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): + try: + return dt.datetime.strptime(value[:19], layout).replace(tzinfo=dt.timezone.utc) + except ValueError: + continue + raise ValueError(f"unsupported timestamp: {value}") + + +def latest_age_minutes(items: list[Any], now: dt.datetime | None = None) -> float | None: + if not items or not isinstance(items[0], dict) or not items[0].get("ts"): + return None + current = now or dt.datetime.now(tz=dt.timezone.utc) + if current.tzinfo is None: + current = current.replace(tzinfo=dt.timezone.utc) + latest = parse_tdengine_utc_timestamp(str(items[0]["ts"])) + return max(0.0, (current.astimezone(dt.timezone.utc) - latest).total_seconds() / 60) + + +def latest_age_message(items: list[Any], now: dt.datetime | None = None) -> str: + latest_age = latest_age_minutes(items, now) + if latest_age is None: + return "latest_age_minutes=unknown" + return f"latest_age_minutes={latest_age:.1f}" def check_realtime(name: str, payload: dict[str, Any]) -> Check: @@ -130,13 +179,32 @@ def build_check_specs( min_raw: int, min_history: int, min_stat: int, + max_raw_age_minutes: float | None = None, ) -> list[CheckSpec]: raw_params = {"dateFrom": date_from, "dateTo": date_to, "limit": 1} history_params = {"protocol": "JT808", "dateFrom": date_from, "dateTo": date_to, "limit": 1} return [ - CheckSpec("gb32960.raw", "/api/history/raw-frames", {**raw_params, "protocol": "GB32960"}, min_raw), - CheckSpec("jt808.raw", "/api/history/raw-frames", {**raw_params, "protocol": "JT808"}, min_raw), - CheckSpec("yutong_mqtt.raw", "/api/history/raw-frames", {**raw_params, "protocol": "YUTONG_MQTT"}, min_raw), + CheckSpec( + "gb32960.raw", + "/api/history/raw-frames", + {**raw_params, "protocol": "GB32960"}, + min_raw, + max_age_minutes=max_raw_age_minutes, + ), + CheckSpec( + "jt808.raw", + "/api/history/raw-frames", + {**raw_params, "protocol": "JT808"}, + min_raw, + max_age_minutes=max_raw_age_minutes, + ), + CheckSpec( + "yutong_mqtt.raw", + "/api/history/raw-frames", + {**raw_params, "protocol": "YUTONG_MQTT"}, + min_raw, + max_age_minutes=max_raw_age_minutes, + ), CheckSpec("jt808.locations", "/api/history/locations", history_params, min_history), CheckSpec("jt808.mileage_points", "/api/history/mileage-points", history_params, min_history), CheckSpec( @@ -230,7 +298,7 @@ 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)) + checks.append(check_total(spec.name, payload, spec.minimum, spec.max_age_minutes)) 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 @@ -243,7 +311,7 @@ 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)) + checks.append(check_total(spec.name, payload, spec.minimum, spec.max_age_minutes)) except Exception as exc: checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}")) return checks, payloads @@ -257,6 +325,12 @@ def parse_args(argv: list[str]) -> argparse.Namespace: parser.add_argument("--min-raw", type=int, default=1) parser.add_argument("--min-history", type=int, default=1) parser.add_argument("--min-stat", type=int, default=1) + parser.add_argument( + "--max-raw-age-minutes", + type=float, + default=15.0, + help="Require latest RAW sample age to be at most this many minutes when checking today's date.", + ) return parser.parse_args(argv) @@ -267,9 +341,11 @@ def main(argv: list[str]) -> int: start = dt.datetime(day.year, day.month, day.day, tzinfo=SHANGHAI) date_from, date_to = shanghai_day_window(start) stat_date = args.date + max_raw_age_minutes = None else: date_from, date_to = shanghai_day_window() stat_date = date_from[:10] + max_raw_age_minutes = args.max_raw_age_minutes specs = build_check_specs( date_from=date_from, date_to=date_to, @@ -277,6 +353,7 @@ def main(argv: list[str]) -> int: min_raw=args.min_raw, min_history=args.min_history, min_stat=args.min_stat, + max_raw_age_minutes=max_raw_age_minutes, ) checks, payloads = query_payloads(args.base_url, specs, args.timeout) checks.extend(run_checks(args.base_url, build_realtime_specs(payloads), args.timeout)) diff --git a/tools/test_go_native_prod_smoke.py b/tools/test_go_native_prod_smoke.py index 2a207589..1e71eca2 100644 --- a/tools/test_go_native_prod_smoke.py +++ b/tools/test_go_native_prod_smoke.py @@ -136,6 +136,39 @@ class GoNativeProdSmokeTest(unittest.TestCase): self.assertEqual(check.status, "fail") self.assertIn("online=false", check.message) + def test_parse_tdengine_utc_timestamp_as_aware_utc(self): + parsed = smoke.parse_tdengine_utc_timestamp("2026-07-01 17:36:02") + + self.assertEqual(parsed, dt.datetime(2026, 7, 1, 17, 36, 2, tzinfo=dt.timezone.utc)) + + def test_total_check_fails_when_latest_sample_is_stale(self): + now = dt.datetime(2026, 7, 1, 18, 0, 0, tzinfo=dt.timezone.utc) + + check = smoke.check_total( + "gb32960.raw", + {"total": 1, "items": [{"ts": "2026-07-01 17:00:00"}]}, + minimum=1, + max_age_minutes=30, + now=now, + ) + + self.assertEqual(check.status, "fail") + self.assertIn("latest_age_minutes=60.0", check.message) + + def test_total_check_passes_when_latest_sample_is_fresh(self): + now = dt.datetime(2026, 7, 1, 18, 0, 0, tzinfo=dt.timezone.utc) + + check = smoke.check_total( + "gb32960.raw", + {"total": 1, "items": [{"ts": "2026-07-01 17:45:00"}]}, + minimum=1, + max_age_minutes=30, + now=now, + ) + + self.assertEqual(check.status, "pass") + self.assertIn("latest_age_minutes=15.0", check.message) + if __name__ == "__main__": unittest.main()