fix: make test longer and increase data rate

Author: nraultwang

tests/ci_tests/conftest.py

@@ -103,10 +103,10 @@ def daq_env():
     uds_mgr = UdsServerManager(uds_paths)
     uds_mgr.start()
 
-    # 1) Start tcpreplay (loop indefinitely at 1 Mbps to lo)
+    # 1) Start tcpreplay (loop indefinitely at 5 Mbps to lo)
     tcpreplay_cmd = [
         "tcpreplay",
-        "--mbps=1",
+        "--mbps=5",
         "--loop=0",
         "--intf1=lo",
         PCAP_FILE,
@@ -118,7 +118,7 @@ def daq_env():
         sys.executable,
         "/app/tests/ci_tests/start_daq.py",
         "--run_dir", str(BASE_DIR),
-        "--max_file_size_mb", "1",
+        "--max_file_size_mb", "5",
         "--bindhost", "lo",
     ]
     for mid in MODULE_IDS:

tests/ci_tests/start_daq.py

@@ -99,7 +99,7 @@ def main():
         close_fds=True,
         stdin=subprocess.DEVNULL,
         stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
+        # stderr=subprocess.DEVNULL,
     )
 
     # Build expected tokens to identify the hashpipe process reliably

tests/ci_tests/stop_daq.py

@@ -1,41 +1,110 @@
-#! /usr/bin/env python3
-
-# This script is run (remotely) on a DAQ node to stop recording
-# - get the PID of the hashpipe process
-# - send it a SIGINT signal
-# - wait for it to exit
-#
-# Then kill any other hashpipe processes
-#
-# On success, print OK.  Otherwise print an error message
-
-import os, signal, sys
+import os, signal, sys, time
 sys.path.append('/app/')
 import control_util as util
+import psutil
+
+GRACE_SECONDS = 5.0
+
+def _send_sigint_to_pid(pid: int):
+    try:
+        p = psutil.Process(pid)
+        if p.name() == util.hashpipe_name:
+            os.kill(pid, signal.SIGINT)
+            return True
+    except Exception:
+        pass
+    return False
+
+def _any_hashpipe_pids():
+    pids = []
+    for p in psutil.process_iter(attrs=["pid","name"]):
+        try:
+            if p.info["name"] == util.hashpipe_name:
+                pids.append(p.info["pid"])
+        except Exception:
+            continue
+    return pids
+
+def _wait_for_exit(pids, timeout_s):
+    deadline = time.time() + timeout_s
+    while time.time() < deadline:
+        alive = []
+        for pid in pids:
+            try:
+                os.kill(pid, 0)
+                alive.append(pid)
+            except Exception:
+                pass
+        if not alive:
+            return True
+        time.sleep(0.1)
+    return False
 
 def main():
+    # 1) Try SIGINT to the recorded PID if present
+    recorded_pid = None
     try:
-        f = open(util.daq_hashpipe_pid_filename, 'r')
-    except:
-        f = None
-    if f:
-        pid = int(f.read())
-        f.close()
-        if not util.stop_hashpipe(pid):
-            print("Couldn't stop hashpipe")
-        os.unlink(util.daq_hashpipe_pid_filename)
-
-    util.kill_hashpipe()
-
-    # if the HK recorder is running on a remote DAQ, we didn't start it.
-    # But it shouldn't be there, so kill it
-    util.kill_hk_recorder()
+        with open(util.daq_hashpipe_pid_filename, 'r') as f:
+            recorded_pid = int(f.read().strip())
+    except Exception:
+        recorded_pid = None
+
+    sent = False
+    target_pids = []
+    if recorded_pid and _send_sigint_to_pid(recorded_pid):
+        sent = True
+        target_pids = [recorded_pid]
+    else:
+        # 2) No valid recorded PID: send SIGINT to all hashpipe processes
+        hp_pids = _any_hashpipe_pids()
+        for pid in hp_pids:
+            try:
+                os.kill(pid, signal.SIGINT)
+                sent = True
+            except Exception:
+                pass
+        target_pids = hp_pids
+
+    # 3) Wait briefly for graceful exit; if still alive, SIGKILL them
+    if target_pids:
+        if not _wait_for_exit(target_pids, GRACE_SECONDS):
+            # escalate
+            for pid in target_pids:
+                try:
+                    os.kill(pid, signal.SIGKILL)
+                except Exception:
+                    pass
+            # Wait one more short interval to ensure death
+            _wait_for_exit(target_pids, 1.0)
+
+    # 4) As a final safety, if any hashpipe remains, kill them (covers races)
+    residual = _any_hashpipe_pids()
+    if residual:
+        for pid in residual:
+            try:
+                os.kill(pid, signal.SIGKILL)
+            except Exception:
+                pass
+
+    # 5) Cleanup pid file and auxiliary daemons
+    try:
+        if os.path.exists(util.daq_hashpipe_pid_filename):
+            os.unlink(util.daq_hashpipe_pid_filename)
+    except Exception:
+        pass
+
+    # HK recorder shouldn’t run in CI; ensure it’s not left behind
+    try:
+        util.kill_hk_recorder()
+    except Exception:
+        pass
 
     try:
         os.unlink(util.daq_run_name_filename)
-    except:
+    except Exception:
         pass
 
     print('stop_daq.py: OK')
 
-main()
+if __name__ == '__main__':
+    main()

tests/ci_tests/test_can_hashpipe_init.py

@@ -35,8 +35,8 @@ def test_hashpipe_valid(daq_env):
     for _ in range(10):
         assert is_hashpipe_running(), "Hashpipe process should be running."
         print_status()
-        time.sleep(1)
-    assert is_hashpipe_running(), "Hashpipe process should still be running after 10 seconds."
+        time.sleep(2)
+    assert is_hashpipe_running(), "Hashpipe process should still be running after 20 seconds."
 
 def _latest_pff_mtime(mod_run: Path):
     pffs = list(mod_run.glob("*.pff"))