loggerd: another rotation test (#19742)

* another rotation test

* make encoder test fast

* fix pc

Co-authored-by: Comma Device <device@comma.ai>

selfdrive/loggerd/tests/test_encoder.py

@@ -4,6 +4,7 @@ import os
 import random
 import shutil
 import subprocess
+import threading
 import time
 import unittest
 from parameterized import parameterized
@@ -18,6 +19,7 @@ from selfdrive.loggerd.config import ROOT, CAMERA_FPS
 
 
 # baseline file sizes for a 2s segment, in bytes
+SEGMENT_LENGTH = 2
 FULL_SIZE = 1253786
 if EON:
   CAMERAS = {
@@ -25,14 +27,12 @@ if EON:
     "dcamera": 770920,
     "qcamera": 38533,
   }
-elif TICI:
-  CAMERAS = {f"{c}camera": FULL_SIZE if c!="q" else 38533 for c in ["f", "e", "d", "q"]}
 else:
-  CAMERAS = {}
+  CAMERAS = {f"{c}camera": FULL_SIZE if c!="q" else 38533 for c in ["f", "e", "d", "q"]}
 
 ALL_CAMERA_COMBINATIONS = [(cameras,) for cameras in [CAMERAS, {k:CAMERAS[k] for k in CAMERAS if k!='dcamera'}]]
 
-FRAME_TOLERANCE = 0
+# we check frame count, so we don't have to be too strict on size
 FILE_SIZE_TOLERANCE = 0.5
 
 class TestEncoder(unittest.TestCase):
@@ -45,10 +45,8 @@ class TestEncoder(unittest.TestCase):
 
   def setUp(self):
     self._clear_logs()
-
-    self.segment_length = 2
     os.environ["LOGGERD_TEST"] = "1"
-    os.environ["LOGGERD_SEGMENT_LENGTH"] = str(self.segment_length)
+    os.environ["LOGGERD_SEGMENT_LENGTH"] = str(SEGMENT_LENGTH)
 
   def tearDown(self):
     self._clear_logs()
@@ -63,7 +61,7 @@ class TestEncoder(unittest.TestCase):
 
   # TODO: this should run faster than real time
   @parameterized.expand(ALL_CAMERA_COMBINATIONS)
-  @with_processes(['camerad', 'sensord', 'loggerd'], init_time=5)
+  @with_processes(['camerad', 'sensord', 'loggerd'])
   def test_log_rotation(self, cameras):
     print("checking targets:", cameras)
     Params().put("RecordFront", "1" if 'dcamera' in cameras else "0")
@@ -73,28 +71,16 @@ class TestEncoder(unittest.TestCase):
       num_segments = random.randint(15, 20) # ffprobe is slow on comma two
 
     # wait for loggerd to make the dir for first segment
-    time.sleep(10)
     route_prefix_path = None
-    with Timeout(30):
+    with Timeout(int(SEGMENT_LENGTH*2)):
       while route_prefix_path is None:
         try:
           route_prefix_path = self._get_latest_segment_path().rsplit("--", 1)[0]
         except Exception:
-          time.sleep(2)
-          continue
-
-    for i in trange(num_segments):
-      # poll for next segment
-      if i < num_segments - 1:
-        with Timeout(self.segment_length*3, error_msg=f"timed out waiting for segment {i}"):
-          while True:
-            seg_num = int(self._get_latest_segment_path().rsplit("--", 1)[1])
-            if seg_num > i:
-              break
           time.sleep(0.1)
-      else:
-        time.sleep(self.segment_length)
+          continue
 
+    def check_seg(i):
       # check each camera file size
       for camera, size in cameras.items():
         ext = "ts" if camera=='qcamera' else "hevc"
@@ -109,16 +95,35 @@ class TestEncoder(unittest.TestCase):
         if camera == 'qcamera':
           continue
 
+        # TODO: this ffprobe call is really slow
         # check frame count
         cmd = f"ffprobe -v error -count_frames -select_streams v:0 -show_entries stream=nb_read_frames \
                -of default=nokey=1:noprint_wrappers=1 {file_path}"
-        expected_frames = self.segment_length * CAMERA_FPS // 2 if (EON and camera=='dcamera') else self.segment_length * CAMERA_FPS
-        frame_tolerance = FRAME_TOLERANCE+1 if (EON and camera=='dcamera') or i==0 else FRAME_TOLERANCE
+        expected_frames = SEGMENT_LENGTH * CAMERA_FPS // 2 if (EON and camera=='dcamera') else SEGMENT_LENGTH * CAMERA_FPS
+        frame_tolerance = 1 if (EON and camera == 'dcamera') else 0
         frame_count = int(subprocess.check_output(cmd, shell=True, encoding='utf8').strip())
 
         self.assertTrue(abs(expected_frames - frame_count) <= frame_tolerance,
                         f"{camera} failed frame count check: expected {expected_frames}, got {frame_count}")
       shutil.rmtree(f"{route_prefix_path}--{i}")
 
+    def join(ts, timeout):
+      for t in ts:
+        t.join(timeout)
+
+    threads = []
+    for i in trange(num_segments):
+      # poll for next segment
+      with Timeout(int(SEGMENT_LENGTH*2), error_msg=f"timed out waiting for segment {i}"):
+        while int(self._get_latest_segment_path().rsplit("--", 1)[1]) <= i:
+          time.sleep(0.1)
+      t = threading.Thread(target=check_seg, args=(i, ))
+      t.start()
+      threads.append(t)
+      join(threads, 0.1)
+
+    with Timeout(20):
+      join(threads, None)
+
 if __name__ == "__main__":
   unittest.main()

selfdrive/loggerd/tests/test_loggerd.py

@@ -15,8 +15,9 @@ from common.basedir import BASEDIR
 from common.timeout import Timeout
 from common.params import Params
 import selfdrive.manager as manager
-from selfdrive.hardware import PC
+from selfdrive.hardware import TICI, PC
 from selfdrive.loggerd.config import ROOT
+from selfdrive.test.helpers import with_processes
 from selfdrive.version import version as VERSION
 from tools.lib.logreader import LogReader
 
@@ -95,6 +96,31 @@ class TestLoggerd(unittest.TestCase):
     for _, k, v in fake_params:
       self.assertEqual(getattr(initData, k), v)
 
+  # TODO: this shouldn't need camerad
+  @with_processes(['camerad'])
+  def test_rotation(self):
+    os.environ["LOGGERD_TEST"] = "1"
+    Params().put("RecordFront", "1")
+    expected_files = {"rlog.bz2", "qlog.bz2", "qcamera.ts", "fcamera.hevc", "dcamera.hevc"}
+    if TICI:
+      expected_files.add("ecamera.hevc")
+
+    for _ in range(5):
+      num_segs = random.randint(1, 10)
+      length = random.randint(2, 5)
+      os.environ["LOGGERD_SEGMENT_LENGTH"] = str(length)
+
+      manager.start_managed_process("loggerd")
+      time.sleep(num_segs * length)
+      manager.kill_managed_process("loggerd")
+
+      route_path = str(self._get_latest_log_dir()).rsplit("--", 1)[0]
+      for n in range(num_segs):
+        p = Path(f"{route_path}--{n}")
+        logged = set([f.name for f in p.iterdir() if f.is_file()])
+        diff = logged ^ expected_files
+        self.assertEqual(len(diff), 0)
+
   def test_bootlog(self):
     # generate bootlog with fake launch log
     launch_log = ''.join([str(random.choice(string.printable)) for _ in range(100)])