hotfix(test): Fix reliability in CAN tests

reseq_ros2/test/test_nodes/test_communication/test_communication_functional.py

@@ -1,5 +1,5 @@
-import subprocess
 import unittest
+from test.utils.test_utils import check_interface_status
 
 import rclpy
 from rclpy.executors import SingleThreadedExecutor
@@ -11,22 +11,6 @@
 # Function Test
 
 
-def check_interface_status(interface_name):
-    """Check if the specified network interface is up and running."""
-    # Run the `ip` command to get interface details
-    result = subprocess.run(['ip', 'link', 'show', interface_name], capture_output=True, text=True)
-
-    # Check if the command was successful
-    if result.returncode != 0:
-        return False, f'Failed to get interface status. Error: {result.stderr.strip()}'
-
-    # Check if the interface is up
-    if 'UP,LOWER_UP' in result.stdout:
-        return True, f'Interface {interface_name} is up and running.'
-    else:
-        return False, f'Interface {interface_name} is not up.'
-
-
 # Define the interface status and message
 interface_status, status_msg = check_interface_status('vcan0')
 
@@ -53,6 +37,7 @@ def setUpClass(cls):
     @classmethod
     def tearDownClass(cls):
         cls.executor.shutdown()
+        cls.communication.canbus.shutdown()
         cls.communication.destroy_node()
         rclpy.shutdown()
 

reseq_ros2/test/test_system/test_topic_correctness.py

@@ -1,5 +1,6 @@
 import importlib
 import struct
+import subprocess
 import sys
 import time
 import unittest
@@ -21,6 +22,18 @@
 from reseq_interfaces.msg import Remote
 
 
+def kill_node(node_name):
+    try:
+        # Use pgrep to find the PID of the node
+        pid = subprocess.check_output(['pgrep', '-f', node_name]).strip()
+
+        if pid:
+            subprocess.run(['kill', pid.decode()])
+
+    except subprocess.CalledProcessError:
+        pass
+
+
 class FrequencyChecker(Node):
     """
     A ROS2 node that checks the frequency of messages on specified topics.
@@ -133,6 +146,10 @@ def setUpClass(cls):
         # Wait for launch
         time.sleep(5)
 
+        # Kill the feedback_replicator node for CAN frequency test
+        if cls.stat[0]:
+            kill_node('feedback_replicator')
+
     @classmethod
     def tearDownClass(cls):
         tear_down_process(cls.process)