108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
import datetime as dt
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
|
|
from tools import gb32960_e2e_smoke as smoke
|
|
|
|
|
|
class Gb32960E2ESmokeTest(unittest.TestCase):
|
|
def test_parse_frame_info_reads_vin_and_collect_time(self):
|
|
frame = bytes.fromhex(
|
|
"232302fe4c544553543230323630363239303030310100081a061d10060f"
|
|
"02002a"
|
|
)
|
|
|
|
info = smoke.parse_frame_info(frame)
|
|
|
|
self.assertEqual(info.vin, "LTEST202606290001")
|
|
self.assertEqual(info.command, 0x02)
|
|
self.assertEqual(info.event_time, dt.datetime(2026, 6, 29, 16, 6, 15, tzinfo=smoke.SHANGHAI))
|
|
|
|
def test_query_range_surrounds_event_time_with_offset(self):
|
|
date_from, date_to = smoke.query_range(
|
|
dt.datetime(2026, 6, 29, 8, 6, 15, tzinfo=smoke.SHANGHAI),
|
|
minutes=5,
|
|
)
|
|
|
|
self.assertEqual(date_from, "2026-06-29T08:01:15+08:00")
|
|
self.assertEqual(date_to, "2026-06-29T08:11:15+08:00")
|
|
|
|
def test_valid_ack_requires_frame_marker(self):
|
|
self.assertTrue(smoke.valid_ack(b"##\x02\x01"))
|
|
self.assertTrue(smoke.valid_ack(b"$$\x02\x01"))
|
|
self.assertFalse(smoke.valid_ack(b""))
|
|
self.assertFalse(smoke.valid_ack(b"\x7e\x80"))
|
|
|
|
def test_archive_path_maps_archive_uri_to_root(self):
|
|
path = smoke.archive_path(
|
|
pathlib.Path("/tmp/archive"),
|
|
"archive://2026/06/29/GB32960/LTEST202606290001/frame.bin",
|
|
)
|
|
|
|
self.assertEqual(
|
|
path,
|
|
pathlib.Path("/tmp/archive/2026/06/29/GB32960/LTEST202606290001/frame.bin"),
|
|
)
|
|
|
|
def test_parse_field_keys_trims_blank_items(self):
|
|
keys = smoke.parse_field_keys("speed_kmh, total_mileage_km, ,longitude")
|
|
|
|
self.assertEqual(keys, ["speed_kmh", "total_mileage_km", "longitude"])
|
|
|
|
def test_first_raw_uri_reads_record_archive_uri(self):
|
|
uri = smoke.first_raw_uri([
|
|
{"rawArchiveUri": ""},
|
|
{"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST/frame.bin"},
|
|
])
|
|
|
|
self.assertEqual(uri, "archive://2026/06/29/GB32960/LTEST/frame.bin")
|
|
|
|
def test_raw_frame_count_sql_filters_by_vin_and_raw_uri(self):
|
|
info = smoke.FrameInfo("LTEST202606290001", 0x02, None)
|
|
sql = smoke.raw_frame_count_sql(info, [
|
|
{"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST202606290001/frame.bin"},
|
|
])
|
|
|
|
self.assertEqual(
|
|
sql,
|
|
"SELECT COUNT(*) FROM raw_frames WHERE protocol = 'GB32960' "
|
|
"AND vehicle_key = 'LTEST202606290001' "
|
|
"AND vin = 'LTEST202606290001' "
|
|
"AND raw_uri = 'archive://2026/06/29/GB32960/LTEST202606290001/frame.bin'",
|
|
)
|
|
|
|
def test_raw_frame_count_sqls_include_each_unique_record_raw_uri(self):
|
|
info = smoke.FrameInfo("LTEST202606290001", 0x02, None)
|
|
sqls = smoke.raw_frame_count_sqls(info, [
|
|
{"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST202606290001/frame-1.bin"},
|
|
{"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST202606290001/frame-2.bin"},
|
|
{"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST202606290001/frame-1.bin"},
|
|
])
|
|
|
|
self.assertEqual(len(sqls), 2)
|
|
self.assertTrue(sqls[0].endswith("frame-1.bin'"))
|
|
self.assertTrue(sqls[1].endswith("frame-2.bin'"))
|
|
|
|
def test_check_archive_ignores_duplicate_uris(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = pathlib.Path(temp_dir)
|
|
archive_file = root / "2026/06/29/GB32960/LTEST/frame.bin"
|
|
archive_file.parent.mkdir(parents=True)
|
|
archive_file.write_bytes(b"raw")
|
|
|
|
checked = smoke.check_archive(
|
|
root,
|
|
[
|
|
{"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST/frame.bin"},
|
|
{"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST/frame.bin"},
|
|
],
|
|
limit=3,
|
|
)
|
|
|
|
self.assertEqual(checked, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|