135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
import pathlib
|
|
import types
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from tools import jt808_e2e_smoke as smoke
|
|
|
|
|
|
class Jt808E2ESmokeTest(unittest.TestCase):
|
|
def test_build_frame_adds_boundaries_and_valid_checksum(self):
|
|
body = smoke.build_location_body(smoke.parse_event_time("2024-01-02T03:04:05"))
|
|
|
|
frame = smoke.build_frame(0x0200, "13079961234", 1, body)
|
|
|
|
self.assertEqual(frame[0], 0x7E)
|
|
self.assertEqual(frame[-1], 0x7E)
|
|
unescaped = smoke.unescape_payload(frame[1:-1])
|
|
self.assertEqual(smoke.bcc(unescaped), 0)
|
|
|
|
def test_next_page_params_include_cursor(self):
|
|
params = smoke.next_page_params(
|
|
{"phone": "13079961234", "limit": 1},
|
|
{"nextCursor": {"ts": "2024-01-01T19:04:05Z", "id": "frame-1"}},
|
|
)
|
|
|
|
self.assertEqual(params["cursorTs"], "2024-01-01T19:04:05Z")
|
|
self.assertEqual(params["cursorId"], "frame-1")
|
|
self.assertEqual(params["phone"], "13079961234")
|
|
|
|
def test_archive_path_maps_archive_uri_to_root(self):
|
|
path = smoke.archive_path(
|
|
pathlib.Path("/tmp/archive"),
|
|
"archive://2026/06/29/JT808/unknown/frame.bin",
|
|
)
|
|
|
|
self.assertEqual(path, pathlib.Path("/tmp/archive/2026/06/29/JT808/unknown/frame.bin"))
|
|
|
|
def test_raw_frame_count_sql_filters_by_vehicle_key_phone_and_raw_uri(self):
|
|
sql = smoke.raw_frame_count_sql("13079961234", [
|
|
{
|
|
"vehicleKey": "jt808:13079961234",
|
|
"rawUri": "archive://2026/06/29/JT808/13079961234/frame.bin",
|
|
},
|
|
])
|
|
|
|
self.assertEqual(
|
|
sql,
|
|
"SELECT COUNT(*) FROM raw_frames WHERE protocol = 'JT808' "
|
|
"AND vehicle_key = 'jt808:13079961234' "
|
|
"AND phone = '13079961234' "
|
|
"AND raw_uri = 'archive://2026/06/29/JT808/13079961234/frame.bin'",
|
|
)
|
|
|
|
def test_raw_frame_count_sqls_include_each_unique_visible_raw_uri(self):
|
|
sqls = smoke.raw_frame_count_sqls("13079961234", [
|
|
{
|
|
"vehicleKey": "jt808:13079961234",
|
|
"rawUri": "archive://2026/06/29/JT808/13079961234/frame-1.bin",
|
|
},
|
|
{
|
|
"vehicleKey": "jt808:13079961234",
|
|
"rawUri": "archive://2026/06/29/JT808/13079961234/frame-2.bin",
|
|
},
|
|
{
|
|
"vehicleKey": "jt808:13079961234",
|
|
"rawUri": "archive://2026/06/29/JT808/13079961234/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_verify_tdengine_raw_frames_allows_probe_without_visible_raw_uri(self):
|
|
verified = smoke.verify_tdengine_raw_frames(
|
|
"http://localhost:6041/rest/sql/vehicle_ts",
|
|
"root",
|
|
"taosdata",
|
|
[],
|
|
0,
|
|
1,
|
|
)
|
|
|
|
self.assertEqual(verified, 0)
|
|
|
|
def test_expected_history_count_waits_for_pagination_sample(self):
|
|
args = types.SimpleNamespace(frames=3, expect_history_count=1, verify_pagination=True)
|
|
|
|
self.assertEqual(smoke.expected_history_count(args), 2)
|
|
|
|
def test_send_phone_workload_can_run_session_phones_with_workers(self):
|
|
event_time = smoke.parse_event_time("2024-01-02T03:04:05")
|
|
args = types.SimpleNamespace(
|
|
connection_mode="session",
|
|
tcp_host="127.0.0.1",
|
|
tcp_port=808,
|
|
frames=3,
|
|
tcp_timeout=1,
|
|
rate=0,
|
|
linger_ms=0,
|
|
post_register_ms=0,
|
|
require_register_ack=True,
|
|
workers=2,
|
|
)
|
|
calls = []
|
|
|
|
def fake_send_phone_sequence(host, port, phone, phone_index, frames, sent_event_time,
|
|
timeout, rate, linger_ms, post_register_ms):
|
|
calls.append((host, port, phone, phone_index, frames, sent_event_time,
|
|
timeout, rate, linger_ms, post_register_ms))
|
|
return b"ack", frames
|
|
|
|
with mock.patch.object(smoke, "send_phone_sequence", side_effect=fake_send_phone_sequence):
|
|
result = smoke.send_phone_workload(
|
|
args,
|
|
["13079962000", "13079962001", "13079962002"],
|
|
event_time,
|
|
)
|
|
|
|
self.assertEqual(result.sent_registers, 3)
|
|
self.assertEqual(result.sent_locations, 9)
|
|
self.assertGreater(result.elapsed_seconds, 0)
|
|
self.assertEqual(
|
|
sorted((call[2], call[3], call[4]) for call in calls),
|
|
[
|
|
("13079962000", 0, 3),
|
|
("13079962001", 1, 3),
|
|
("13079962002", 2, 3),
|
|
],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|