test: improve isolation between test uds test cases by removing stale sockets and cancelling zombie uds client handlers

Author: nraultwang

.github/workflows/ci.yml

@@ -10,6 +10,7 @@ jobs:
   compile-test:
     name: Test compilation and individual test files
     runs-on: ubuntu-latest
+    timeout-minutes: 25  # Overall job timeout
 
     steps:
     - name: Checkout repository
@@ -29,23 +30,30 @@ jobs:
         cache-from: type=gha
         cache-to: type=gha,mode=max
 
-    # Individual test file runs for better granularity
-    - name: Run test_can_hashpipe_init.py
+    # Quick functionality tests
+    - name: Run basic functionality tests
+      timeout-minutes: 5
       run: |
         docker run --rm --shm-size=2g panoseti-daq:latest \
           python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_can_hashpipe_init.py
 
-    - name: Run test_pff_header_consistency.py
+    # Framework resilience tests
+    - name: Run UDS resilience tests  
+      timeout-minutes: 5
       run: |
         docker run --rm --shm-size=2g panoseti-daq:latest \
-          python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_pff_header_consistency.py
+          python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_uds_resilience.py
 
-    - name: Run test_uds_data_path.py
+    # Data integrity tests
+    - name: Run UDS data path tests
+      timeout-minutes: 5
       run: |
         docker run --rm --shm-size=2g panoseti-daq:latest \
           python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_uds_data_path.py
 
-    - name: Run test_uds_resilience.py
+    # Filesystem-dependent tests (need more time in containers)
+    - name: Run PFF header consistency tests
+      timeout-minutes: 5
       run: |
         docker run --rm --shm-size=2g panoseti-daq:latest \
-          python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_uds_resilience.py
+          python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_pff_header_consistency.py

run_ci_tests.sh

@@ -1,19 +1,33 @@
 #!/bin/bash
+
 set -e
 
 # Define the name for the Docker image
-IMAGE_NAME="panoseti-daq-ci"
+IMAGE_NAME="panoseti-daq"
 
 echo "--- Building CI Docker Image: $IMAGE_NAME ---"
 docker build -t $IMAGE_NAME -f tests/ci_tests/Dockerfile .
 
 echo "--- Running Integration Tests ---"
-# Run the tests inside the container.
-# The RUN_REAL_DATA_TESTS=1 environment variable enables the fixture.
-# The --rm flag ensures the container is removed after the test run.
-docker run --rm \
-    --shm-size=2g \
-    $IMAGE_NAME \
-    python3 -m pytest -v -s --maxfail=2 tests/ci_tests/
+
+# Run basic functionality tests first (these are most likely to pass)
+echo "=== Running Basic Functionality Tests ==="
+docker run --rm --shm-size=2g $IMAGE_NAME \
+    python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_can_hashpipe_init.py
+
+# Run UDS resilience tests (test framework robustness)
+echo "=== Running UDS Resilience Tests ==="
+docker run --rm --shm-size=2g $IMAGE_NAME \
+    python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_uds_resilience.py
+
+# Run UDS data path tests (test data integrity)
+echo "=== Running UDS Data Path Tests ==="
+docker run --rm --shm-size=2g $IMAGE_NAME \
+    python3 -m pytest -s -v --maxfail=1 tests/ci_tests/test_uds_data_path.py
+
+# Run PFF header consistency tests with longer timeout (filesystem-dependent)
+echo "=== Running PFF Header Consistency Tests ==="
+docker run --rm --shm-size=2g $IMAGE_NAME \
+    python3 -m pytest -s -v --maxfail=1 --timeout=180 tests/ci_tests/test_pff_header_consistency.py
 
 echo "--- CI Test Run Completed Successfully ---"

tests/ci_tests/conftest.py

@@ -9,7 +9,6 @@
 from pathlib import Path
 import stat
 import pytest
-
 from control_util import is_hashpipe_running
 from uds_server import UdsServer
 
@@ -59,64 +58,89 @@ def __init__(self, socket_paths):
 
     def start(self):
         if self._is_stopped:
-            return False  # Don't restart if explicitly stopped
-            
+            # Reset state for restart
+            self._is_stopped = False
+            self.started.clear()
+            self.servers.clear()
+
         def runner():
             self.loop = asyncio.new_event_loop()
             asyncio.set_event_loop(self.loop)
-
+            
             async def start_all():
-                for name, path in self.socket_paths.items():
-                    srv = UdsServer(str(path))
-                    await srv.start()
-                    self.servers[name] = srv
-                self.started.set()
-
-            self.loop.run_until_complete(start_all())
-            self.loop.run_forever()
+                try:
+                    for name, path in self.socket_paths.items():
+                        srv = UdsServer(str(path))
+                        await srv.start()
+                        self.servers[name] = srv
+                        print(f"Started UDS server for {name} at {path}")
+                    self.started.set()
+                except Exception as e:
+                    print(f"Failed to start UDS servers: {e}")
+                    raise
+            
+            try:
+                self.loop.run_until_complete(start_all())
+                self.loop.run_forever()
+            except Exception as e:
+                print(f"UDS server loop failed: {e}")
 
         self.thread = threading.Thread(target=runner, daemon=True)
         self.thread.start()
 
-        # Wait until servers started
-        self.started.wait(timeout=10)
-        if not self.started.is_set():
-            raise RuntimeError("Failed to start UDS servers")
-            
+        # Wait for startup with extended timeout
+        if not self.started.wait(timeout=15):
+            raise RuntimeError("Failed to start UDS servers within timeout")
+        
+        # Give hashpipe time to detect and connect to new sockets
+        time.sleep(2)
         return True
 
     def stop(self):
         self._is_stopped = True
-        
         if self.loop is None or not self.loop.is_running():
             return
 
-        # Schedule the stop coroutine in the event loop
         async def stop_all():
+            tasks = []
             for srv in self.servers.values():
-                await srv.stop()
+                tasks.append(srv.stop())
+            
+            if tasks:
+                await asyncio.gather(*tasks, return_exceptions=True)
 
-        # Run the stop coroutine in the event loop
+        # Schedule the stop coroutine
         self._stop_future = asyncio.run_coroutine_threadsafe(stop_all(), self.loop)
 
-        # Wait for completion
+        # Wait for completion with extended timeout
         try:
-            self._stop_future.result(timeout=10)
-        except Exception:
-            pass
+            self._stop_future.result(timeout=20)
+        except Exception as e:
+            print(f"Error during UDS server shutdown: {e}")
 
         # Stop the event loop
         self.loop.call_soon_threadsafe(self.loop.stop)
 
         # Wait for thread to finish
-        if self.thread:
-            self.thread.join(timeout=10)
+        if self.thread and self.thread.is_alive():
+            self.thread.join(timeout=20)
+            
+        # Ensure all socket files are cleaned up
+        for path in self.socket_paths.values():
+            try:
+                if os.path.exists(str(path)):
+                    os.unlink(str(path))
+            except Exception:
+                pass
+            
+        # Clear state
+        self.servers.clear()
+        self.started.clear()
 
 @pytest.fixture(scope="session")
 def daq_env():
     if not is_utility_available("hashpipe"):
         pytest.fail("hashpipe not found in PATH")
-
     if not is_utility_available("tcpreplay"):
         pytest.fail("tcpreplay not found in PATH")
 
@@ -132,21 +156,20 @@ def daq_env():
     tcpreplay_cmd = [
         "tcpreplay",
         "--mbps=5",
-        "--loop=0", 
+        "--loop=0",
         "--intf1=lo",
         PCAP_FILE,
     ]
     tcpreplay_proc = subprocess.Popen(tcpreplay_cmd)
 
-    # 2) Start hashpipe via start_daq.py (ensure your start_daq.py uses psutil-based PID find)
+    # 2) Start hashpipe via start_daq.py
     start_daq = [
         sys.executable,
         "/app/tests/ci_tests/start_daq.py",
         "--run_dir", str(BASE_DIR),
         "--max_file_size_mb", "5",
         "--bindhost", "lo",
     ]
-
     for mid in MODULE_IDS:
         start_daq.extend(["--module_id", str(mid)])
 
@@ -163,13 +186,6 @@ def daq_env():
             print("No PID file was created by start_daq.py")
         raise
 
-    # 4) Optional: verify at least one server saw a connection within 10s
-    start = time.time()
-    while time.time() - start < 10:
-        # If any server's connected event is set, we know hashpipe connected to at least one DP
-        # Skip strict requirement; filesystem checks will also verify pipeline
-        break
-
     env = {
         "base_dir": BASE_DIR,
         "run_name": RUN_NAME,
@@ -182,7 +198,6 @@ def daq_env():
 
     try:
         yield env
-
     finally:
         print("\n-- Tearing down DAQ environment --")
 
@@ -201,7 +216,6 @@ def daq_env():
             sys.executable,
             "/app/tests/ci_tests/stop_daq.py",
         ]
-
         try:
             cp = subprocess.run(stop_daq, cwd=BASE_DIR, capture_output=True, text=True, timeout=15)
             print("stop_daq.py stdout:\n", cp.stdout)