Add in a test for showimage_py and cam2image_py.

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

Author: clalancette

image_tools_py/cam2image_py.py

@@ -89,6 +89,9 @@ def main(args=None):
     parser.add_argument(
         '-s', '--show', dest='show_camera', action='store', default=0, type=int, choices=[0, 1],
         help='Show the camera stream')
+    parser.add_argument(
+        '-t', '--topic', dest='topic', action='store', default='image', type=str,
+        help='Topic to publish on')
     parser.add_argument(
         '-x', '--width', dest='width', action='store', default=320, type=int,
         help='Image width')
@@ -119,6 +122,8 @@ def main(args=None):
     # parameter.
     custom_camera_qos_profile.history = args.history_policy
 
+    print("Publishing data on topic '%s'" % (args.topic))
+
     # Create the image publisher with our custom QoS profile.
     pub = node.create_publisher(
         sensor_msgs.msg.Image, 'image',

image_tools_py/package.xml

@@ -4,9 +4,9 @@
   <name>image_tools_py</name>
   <version>0.0.0</version>
   <description>Python tools to capture/play back images to/from DDS subscriptions/publications.</description>
-  <maintainer email="clalancette@osrfoundation.org">Chris Lalancette</maintainer>
+  <maintainer email="clalancette@openrobotics.org">Chris Lalancette</maintainer>
   <license>Apache License 2.0</license>
-  <author email="clalancette@osrfoundation.org">Chris Lalancette</author>
+  <author email="clalancette@openrobotics.org">Chris Lalancette</author>
 
   <exec_depend>python3-opencv</exec_depend>
   <exec_depend>rclpy</exec_depend>

image_tools_py/setup.py

@@ -1,7 +1,50 @@
+import os
+
 from setuptools import setup
+from setuptools.command.install_scripts import install_scripts
 
 package_name = 'image_tools_py'
 
+
+# This configuration deserves some explanation.  The goal is to generate a
+# test file called something like 'test_showimage_cam2image__rmw_fastrtps_.py'
+# that contains the test from test_showimage_cam2image.py.in, customized
+# for the rmw_implementation we want to test and for absolute paths of this
+# installation.  To do that, we override the 'install_scripts' stage of
+# setuptools, since that is the stage that knows the paths to the 'cam2image_py'
+# and 'showimage_py' executables.
+#
+# TODO(clalancette): One downside to what is being done below is that the output
+# file is being left in the source directory.  That's currently necessary
+# because the pytest invocation only looks there for test scripts.  It would
+# be nicer to generate the output file in the 'build' directory and then point
+# pytest at it, but I don't know how to do that.
+class custom_install_scripts(install_scripts):
+
+    def run(self):
+        rmw_implementations = ['rmw_fastrtps_cpp']
+
+        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')),
+            }
+
+            infile = os.path.join('test', 'test_showimage_cam2image.py.in')
+            outfile = os.path.join('test', 'test_showimage_cam2image__%s_.py' % (rmw_impl))
+            with open(infile, 'r') as infp:
+                with open(outfile, 'w') as outfp:
+                    for line in infp:
+                        for subst in substs:
+                            line = line.replace(subst, "'%s'  # noqa" % (substs[subst]))
+                        outfp.write(line)
+
+        install_scripts.run(self)
+
+
 setup(
     name=package_name,
     version='0.0.0',
@@ -19,9 +62,9 @@
     install_requires=['setuptools'],
     zip_safe=True,
     author='Chris Lalancette',
-    author_email='clalancette@osrfoundation.org',
+    author_email='clalancette@openrobotics.org',
     maintainer='Chris Lalancette',
-    maintainer_email='clalancette@osrfoundation.org',
+    maintainer_email='clalancette@openrobotics.org',
     keywords=['ROS'],
     classifiers=[
         'Intended Audience :: Developers',
@@ -38,4 +81,5 @@
             'showimage_py = showimage_py:main',
         ],
     },
+    cmdclass={'install_scripts': custom_install_scripts},
 )

image_tools_py/test/cam2image_py.txt

@@ -0,0 +1,2 @@
+Publishing data on topic 'image'
+Publishing image #1

image_tools_py/test/showimage_py.regex

@@ -0,0 +1 @@
+Received image #\d+

image_tools_py/test/test_flake8.py

@@ -0,0 +1,23 @@
+# Copyright 2017 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.
+# 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.
+
+from ament_flake8.main import main
+import pytest
+
+
+@pytest.mark.flake8
+@pytest.mark.linter
+def test_flake8():
+    rc = main(argv=[])
+    assert rc == 0, 'Found errors'

image_tools_py/test/test_showimage_cam2image.py.in

@@ -0,0 +1,102 @@
+# Copyright 2016 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.
+# 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
+
+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()