Passing parameter for launch testing in ros2

edit

Hi,
I am running launch testing where I have a test node that's used for testing if messages are been transmitted. When I run individual nodes like pub_demo or sub_demo it works, it takes parameters from the config file. But when I run the demo_pkg_test.py using launch_test these two nodes don't take the parameters. Even the test node doesn't take the parameters.

src
├── demo_pkg
│   ├── CMakeLists.txt
│   ├── config
│   │   └── params.yaml
│   ├── include
│   │   └── demo_pkg
│   │       └── qos_custom_profile.hpp
│   ├── package.xml
│   ├── src
│   │   ├── pub_demo.cpp
│   │   └── sub_demo.cpp
│   └── test
│       ├── demo_pkg_test.py

Below is my code

Streaming = QoSProfile(durability=qos.QoSDurabilityPolicy.VOLATILE,
                   reliability=qos.QoSReliabilityPolicy.BEST_EFFORT, history=qos.QoSHistoryPolicy.KEEP_LAST, depth=1)
State = QoSProfile(durability=qos.QoSDurabilityPolicy.TRANSIENT_LOCAL,
               reliability=qos.QoSReliabilityPolicy.RELIABLE, history=qos.QoSHistoryPolicy.KEEP_LAST, depth=20)


@pytest.mark.launch_test
@launch_testing.markers.keep_alive
def generate_test_description():

config = os.path.join('/home/'+$USER+'/dev_ws/src/demo_pkg/config/params.yaml')


print(config)
return launch.LaunchDescription([
    launch_ros.actions.Node(
        executable='pub_ex',
        package='demo_pkg',
        name='pub_node',
        parameters=[config]
    ),
    launch_ros.actions.Node(
        executable='sub_ex',
        package='demo_pkg',
        name='sub_node',
        parameters=[config]
    ),
    launch_testing.actions.ReadyToTest()
])


class TestFixture(unittest.TestCase):

def test_pin_robot_enable(self):
    rclpy.init()
    node = Node('test_node')
    self.declare_parameters(
        namespace = '',
        parameters = [
        ('robot_name',None)])
    robot_name = self.get_parameter('robot_name').get_parameter_value().string_value
    print("-------1----",robot_name)
    self.robot_enable_data = []
    sub = node.create_subscription(
        Bool,
        '/sub_topic',
        self.topicMsgData,
        Streaming
    )
    self.pub = node.create_publisher(
        Bool,
        '/pub_topic' ,
        Streaming
    )
    timer_period = 0.5  # seconds
    self.timer = node.create_timer(timer_period, self.pub_callback)
    self.value = Bool()

    try:
        end_time = time.time() + 10
        self.test_data_recv = []
        while time.time() < end_time:
            rclpy.spin_once(node, timeout_sec=0.1)

        if self.test_data_recv == []:
            test_data = ""
        else:
            test_data = self.test_data[-1]


        self.assertEqual(test_data, True,
                         "Error")

    finally:
        node.destroy_subscription(sub)
    rclpy.shutdown()

def pub_callback(self):
    data= Bool()
    data.data = True
    self.pub.publish(data)

def topicMsgData(self, msg):
    self.test_data_recv.append(msg.data)

Below is my params.yaml file data.

pub_node:
  ros__parameters:
    robot_name: "BOT02/"

sub_node:
  ros__parameters:
    robot_name: "BOT02/"

test_node:
  ros__parameters:
    robot_name: "BOT02/"

Thanks in advance, any help is much appreciated.