test: include realtime in production smoke

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,