Warn that old-style ready_fn and test attributes will be deprecated

launch_testing/launch_testing/loader.py

@@ -15,11 +15,22 @@
 import functools
 import inspect
 import itertools
+import os
 import unittest
+import warnings
 
 from .actions import ReadyToTest
 
 
+# Patch up the warnings module to streamline the warning messages.  See
+# https://docs.python.org/3/library/warnings.html#warnings.showwarning
+def slim_formatwarning(msg, *args, **kwargs):
+    return 'Warning: ' + str(msg) + os.linesep
+
+
+warnings.formatwarning = slim_formatwarning
+
+
 def _normalize_ld(launch_description_fn):
     # A launch description fn can return just a launch description, or a tuple of
     # (launch_description, test_context).  This wrapper function normalizes things
@@ -35,8 +46,15 @@ def wrapper(**kwargs):
         fn_args = inspect.getfullargspec(launch_description_fn)
 
         if 'ready_fn' in fn_args.args + fn_args.kwonlyargs:
-            # This is an old-style launch_description function which epects ready_fn to be passed
+            # This is an old-style launch_description function which expects ready_fn to be passed
             # in to the function
+            # This type of launch description will be deprecated in the future.  Warn about it
+            # here
+            warnings.warn(
+                'Passing ready_fn as an argument to generate_test_description will '
+                'be removed in a future release.  Include a launch_testing.actions.ReadyToTest '
+                'action in the LaunchDescription instead.'
+            )
             return normalize(launch_description_fn(**kwargs))
         else:
             # This is a new-style launch_description which should contain a ReadyToTest action
@@ -240,14 +258,23 @@ def _partially_bind_matching_args(unbound_function, arg_candidates):
 
 
 def _give_attribute_to_tests(data, attr_name, test_suite):
-    # Test suites can contain other test suites which will eventually contain
-    # the actual test classes to run.  This function will recursively drill down until
-    # we find the actual tests and give the tests a reference to the process
+
+    def _warn_getter(self):
+        if not hasattr(self, '__warned'):
+            warnings.warn(
+                'Automatically adding attributes like self.{0} '
+                'to the test class will be deprecated in a future release.  '
+                'Instead, add {0} to the test method argument list to '
+                'access the test object you need'.format(attr_name)
+            )
+            setattr(self, '__warned', True)
+
+        return data
 
     # The effect of this is that every test will have `self.attr_name` available to it so that
     # it can interact with ROS2 or the process exit coes, or IO or whatever data we want
-    for test in _iterate_tests_in_test_suite(test_suite):
-        setattr(test, attr_name, data)
+    for cls in _iterate_test_classes_in_test_suite(test_suite):
+        setattr(cls, attr_name, property(fget=_warn_getter))
 
 
 def _iterate_test_classes_in_test_suite(test_suite):

launch_testing/test/launch_testing/examples/good_proc_launch_test.py

@@ -22,6 +22,7 @@
 import launch.actions
 
 import launch_testing
+import launch_testing.actions
 from launch_testing.asserts import assertSequentialStdout
 
 import pytest
@@ -44,13 +45,13 @@
 
 
 @pytest.mark.launch_test
-def generate_test_description(ready_fn):
+def generate_test_description():
 
     return launch.LaunchDescription([
         dut_process,
 
         # Start tests right away - no need to wait for anything
-        launch.actions.OpaqueFunction(function=lambda context: ready_fn()),
+        launch_testing.actions.ReadyToTest(),
     ])
 
 

launch_testing/test/launch_testing/test_proc_info_assertions.py

@@ -20,13 +20,14 @@
 import launch
 import launch.actions
 import launch.events.process
+import launch_testing.actions
 from launch_testing.test_runner import LaunchTestRunner
 import launch_testing.util
 
 
 def test_wait_for_shutdown(source_test_loader):
 
-    def generate_test_description(ready_fn):
+    def generate_test_description():
         TEST_PROC_PATH = os.path.join(
             ament_index_python.get_package_prefix('launch_testing'),
             'lib/launch_testing',
@@ -52,7 +53,7 @@ def generate_test_description(ready_fn):
                     )
                 ]
             ),
-            launch.actions.OpaqueFunction(function=lambda context: ready_fn())
+            launch_testing.actions.ReadyToTest(),
         ]), {'good_process': good_process}
 
     # This is kind of a weird test-within-a-test, but it's the easiest way to get
@@ -82,7 +83,7 @@ def test_02_check_when_process_exits(self, proc_info, good_process):
 
 def test_wait_for_startup(source_test_loader):
 
-    def generate_test_description(ready_fn):
+    def generate_test_description():
         TEST_PROC_PATH = os.path.join(
             ament_index_python.get_package_prefix('launch_testing'),
             'lib/launch_testing',
@@ -99,7 +100,7 @@ def generate_test_description(ready_fn):
                 period=10.0,
                 actions=[good_process]
             ),
-            launch.actions.OpaqueFunction(function=lambda context: ready_fn())
+            launch_testing.actions.ReadyToTest(),
         ]), {'good_process': good_process}
 
     # This is kind of a weird test-within-a-test, but it's the easiest way to get

launch_testing/test/launch_testing/test_resolve_process.py

@@ -21,6 +21,7 @@
 import launch.actions
 import launch.substitutions
 import launch_testing
+import launch_testing.actions
 from launch_testing.loader import LoadTestsFromPythonModule
 from launch_testing.test_runner import LaunchTestRunner
 import launch_testing.util
@@ -86,7 +87,7 @@ def setUpClass(cls):
         proc_env = os.environ.copy()
         proc_env['PYTHONUNBUFFERED'] = '1'
 
-        def generate_test_description(ready_fn):
+        def generate_test_description():
             no_arg_proc = launch.actions.ExecuteProcess(
                 cmd=[sys.executable],
                 env=proc_env
@@ -106,7 +107,7 @@ def generate_test_description(ready_fn):
                 no_arg_proc,
                 one_arg_proc,
                 two_arg_proc,
-                launch.actions.OpaqueFunction(function=lambda ctx: ready_fn())
+                launch_testing.actions.ReadyToTest(),
             ])
 
             return (ld, locals())

launch_testing/test/launch_testing/test_runner_results.py

@@ -20,6 +20,7 @@
 import launch
 import launch.actions
 import launch_testing
+import launch_testing.actions
 from launch_testing.loader import TestRun as TR
 from launch_testing.test_runner import LaunchTestRunner
 import mock
@@ -29,7 +30,7 @@
 # indicate failure
 def test_dut_that_shuts_down(capsys):
 
-    def generate_test_description(ready_fn):
+    def generate_test_description():
         TEST_PROC_PATH = os.path.join(
             ament_index_python.get_package_prefix('launch_testing'),
             'lib/launch_testing',
@@ -41,7 +42,7 @@ def generate_test_description(ready_fn):
                 cmd=[sys.executable, TEST_PROC_PATH]
             ),
 
-            launch.actions.OpaqueFunction(function=lambda context: ready_fn()),
+            launch_testing.actions.ReadyToTest(),
         ])
 
     with mock.patch('launch_testing.test_runner._RunnerWorker._run_test'):
@@ -64,7 +65,7 @@ def test_dut_that_has_exception(capsys):
     # This is the same as above, but we also want to check we get extra output from processes
     # that had an exit code
 
-    def generate_test_description(ready_fn):
+    def generate_test_description():
         TEST_PROC_PATH = os.path.join(
             ament_index_python.get_package_prefix('launch_testing'),
             'lib/launch_testing',
@@ -88,7 +89,7 @@ def generate_test_description(ready_fn):
                 cmd=[sys.executable, EXIT_PROC_PATH, '--silent']
             ),
 
-            launch.actions.OpaqueFunction(function=lambda context: ready_fn()),
+            launch_testing.actions.ReadyToTest(),
         ])
 
     with mock.patch('launch_testing.test_runner._RunnerWorker._run_test'):
@@ -119,13 +120,13 @@ def test_ok(self):
         'good_proc'
     )
 
-    def generate_test_description(ready_fn):
+    def generate_test_description():
         return launch.LaunchDescription([
             launch.actions.ExecuteProcess(
                 cmd=[sys.executable, TEST_PROC_PATH]
             ),
 
-            launch.actions.OpaqueFunction(function=lambda context: ready_fn()),
+            launch_testing.actions.ReadyToTest(),
         ])
 
     runner = LaunchTestRunner(
@@ -146,7 +147,7 @@ def test_parametrized_run_with_one_failure(source_test_loader):
 
     # Test Data
     @launch_testing.parametrize('arg_val', [1, 2, 3, 4, 5])
-    def generate_test_description(arg_val, ready_fn):
+    def generate_test_description(arg_val):
         TEST_PROC_PATH = os.path.join(
             ament_index_python.get_package_prefix('launch_testing'),
             'lib/launch_testing',
@@ -162,7 +163,7 @@ def generate_test_description(arg_val, ready_fn):
                 cmd=[sys.executable, TEST_PROC_PATH],
                 env=proc_env,
             ),
-            launch.actions.OpaqueFunction(function=lambda context: ready_fn())
+            launch_testing.actions.ReadyToTest(),
         ])
 
     def test_fail_on_two(self, proc_output, arg_val):
@@ -192,7 +193,7 @@ def test_fail_on_three(self, arg_val):
 def test_skipped_launch_description(source_test_loader):
 
     @unittest.skip('skip reason string')
-    def generate_test_description(ready_fn):
+    def generate_test_description():
         raise Exception('This should never be invoked')  # pragma: no cover
 
     def test_fail_always(self):

launch_testing/test/launch_testing/test_xml_output.py

@@ -99,7 +99,7 @@ class TestHost(unittest.TestCase):
 
     def test_fail_results_serialize(self):
 
-        def generate_test_description(ready_fn):
+        def generate_test_description():
             raise Exception('This should never be invoked')  # pragma: no cover
 
         def test_fail_always(self):
@@ -140,7 +140,7 @@ def test_skip_results_serialize(self):
         # This checks the case where all unit tests are skipped because of a skip
         # decorator on the generate_test_description function
         @unittest.skip('skip reason string')
-        def generate_test_description(ready_fn):
+        def generate_test_description():
             raise Exception('This should never be invoked')  # pragma: no cover
 
         def test_fail_always(self):