299 lines
11 KiB
Python
299 lines
11 KiB
Python
import datetime as dt
|
|
import unittest
|
|
|
|
from tools import go_native_prod_smoke as smoke
|
|
|
|
|
|
class GoNativeProdSmokeTest(unittest.TestCase):
|
|
def test_shanghai_day_window_uses_plus_eight_bounds(self):
|
|
now = dt.datetime(2026, 7, 2, 1, 35, tzinfo=smoke.SHANGHAI)
|
|
|
|
date_from, date_to = smoke.shanghai_day_window(now)
|
|
|
|
self.assertEqual(date_from, "2026-07-02T00:00:00+08:00")
|
|
self.assertEqual(date_to, "2026-07-03T00:00:00+08:00")
|
|
|
|
def test_api_url_encodes_plus_eight_date_range(self):
|
|
url = smoke.api_url(
|
|
"http://example.test/base/",
|
|
"/api/history/raw-frames",
|
|
{
|
|
"protocol": "JT808",
|
|
"dateFrom": "2026-07-02T00:00:00+08:00",
|
|
"dateTo": "2026-07-03T00:00:00+08:00",
|
|
"limit": 1,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
url,
|
|
"http://example.test/base/api/history/raw-frames?"
|
|
"protocol=JT808&dateFrom=2026-07-02T00%3A00%3A00%2B08%3A00"
|
|
"&dateTo=2026-07-03T00%3A00%3A00%2B08%3A00&limit=1",
|
|
)
|
|
|
|
def test_total_check_passes_when_total_reaches_minimum(self):
|
|
check = smoke.check_total(
|
|
"jt808.raw",
|
|
{"total": 2, "items": [{"vehicle_key": "JT808:013307811170"}]},
|
|
minimum=1,
|
|
)
|
|
|
|
self.assertEqual(check.status, "pass")
|
|
self.assertEqual(check.count, 2)
|
|
self.assertIn("JT808:013307811170", check.message)
|
|
|
|
def test_total_check_fails_when_total_is_below_minimum(self):
|
|
check = smoke.check_total("gb32960.raw", {"total": 0, "items": []}, minimum=1)
|
|
|
|
self.assertEqual(check.status, "fail")
|
|
self.assertEqual(check.count, 0)
|
|
self.assertIn("expected >= 1", check.message)
|
|
|
|
def test_overall_status_fails_if_any_check_fails(self):
|
|
checks = [
|
|
smoke.Check("a", "pass", 1, 1, "ok"),
|
|
smoke.Check("b", "fail", 0, 1, "bad"),
|
|
]
|
|
|
|
self.assertEqual(smoke.overall_status(checks), "fail")
|
|
|
|
def test_build_checks_cover_three_raw_protocols_and_two_daily_metric_protocols(self):
|
|
specs = smoke.build_check_specs(
|
|
date_from="2026-07-02T00:00:00+08:00",
|
|
date_to="2026-07-03T00:00:00+08:00",
|
|
stat_date="2026-07-02",
|
|
min_raw=1,
|
|
min_history=1,
|
|
min_stat=1,
|
|
)
|
|
|
|
names = [spec.name for spec in specs]
|
|
|
|
self.assertIn("gb32960.raw", names)
|
|
self.assertIn("jt808.raw", names)
|
|
self.assertIn("yutong_mqtt.raw", names)
|
|
self.assertIn("gb32960.locations", names)
|
|
self.assertIn("gb32960.mileage_points", names)
|
|
self.assertIn("jt808.locations", names)
|
|
self.assertIn("jt808.mileage_points", names)
|
|
self.assertIn("yutong_mqtt.locations", names)
|
|
self.assertIn("yutong_mqtt.mileage_points", names)
|
|
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)
|
|
|
|
def test_realtime_check_fails_when_snapshot_is_stale(self):
|
|
now = dt.datetime(2026, 7, 2, 0, 30, 0, tzinfo=dt.timezone.utc)
|
|
|
|
check = smoke.check_realtime(
|
|
"jt808.realtime",
|
|
{"vehicle_key": "JT808:013307811170", "updated_at_ms": 1782950400000},
|
|
max_age_minutes=15,
|
|
now=now,
|
|
)
|
|
|
|
self.assertEqual(check.status, "fail")
|
|
self.assertIn("updated_age_minutes=30.0", check.message)
|
|
|
|
def test_realtime_check_passes_when_snapshot_is_fresh(self):
|
|
now = dt.datetime(2026, 7, 2, 0, 30, 0, tzinfo=dt.timezone.utc)
|
|
|
|
check = smoke.check_realtime(
|
|
"jt808.realtime",
|
|
{"vehicle_key": "JT808:013307811170", "updated_at_ms": 1782951600000},
|
|
max_age_minutes=15,
|
|
now=now,
|
|
)
|
|
|
|
self.assertEqual(check.status, "pass")
|
|
self.assertIn("updated_age_minutes=10.0", 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)
|
|
|
|
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)
|
|
|
|
def test_history_check_requires_frame_id_backlink(self):
|
|
check = smoke.check_total(
|
|
"jt808.locations",
|
|
{"total": 1, "items": [{"ts": "2026-07-01 17:45:00", "frame_id": ""}]},
|
|
minimum=1,
|
|
require_frame_id=True,
|
|
)
|
|
|
|
self.assertEqual(check.status, "fail")
|
|
self.assertIn("missing frame_id", check.message)
|
|
|
|
def test_history_check_accepts_frame_id_backlink(self):
|
|
check = smoke.check_total(
|
|
"jt808.mileage_points",
|
|
{"total": 1, "items": [{"ts": "2026-07-01 17:45:00", "frame_id": "go_abc"}]},
|
|
minimum=1,
|
|
require_frame_id=True,
|
|
)
|
|
|
|
self.assertEqual(check.status, "pass")
|
|
self.assertIn("frame_id=ok", check.message)
|
|
|
|
def test_daily_mileage_formula_requires_latest_minus_first(self):
|
|
check = smoke.check_total(
|
|
"jt808.daily_mileage",
|
|
{
|
|
"total": 1,
|
|
"items": [{
|
|
"metric_key": "daily_mileage_km",
|
|
"metric_value": 40.0,
|
|
"first_total_mileage_km": 4434.9,
|
|
"latest_total_mileage_km": 4481.0,
|
|
}],
|
|
},
|
|
minimum=1,
|
|
require_metric_formula=True,
|
|
)
|
|
|
|
self.assertEqual(check.status, "fail")
|
|
self.assertIn("metric_formula mismatch", check.message)
|
|
|
|
def test_daily_total_formula_requires_latest_total(self):
|
|
check = smoke.check_total(
|
|
"jt808.daily_total_mileage",
|
|
{
|
|
"total": 1,
|
|
"items": [{
|
|
"metric_key": "daily_total_mileage_km",
|
|
"metric_value": 4481.0,
|
|
"first_total_mileage_km": 4434.9,
|
|
"latest_total_mileage_km": 4481.0,
|
|
}],
|
|
},
|
|
minimum=1,
|
|
require_metric_formula=True,
|
|
)
|
|
|
|
self.assertEqual(check.status, "pass")
|
|
self.assertIn("metric_formula=ok", check.message)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|