69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
import datetime
|
|
import json
|
|
import pickle
|
|
import unittest
|
|
|
|
import geopy
|
|
from dse_shared_libs.daf import DAF
|
|
from dse_shared_libs.target_velocity import TargetVelocity
|
|
from dse_shared_libs.traffic_light_color import TrafficLightColor
|
|
from dse_shared_libs.traffic_light_state import TrafficLightState
|
|
from redis import StrictRedis
|
|
|
|
from event_logger import EventLogger
|
|
|
|
|
|
class TestEventLogger(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.el = EventLogger(StrictRedis(), False, False)
|
|
self.timestamp = datetime.datetime.utcnow()
|
|
|
|
def test_unpack_daf(self):
|
|
daf = DAF(vehicle_identification_number='my_vin',
|
|
gps_location=geopy.Point(0, 0, 0),
|
|
velocity=130.0,
|
|
near_crash_event=False,
|
|
timestamp=self.timestamp,
|
|
)
|
|
|
|
key, message = self.el._unpack_message_to_log(pickle.dumps(daf))
|
|
|
|
self.assertEqual(key, 'DAF:my_vin')
|
|
self.assertEqual(message, json.dumps(daf.to_dict()))
|
|
|
|
def test_unpack_target_velocity(self):
|
|
tv = TargetVelocity(vin='my_other_vin',
|
|
target_velocity=120.0,
|
|
timestamp=self.timestamp)
|
|
|
|
key, message = self.el._unpack_message_to_log(pickle.dumps(tv))
|
|
|
|
self.assertEqual(key, 'TV:my_other_vin')
|
|
self.assertEqual(message, json.dumps(tv.to_dict()))
|
|
|
|
def test_unpack_traffic_light_state(self):
|
|
tls = TrafficLightState(tlid='my_traffic_light',
|
|
color=TrafficLightColor.GREEN,
|
|
last_switch=self.timestamp)
|
|
|
|
key, message = self.el._unpack_message_to_log(pickle.dumps(tls))
|
|
|
|
self.assertEqual(key, 'TL:my_traffic_light')
|
|
self.assertEqual(message, json.dumps(tls.to_dict()))
|
|
|
|
def test_unpack_unknown_dict(self):
|
|
unknown = dict(foo='foo', bar='bar')
|
|
|
|
key, message = self.el._unpack_message_to_log(pickle.dumps(unknown))
|
|
|
|
self.assertEqual(key, 'UNKNOWN')
|
|
self.assertEqual(message, json.dumps(unknown))
|
|
|
|
def test_unpack_unknown_object(self):
|
|
obj = datetime.datetime.utcnow()
|
|
|
|
key, message = self.el._unpack_message_to_log(pickle.dumps(obj))
|
|
|
|
self.assertEqual(key, 'UNKNOWN')
|
|
self.assertEqual(message, str(obj))
|