Update apex rostest. (#14)

* Give more info when output is not in expected sequence
* Real captured output has newline in it.  Update test data to reflect
* Fix exception with process name resolution, add test
* Add infrastructure to coordinate ROS_DOMAIN_IDs between apex_launchtest processes
* Isolate tests with ROS_DOMAIN_ID by default
* Address PR comments
* Fix style
  - Defaults in the docker image we run the test in changed. We're now failing for style
    reasons. Mostly need to re-order imports and change " quotes to '
* apex_launchtest_ros add better example
  - Fix up style issues
  - Typo
  - Update apex_launchtest_ros/examples/README.md
  - Formatting
  - formatting
  - Fix bad spelling
  - Improve sentence
  - Fix Typo
* Improve apex launchtest ros
  - Add the parametrization infrastrcture
  - Refactor the apex runner into an ApexRunner and a RunnerWorker
  - This veresion should still run existing tests.  This is mostly a refactor commit
* Refactor to allow for parametrized launch descriptions
  - Combine pre and post results into one set of results, add class names so the tests
    are distinguishable
  - Update tests to match new 'run' API that returns a dictionary of {test_run: result} instead
    of a tuple of (pre_shutdown, post_shutdown) results
* Add parametrized example, finish refactor
* Beef up test coverage for new features
* Push changes into apex_rostest subdirectory.
* Prune unnecesary files.

Co-Authored-By: pbaughman <[email protected]>
Signed-off-by: Pete Baughman <[email protected]>
Signed-off-by: Michel Hidalgo <[email protected]>

Author: hidmic

apex_rostest/apex_launchtest_ros/apex_launchtest_ros/__init__.py

@@ -13,9 +13,11 @@
 # limitations under the License.
 
 
+from .data_republisher import DataRepublisher
 from .message_pump import MessagePump
 
 
 __all__ = [
+    'DataRepublisher',
     'MessagePump',
 ]

apex_rostest/apex_launchtest_ros/apex_launchtest_ros/data_republisher.py

@@ -0,0 +1,79 @@
+# Copyright 2019 Apex.AI, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+class DataRepublisher:
+    """Republish mesasges with a transform function applied."""
+
+    def __init__(self, node, listen_topic, publish_topic, msg_type, transform_fn):
+        """
+        Create a DataRepublisher.
+
+        :param node: A rclpy node that will run the publisher and subscriber
+
+        :param listen_topic: The topic to listen for incoming messages on
+
+        :param publish_topic: The topic to republish messages on
+
+        :param msg_type: The type of ROS msg to receive and republish
+
+        :param transfor_fn: A function that takes one mesasge and returns a new message to
+        republish or None to drop the message
+        """
+        self.__num_received = 0
+        self.__republished_list = []
+
+        self.__node = node
+        self.__subscriber = node.create_subscription(
+            msg_type,
+            listen_topic,
+            callback=self.__cb
+        )
+
+        self.__publisher = node.create_publisher(
+            msg_type,
+            publish_topic
+        )
+
+        self.__transform_fn = transform_fn
+
+    def shutdown(self):
+        """Stop republishing messages."""
+        self.__node.destroy_subscription(self.__subscriber)
+        self.__node.destroy_publisher(self.__publisher)
+
+    def get_num_received(self):
+        """Get the number of messages received on the listen_topic."""
+        return self.__num_received
+
+    def get_num_republished(self):
+        """
+        Get the number of messages published on publish_topic.
+
+        This may be lower than get_num_received if the transform_fn indicated a message
+        should be dropped
+        """
+        return len(self.__republished_list)
+
+    def get_republished(self):
+        """Get a list of all of the transformed messages republished."""
+        return self.__republished_list
+
+    def __cb(self, msg):
+        self.__num_received += 1
+        repub = self.__transform_fn(msg)
+
+        if repub:
+            self.__republished_list.append(msg)
+            self.__publisher.publish(repub)

apex_rostest/apex_launchtest_ros/apex_launchtest_ros/message_pump.py

@@ -25,7 +25,7 @@ def __init__(self, node, context=None):
         self._node = node
         self._thread = threading.Thread(
             target=self._run,
-            name="msg_pump_thread",
+            name='msg_pump_thread',
         )
         self._run = True
         self._context = context
@@ -37,7 +37,7 @@ def stop(self):
         self._run = False
         self._thread.join(timeout=5.0)
         if self._thread.is_alive():
-            raise Exception("Timed out waiting for message pump to stop")
+            raise Exception('Timed out waiting for message pump to stop')
 
     def _run(self):
         executor = rclpy.executors.SingleThreadedExecutor(context=self._context)

apex_rostest/apex_launchtest_ros/examples/README.md

@@ -1,8 +1,31 @@
 # Examples
 
-## message_counter.launch.py
+## `talker_listener.test.py`
 
 Usage:
-> apex_launchtest examples/message_counter.launch.py
+> apex_launchtest examples/talker_listener.test.py
 
-Launches a message counter node and runs a few tests to interact with the node
+This test launches the talker and listener example nodes from demo_nodes_py and interacts
+with them via their ROS interfaces.  Remapping rules are used so that one of the tests can sit in
+between the talker and the listener and change the data on the fly.
+
+Node that in the setUpClass method, the test makes sure that the listener is subscribed and
+republishing messages.  Since the listener process provides no synchronization mechanism to
+inform the outside world that it's up and running, this step is necessary especially in resource
+constrained environments where process startup may take a non negligible amount of time.  This
+is often the cause of "flakyness" in tests on CI systems.  A more robust design of the talker and
+listener processes might provide some positive feedback that the node is up and running, but these
+are simple example nodes.
+
+#### test_fuzzy_data
+This test gives an example of what a test that fuzzes data might look like.  A ROS subscriber
+and publisher pair encapsulated in a `DataRepublisher` object changes the string "Hello World" to
+"Aloha World" as it travels between the talker and the listener.
+
+#### test_listener_receives
+This test publishes unique messages on the `/chatter` topic and asserts that the same messages
+go to the stdout of the listener node
+
+#### test_talker_transmits
+This test subscribes to the remapped `/talker_chatter` topic and makes sure the talker node also
+writes the data it's transmitting to stdout

apex_rostest/apex_launchtest_ros/examples/talker_listener.test.py

@@ -0,0 +1,188 @@
+# Copyright 2019 Apex.AI, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import time
+import unittest
+import uuid
+
+from apex_launchtest.util import NoMatchingProcessException
+import apex_launchtest_ros
+import launch
+import launch_ros
+import launch_ros.actions
+import rclpy
+import rclpy.context
+import rclpy.executors
+import std_msgs.msg
+
+
+def generate_test_description(ready_fn):
+    # Necessary to get real-time stdout from python processes:
+    proc_env = os.environ.copy()
+    proc_env['PYTHONUNBUFFERED'] = '1'
+
+    # Normally, talker publishes on the 'chatter' topic and listener listens on the
+    # 'chatter' topic, but we want to show how to use remappings to munge the data so we
+    # will remap these topics when we launch the nodes and insert our own node that can
+    # change the data as it passes through
+    talker_node = launch_ros.actions.Node(
+        package='demo_nodes_py',
+        node_executable='talker',
+        env=proc_env,
+        remappings=[('chatter', 'talker_chatter')],
+    )
+
+    listener_node = launch_ros.actions.Node(
+        package='demo_nodes_py',
+        node_executable='listener',
+        env=proc_env,
+    )
+
+    return (
+        launch.LaunchDescription([
+            talker_node,
+            listener_node,
+            # Start tests right away - no need to wait for anything
+            launch.actions.OpaqueFunction(function=lambda context: ready_fn()),
+        ]),
+        {
+            'talker': talker_node,
+            'listener': listener_node,
+        }
+    )
+
+
+class TestTalkerListenerLink(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls, proc_output, listener):
+        cls.context = rclpy.context.Context()
+        rclpy.init(context=cls.context)
+        cls.node = rclpy.create_node('test_node', context=cls.context)
+
+        # The demo node listener has no synchronization to indicate when it's ready to start
+        # receiving messages on the /chatter topic.  This plumb_listener method will attempt
+        # to publish for a few seconds until it sees output
+        publisher = cls.node.create_publisher(
+            std_msgs.msg.String,
+            'chatter'
+        )
+        msg = std_msgs.msg.String()
+        msg.data = 'test message {}'.format(uuid.uuid4())
+        for _ in range(5):
+            try:
+                publisher.publish(msg)
+                proc_output.assertWaitFor(
+                    msg=msg.data,
+                    process=listener,
+                    timeout=1.0
+                )
+            except AssertionError:
+                continue
+            except NoMatchingProcessException:
+                continue
+            else:
+                return
+        else:
+            assert False, 'Failed to plumb chatter topic to listener process'
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.node.destroy_node()
+
+    def spin_rclpy(self, timeout_sec):
+        executor = rclpy.executors.SingleThreadedExecutor(context=self.context)
+        executor.add_node(self.node)
+        try:
+            executor.spin_once(timeout_sec=timeout_sec)
+        finally:
+            executor.remove_node(self.node)
+
+    def test_talker_transmits(self, talker):
+        # Expect the talker to publish strings on '/talker_chatter' and also write to stdout
+        msgs_rx = []
+        sub = self.node.create_subscription(
+            std_msgs.msg.String,
+            'talker_chatter',
+            callback=lambda msg: msgs_rx.append(msg)
+        )
+        self.addCleanup(self.node.destroy_subscription, sub)
+
+        # Wait until the talker transmits two messages over the ROS topic
+        end_time = time.time() + 10
+        while time.time() < end_time:
+            self.spin_rclpy(1.0)
+            if len(msgs_rx) > 2:
+                break
+
+        self.assertGreater(len(msgs_rx), 2)
+
+        # Make sure the talker also output the same data via stdout
+        for txt in [msg.data for msg in msgs_rx]:
+            self.proc_output.assertWaitFor(
+                msg=txt,
+                process=talker
+            )
+
+    def test_listener_receives(self, listener):
+        pub = self.node.create_publisher(
+            std_msgs.msg.String,
+            'chatter'
+        )
+        self.addCleanup(self.node.destroy_publisher, pub)
+
+        # Publish some unique messages on /chatter and verify that the listener gets them
+        # and prints them
+        for _ in range(5):
+            msg = std_msgs.msg.String()
+            msg.data = str(uuid.uuid4())
+
+            pub.publish(msg)
+            self.proc_output.assertWaitFor(
+                msg=msg.data,
+                process=listener
+            )
+
+    def test_fuzzy_data(self, listener):
+        # This test shows how to insert a node in between the talker and the listener to
+        # change the data.  Here we're going to change 'Hello World' to 'Aloha World'
+        def data_mangler(msg):
+            msg.data = msg.data.replace('Hello', 'Aloha')
+            return msg
+
+        republisher = apex_launchtest_ros.DataRepublisher(
+            self.node,
+            'talker_chatter',
+            'chatter',
+            std_msgs.msg.String,
+            data_mangler
+        )
+        self.addCleanup(republisher.shutdown)
+
+        # Spin for a few seconds until we've republished some mangled messages
+        end_time = time.time() + 10
+        while time.time() < end_time:
+            self.spin_rclpy(1.0)
+            if republisher.get_num_republished() > 2:
+                break
+
+        self.assertGreater(republisher.get_num_republished(), 2)
+
+        # Sanity check that we're changing 'Hello World'
+        self.proc_output.assertWaitFor('Aloha World')
+
+        # Check for the actual messages we sent
+        for msg in republisher.get_republished():
+            self.proc_output.assertWaitFor(msg.data, listener)

apex_rostest/apex_launchtest_ros/package.xml

@@ -15,6 +15,8 @@
   <test_depend>ament_flake8</test_depend>
   <test_depend>ament_pep257</test_depend>
   <test_depend>python3-pytest</test_depend>
+  <test_depend>demo_nodes_py</test_depend>
+  <test_depend>std_msgs</test_depend>
 
   <export>
     <build_type>ament_python</build_type>

apex_rostest/apex_launchtest_ros/setup.cfg

@@ -2,3 +2,7 @@
     # This will let coverage find files with 0% coverage (not hit by tests at all)
     source = .
     omit = setup.py
+
+[flake8]
+    # For the purpose of import-order style checker, consider apex_launchtest as a local package
+    import-order-style = "apex_launchtest"

apex_rostest/apex_launchtest_ros/setup.py

@@ -1,8 +1,9 @@
 #!/usr/bin/env python
 
-from setuptools import setup
 import glob
 
+from setuptools import setup
+
 package_name = 'apex_launchtest_ros'
 
 setup(
@@ -21,6 +22,6 @@
     packages=[
         'apex_launchtest_ros',
     ],
-    tests_require=["pytest"],
+    tests_require=['pytest'],
     zip_safe=True,
 )