locationd: speedup unit tests (#29262)

selfdrive/locationd/test/test_locationd.py

@@ -14,7 +14,6 @@ from selfdrive.manager.process_config import managed_processes
 
 
 class TestLocationdProc(unittest.TestCase):
-  MAX_WAITS = 1000
   LLD_MSGS = ['gpsLocationExternal', 'cameraOdometry', 'carState', 'liveCalibration',
               'accelerometer', 'gyroscope', 'magnetometer']
 
@@ -23,30 +22,20 @@ class TestLocationdProc(unittest.TestCase):
 
     self.pm = messaging.PubMaster(self.LLD_MSGS)
 
-    Params().put_bool("UbloxAvailable", True)
+    self.params = Params()
+    self.params.put_bool("UbloxAvailable", True)
     managed_processes['locationd'].prepare()
     managed_processes['locationd'].start()
 
-    time.sleep(1)
-
   def tearDown(self):
     managed_processes['locationd'].stop()
 
-  def send_msg(self, msg):
+  def get_msg(self, name, t):
-    self.pm.send(msg.which(), msg)
-    waits_left = self.MAX_WAITS
-    while waits_left and not self.pm.all_readers_updated(msg.which()):
-      time.sleep(0)
-      waits_left -= 1
-    time.sleep(0.0001)
-
-  def get_fake_msg(self, name, t):
     try:
       msg = messaging.new_message(name)
     except capnp.lib.capnp.KjException:
       msg = messaging.new_message(name, 0)
 
-
     if name == "gpsLocationExternal":
       msg.gpsLocationExternal.flags = 1
       msg.gpsLocationExternal.verticalAccuracy = 1.0
@@ -74,27 +63,27 @@ class TestLocationdProc(unittest.TestCase):
     return msg
 
   def test_params_gps(self):
-    # first reset params
+    self.params.remove('LastGPSPosition')
-    Params().remove('LastGPSPosition')
 
     self.x = -2710700 + (random.random() * 1e5)
     self.y = -4280600 + (random.random() * 1e5)
     self.z = 3850300 + (random.random() * 1e5)
     self.lat, self.lon, self.alt = ecef2geodetic([self.x, self.y, self.z])
 
-    self.fake_duration = 90  # secs
     # get fake messages at the correct frequency, listed in services.py
-    fake_msgs = []
+    msgs = []
-    for sec in range(self.fake_duration):
+    for sec in range(65):
       for name in self.LLD_MSGS:
         for j in range(int(service_list[name].frequency)):
-          fake_msgs.append(self.get_fake_msg(name, int((sec + j / service_list[name].frequency) * 1e9)))
+          msgs.append(self.get_msg(name, int((sec + j / service_list[name].frequency) * 1e9)))
 
-    for fake_msg in sorted(fake_msgs, key=lambda x: x.logMonoTime):
+    for msg in sorted(msgs, key=lambda x: x.logMonoTime):
-      self.send_msg(fake_msg)
+      self.pm.send(msg.which(), msg)
+      if msg.which() == "cameraOdometry":
+        self.pm.wait_for_readers_to_update(msg.which(), 0.1)
     time.sleep(1)  # wait for async params write
 
-    lastGPS = json.loads(Params().get('LastGPSPosition'))
+    lastGPS = json.loads(self.params.get('LastGPSPosition'))
     self.assertAlmostEqual(lastGPS['latitude'], self.lat, places=3)
     self.assertAlmostEqual(lastGPS['longitude'], self.lon, places=3)
     self.assertAlmostEqual(lastGPS['altitude'], self.alt, places=3)