test: verify daily metric formulas

tools/go_native_prod_smoke.py

@@ -27,6 +27,7 @@ class CheckSpec:
     max_age_minutes: float | None = None
     require_parsed_json: bool = False
     require_frame_id: bool = False
+    require_metric_formula: bool = False
 
 
 @dataclass(frozen=True)
@@ -70,6 +71,7 @@ def check_total(
     now: dt.datetime | None = None,
     require_parsed_json: bool = False,
     require_frame_id: bool = False,
+    require_metric_formula: bool = False,
 ) -> Check:
     count = int(payload.get("total") or 0)
     items = payload.get("items") or []
@@ -99,11 +101,18 @@ def check_total(
             frame_message = frame_id_message(items)
             if not frame_message.startswith("frame_id=ok"):
                 return Check(name, "fail", count, minimum, f"count={count}; {frame_message}; sample={sample}")
+        metric_message = ""
+        if require_metric_formula:
+            metric_message = metric_formula_message(items)
+            if not metric_message.startswith("metric_formula=ok"):
+                return Check(name, "fail", count, minimum, f"count={count}; {metric_message}; sample={sample}")
         details = [f"count={count}", age_message]
         if parsed_message:
             details.append(parsed_message)
         if frame_message:
             details.append(frame_message)
+        if metric_message:
+            details.append(metric_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}")
 
@@ -161,6 +170,28 @@ def frame_id_message(items: list[Any]) -> str:
     return "missing frame_id"
 
 
+def metric_formula_message(items: list[Any]) -> str:
+    if not items or not isinstance(items[0], dict):
+        return "missing metric row"
+    row = items[0]
+    try:
+        metric_value = float(row.get("metric_value"))
+        first_total = float(row.get("first_total_mileage_km"))
+        latest_total = float(row.get("latest_total_mileage_km"))
+    except (TypeError, ValueError):
+        return "missing metric formula fields"
+    metric_key = str(row.get("metric_key") or "")
+    if metric_key == "daily_mileage_km":
+        expected = max(0.0, latest_total - first_total)
+    elif metric_key == "daily_total_mileage_km":
+        expected = latest_total
+    else:
+        return "unsupported metric_key=" + metric_key
+    if abs(metric_value - expected) <= 0.001:
+        return "metric_formula=ok"
+    return f"metric_formula mismatch value={metric_value:.3f} expected={expected:.3f}"
+
+
 def check_realtime(name: str, payload: dict[str, Any]) -> Check:
     if payload.get("online") is False:
         return Check(name, "fail", 0, 1, "online=false")
@@ -272,6 +303,7 @@ def build_check_specs(
                 "limit": 1,
             },
             min_stat,
+            require_metric_formula=True,
         ),
         CheckSpec(
             "gb32960.daily_total_mileage",
@@ -284,6 +316,7 @@ def build_check_specs(
                 "limit": 1,
             },
             min_stat,
+            require_metric_formula=True,
         ),
         CheckSpec(
             "jt808.daily_mileage",
@@ -296,6 +329,7 @@ def build_check_specs(
                 "limit": 1,
             },
             min_stat,
+            require_metric_formula=True,
         ),
         CheckSpec(
             "jt808.daily_total_mileage",
@@ -308,6 +342,7 @@ def build_check_specs(
                 "limit": 1,
             },
             min_stat,
+            require_metric_formula=True,
         ),
     ]
 
@@ -359,6 +394,7 @@ def run_checks(base_url: str, specs: list[CheckSpec], timeout: float) -> list[Ch
                     spec.max_age_minutes,
                     require_parsed_json=spec.require_parsed_json,
                     require_frame_id=spec.require_frame_id,
+                    require_metric_formula=spec.require_metric_formula,
                 ))
         except Exception as exc:  # pragma: no cover - exercised by live failures.
             checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}"))
@@ -379,6 +415,7 @@ def query_payloads(base_url: str, specs: list[CheckSpec], timeout: float) -> tup
                 spec.max_age_minutes,
                 require_parsed_json=spec.require_parsed_json,
                 require_frame_id=spec.require_frame_id,
+                require_metric_formula=spec.require_metric_formula,
             ))
         except Exception as exc:
             checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}"))