Revamp the test.

Make it use the new test infrastructure.

Signed-off-by: Chris Lalancette <[email protected]>

Author: clalancette

image_tools_py/package.xml

@@ -13,15 +13,11 @@
   <exec_depend>std_msgs</exec_depend>
   <exec_depend>sensor_msgs</exec_depend>
 
-  <!-- These test dependencies are optional
-  Their purpose is to make sure that the code passes the linters -->
   <test_depend>ament_copyright</test_depend>
   <test_depend>ament_flake8</test_depend>
   <test_depend>ament_pep257</test_depend>
-  <test_depend>ament_pep8</test_depend>
-  <test_depend>ament_pyflakes</test_depend>
   <test_depend>launch</test_depend>
-  <test_depend>launch_testing</test_depend>
+  <test_depend>python3-pytest</test_depend>
 
   <export>
     <build_type>ament_python</build_type>

image_tools_py/setup.py

@@ -26,11 +26,9 @@ def run(self):
 
         for rmw_impl in rmw_implementations:
             substs = {
-                '@rmw_implementation@': rmw_impl,
-                '@showimage_py_exe@': os.path.join(self.install_dir, 'showimage_py'),
-                '@showimage_py_outfile@': os.path.realpath(os.path.join('test', 'showimage_py')),
-                '@cam2image_py_exe@': os.path.join(self.install_dir, 'cam2image_py'),
-                '@cam2image_py_outfile@': os.path.realpath(os.path.join('test', 'cam2image_py')),
+                '@RMW_IMPLEMENTATION@': rmw_impl,
+                '@RCLPY_DEMO_SHOWIMAGE_EXE@': os.path.join(self.install_dir, 'showimage_py'),
+                '@RCLPY_DEMO_CAM2IMAGE_EXE@': os.path.join(self.install_dir, 'cam2image_py'),
             }
 
             infile = os.path.join('test', 'test_showimage_cam2image.py.in')
@@ -39,7 +37,7 @@ def run(self):
                 with open(outfile, 'w') as outfp:
                     for line in infp:
                         for subst in substs:
-                            line = line.replace(subst, "r'%s'  # noqa" % (substs[subst]))
+                            line = line.replace(subst, substs[subst])
                         outfp.write(line)
 
         install_scripts.run(self)

image_tools_py/test/cam2image_py.txt

@@ -1,4 +1,4 @@
-# Copyright 2016 Open Source Robotics Foundation, Inc.
+# Copyright 2019 Open Source Robotics Foundation, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -13,90 +13,102 @@
 # limitations under the License.
 
 import os
-
-from launch.legacy import LaunchDescriptor
-from launch.legacy.exit_handler import ignore_exit_handler
-from launch.legacy.exit_handler import primary_ignore_returncode_exit_handler
-from launch.legacy.launcher import DefaultLauncher
-from launch.legacy.output_handler import ConsoleOutput
-from launch_testing import create_handler
-
-
-def setup():
-    os.environ['OSPL_VERBOSITY'] = '8'  # 8 = OS_NONE
-    # bare minimum formatting for console output matching
-    os.environ['RCUTILS_CONSOLE_OUTPUT_FORMAT'] = '{message}'
-
-
-def test_reliable_qos():
-    pub_executable_args = ['-r', '1', '-s', '0', '-b', '-f', '5']
-    sub_executable_args = ['-r', '1', '-s', '0']
-    _test_executables(pub_executable_args, sub_executable_args)
-
-
-def _test_executables(publisher_executable_args, subscriber_executable_args):
-    output_handlers = []
-    launch_descriptor = LaunchDescriptor()
-
-    rmw_implementation = @rmw_implementation@
-    env = dict(os.environ)
-    env['RMW_IMPLEMENTATION'] = rmw_implementation
-    env['PYTHONUNBUFFERED'] = '1'
-
-    # Launch the process that will receive the images.
-    # This is the process that gets to decide when to tear the launcher down.
-    # It will exit when the regex for receiving images is matched.
-    showimage_executable = @showimage_py_exe@
-    showimage_output_file = @showimage_py_outfile@
-    showimage_name = 'test_showimage_py'
-    showimage_handler = create_handler(
-        showimage_name, launch_descriptor, showimage_output_file, exit_on_match=True,
-        filtered_rmw_implementation=rmw_implementation)
-    assert showimage_handler, 'Cannot find appropriate handler for %s' % showimage_output_file
-    output_handlers.append(showimage_handler)
-
-    command = [showimage_executable]
-    command.extend(subscriber_executable_args)
-    launch_descriptor.add_process(
-        cmd=command,
-        name=showimage_name,
-        exit_handler=primary_ignore_returncode_exit_handler,
-        output_handlers=[showimage_handler, ConsoleOutput()],
-        env=env,
-    )
-
-    # Launch the process that will publish the images.
-    # This process will be exited when the launcher is torn down.
-    cam2image_executable = @cam2image_py_exe@
-    cam2image_output_file = @cam2image_py_outfile@
-    cam2image_name = 'test_cam2image_py'
-    cam2image_handler = create_handler(
-        cam2image_name, launch_descriptor, cam2image_output_file, exit_on_match=False,
-        filtered_rmw_implementation=rmw_implementation)
-    assert cam2image_handler, 'Cannot find appropriate handler for %s' % cam2image_output_file
-    output_handlers.append(cam2image_handler)
-
-    command = [cam2image_executable]
-    command.extend(publisher_executable_args)
-    launch_descriptor.add_process(
-        cmd=command,
-        name=cam2image_name,
-        exit_handler=ignore_exit_handler,
-        output_handlers=[cam2image_handler, ConsoleOutput()],
-        env=env,
-    )
-
-    launcher = DefaultLauncher()
-    launcher.add_launch_descriptor(launch_descriptor)
-    rc = launcher.launch()
-
-    assert rc == 0, \
-        "The launch file failed with exit code '" + str(rc) + "'. "
-
-    # Check that the output received by the handlers is what was expected.
-    for handler in output_handlers:
-        handler.check()
-
-
-if __name__ == '__main__':
-    test_reliable_qos()
+import time
+import unittest
+
+from launch import LaunchDescription
+from launch import LaunchService
+from launch.actions import EmitEvent
+from launch.actions import ExecuteProcess
+from launch.actions import RegisterEventHandler
+from launch.event_handlers import OnProcessIO
+from launch.events import Shutdown
+
+
+# TODO(clalancette): we can't currently use launch_testing since there is
+# currently no integration with pure python setup.py.  Once we have a solution
+# for https://github.com/ros2/launch/issues/237, we should revisit this.
+class TestExecutablesDemo(unittest.TestCase):
+
+    def __init__(self, name):
+        super().__init__(name)
+        self._start_time = time.time()
+        self._ls = LaunchService()
+        self._ls.include_launch_description(self.generate_launch_description())
+        self._saw_cam2image_output = False
+        self._saw_showimage_output = False
+
+    def generate_launch_description(self):
+        launch_description = LaunchDescription()
+        publisher_executable_args = ['-r', '1', '-s', '0', '-b', '-f', '5']
+        subscriber_executable_args = ['-r', '1', '-s', '0']
+
+        env = dict(os.environ)
+        env['PYTHONUNBUFFERED'] = '1'
+        env['OSPL_VERBOSITY'] = '8'  # 8 = OS_NONE
+        # bare minimum formatting for console output matching
+        env['RCUTILS_CONSOLE_OUTPUT_FORMAT'] = '{message}'
+        env['RMW_IMPLEMENTATION'] = r'@RMW_IMPLEMENTATION@'
+
+        # Launch the process that will receive the images.
+        # This is the process that gets to decide when to tear the launcher down.
+        # It will exit when the regex for receiving images is matched.
+        showimage_executable = r'@RCLPY_DEMO_SHOWIMAGE_EXE@'  # noqa: E501
+        showimage_name = 'test_showimage_py'
+
+        command = [showimage_executable]
+        command.extend(subscriber_executable_args)
+        showimage_process = ExecuteProcess(
+            cmd=command,
+            name=showimage_name,
+            env=env,
+            output='screen'
+        )
+        launch_description.add_action(showimage_process)
+
+        # Launch the process that will publish the images.
+        # This process will be exited when the launcher is torn down.
+        cam2image_executable = r'@RCLPY_DEMO_CAM2IMAGE_EXE@'  # noqa: E501
+        cam2image_name = 'test_cam2image_py'
+
+        command = [cam2image_executable]
+        command.extend(publisher_executable_args)
+        cam2image_process = ExecuteProcess(
+            cmd=command,
+            name=cam2image_name,
+            env=env,
+            output='screen'
+        )
+        launch_description.add_action(cam2image_process)
+
+        launch_description.add_action(
+            RegisterEventHandler(
+                OnProcessIO(
+                    on_stdout=self.append,
+                    on_stderr=self.append,
+                )
+            )
+        )
+
+        return launch_description
+
+    def append(self, process_io):
+        if 'cam2image' in process_io.process_name:
+            if 'Publishing image #' in process_io.text.decode():
+                self._saw_cam2image_output = True
+        elif 'showimage' in process_io.process_name:
+            if 'Received image #' in process_io.text.decode():
+                self._saw_showimage_output = True
+
+        if self._saw_cam2image_output and self._saw_showimage_output:
+            # We've seen all required arguments from the test, quit
+            return EmitEvent(event=Shutdown(reason='finished', due_to_sigint=False))
+
+        if time.time() - self._start_time > 10.0:
+            # We've waited more than 10 seconds with no output; raise an error
+            return EmitEvent(event=Shutdown(reason='timeout', due_to_sigint=False))
+
+    def test_reliable_qos(self):
+        self._ls.run()
+        self._ls.shutdown()
+        self.assertTrue(self._saw_cam2image_output and self._saw_showimage_output)