Support CLI commands testing

launch_testing/launch_testing/asserts/assert_output.py

@@ -20,56 +20,12 @@
 further reference.
 """
 
-
-import os
-
 from osrf_pycommon.terminal_color import remove_ansi_escape_sequences
 
+from ..tools.text import build_text_match
 from ..util import resolveProcesses
 
 
-def normalize_lineseps(lines):
-    r"""Normalize and then return the given lines to all use '\n'."""
-    lines = lines.replace(os.linesep, '\n')
-    # This happens (even on Linux and macOS) when capturing I/O from an
-    # emulated tty.
-    lines = lines.replace('\r\n', '\n')
-    return lines
-
-
-def get_matching_function(expected_output):
-    if isinstance(expected_output, (list, tuple)):
-        if len(expected_output) > 0:
-            if isinstance(expected_output[0], str):
-                def _match(expected, actual):
-                    lines = actual.splitlines()
-                    for pattern in expected:
-                        if pattern not in lines:
-                            return False
-                        index = lines.index(pattern)
-                        lines = lines[index + 1:]
-                    return True
-                return _match
-            if hasattr(expected_output[0], 'search'):
-                def _match(expected, actual):
-                    start = 0
-                    actual = normalize_lineseps(actual)
-                    for pattern in expected:
-                        match = pattern.search(actual, start)
-                        if match is None:
-                            return False
-                        start = match.end()
-                    return True
-                return _match
-    elif isinstance(expected_output, str):
-        return lambda expected, actual: expected in actual
-    elif hasattr(expected_output, 'search'):
-        return lambda expected, actual: (
-            expected.search(normalize_lineseps(actual)) is not None
-        )
-    raise ValueError('Unknown format for expected output')
-
-
 def assertInStdout(proc_output,
                    expected_output,
                    process,
@@ -86,7 +42,7 @@ def assertInStdout(proc_output,
     :type proc_output: An launch_testing.IoHandler
 
     :param expected_output: The output to search for
-    :type expected_output: string or regex pattern or list of the aforementioned types
+    :type expected_output: string or regex pattern or a list of the aforementioned types
 
     :param process: The process whose output will be searched
     :type process: A string (search by process name) or a launch.actions.ExecuteProcess object
@@ -118,7 +74,7 @@ def assertInStdout(proc_output,
     if output_filter is not None:
         if not callable(output_filter):
             raise ValueError('output_filter is not callable')
-    output_match = get_matching_function(expected_output)
+    output_match = build_text_match(expected_output)
 
     for proc in resolved_procs:  # Nominally just one matching proc
         full_output = ''.join(
@@ -128,7 +84,7 @@ def assertInStdout(proc_output,
             full_output = remove_ansi_escape_sequences(full_output)
         if output_filter is not None:
             full_output = output_filter(full_output)
-        if output_match(expected_output, full_output):
+        if output_match(full_output) is not None:
             break
     else:
         names = ', '.join(sorted(p.process_details['name'] for p in resolved_procs))

launch_testing/launch_testing/io_handler.py

@@ -39,12 +39,14 @@ def __init__(self):
         self._sequence_list = []  # A time-ordered list of IO from all processes
         self._process_name_dict = {}  # A dict of time ordered lists of IO key'd by the process
 
+    def track(self, process_name):
+        if process_name not in self._process_name_dict:
+            self._process_name_dict[process_name] = []
+
     def append(self, process_io):
         self._sequence_list.append(process_io)
-
         if process_io.process_name not in self._process_name_dict:
             self._process_name_dict[process_io.process_name] = []
-
         self._process_name_dict[process_io.process_name].append(process_io)
 
     def __iter__(self):
@@ -56,7 +58,7 @@ def processes(self):
 
         :returns [launch.actions.ExecuteProcess]:
         """
-        return [val[0].action for val in self._process_name_dict.values()]
+        return [val[0].action for val in self._process_name_dict.values() if len(val) > 0]
 
     def process_names(self):
         """
@@ -79,7 +81,7 @@ def __getitem__(self, key):
             return list(self._process_name_dict[key.process_details['name']])
 
 
-class ActiveIoHandler(IoHandler):
+class ActiveIoHandler:
     """
     Holds stdout captured from running processes.
 
@@ -93,6 +95,15 @@ def __init__(self):
         # by composition so we can still give out the unsynchronized version
         self._io_handler = IoHandler()
 
+    @property
+    def io_event(self):
+        return self._sync_lock
+
+    def track(self, process_name):
+        with self._sync_lock:
+            self._io_handler.track(process_name)
+            self._sync_lock.notify()
+
     def append(self, process_io):
         with self._sync_lock:
             self._io_handler.append(process_io)

launch_testing/launch_testing/loader.py

@@ -80,6 +80,8 @@ def __init__(self,
                  pre_shutdown_tests,
                  post_shutdown_tests):
         self.name = name
+        if not hasattr(test_description_function, '__markers__'):
+            test_description_function.__markers__ = {}
         self._test_description_function = test_description_function
         self.normalized_test_description = _normalize_ld(test_description_function)
 
@@ -100,6 +102,10 @@ def __init__(self,
                 setattr(tc, '_testMethodName', new_name)
                 setattr(tc, new_name, test_method)
 
+    @property
+    def markers(self):
+        return self._test_description_function.__markers__
+
     def bind(self, tests, injected_attributes={}, injected_args={}):
         """
         Bind injected_attributes and injected_args to tests.
@@ -176,13 +182,16 @@ class _loader(unittest.TestLoader):
         """TestLoader selectively loads pre-shutdown or post-shutdown tests."""
 
         def loadTestsFromTestCase(self, testCaseClass):
-
             if getattr(testCaseClass, '__post_shutdown_test__', False) == load_post_shutdown:
-                cases = super(_loader, self).loadTestsFromTestCase(testCaseClass)
+                # Isolate test classes instances on a per parameterization basis
+                cases = super(_loader, self).loadTestsFromTestCase(
+                    type(testCaseClass.__name__, (testCaseClass,), {
+                        '__module__': testCaseClass.__module__
+                    })
+                )
                 return cases
-            else:
-                # Empty test suites will be ignored by the test runner
-                return self.suiteClass()
+            # Empty test suites will be ignored by the test runner
+            return self.suiteClass()
 
     return _loader()
 
@@ -224,9 +233,9 @@ def _bind_test_args_to_tests(context, test_suite):
 
 
 def _partially_bind_matching_args(unbound_function, arg_candidates):
-    function_arg_names = inspect.getfullargspec(unbound_function).args
+    function_args = inspect.signature(unbound_function).parameters
     # We only want to bind the part of the context matches the test args
-    matching_args = {k: v for (k, v) in arg_candidates.items() if k in function_arg_names}
+    matching_args = {k: v for (k, v) in arg_candidates.items() if k in function_args}
     return functools.partial(unbound_function, **matching_args)
 
 

launch_testing/launch_testing/markers.py

@@ -0,0 +1,41 @@
+# 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.
+# 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 functools
+
+
+def keep_alive(test_description):
+    """Mark a test launch description to be kept alive after fixture processes' termination."""
+    if not hasattr(test_description, '__markers__'):
+        test_description.__markers__ = {}
+    test_description.__markers__['keep_alive'] = True
+    return test_description
+
+
+def retry_on_failure(*, times):
+    """Mark a test case to be retried up to `times` on AssertionError."""
+    assert times > 0
+
+    def _decorator(func):
+        @functools.wraps(func)
+        def _wrapper(*args, **kwargs):
+            n = times
+            while n > 1:
+                try:
+                    return func(*args, **kwargs)
+                except AssertionError:
+                    n -= 1
+            return func(*args, **kwargs)
+        return _wrapper
+    return _decorator

launch_testing/launch_testing/proc_info_handler.py

@@ -70,7 +70,7 @@ def __getitem__(self, key):
             return self._proc_info[key]
 
 
-class ActiveProcInfoHandler(ProcInfoHandler):
+class ActiveProcInfoHandler:
     """Allows tests to wait on process termination before proceeding."""
 
     def __init__(self):
@@ -79,6 +79,10 @@ def __init__(self):
         # by composition so we can still give out the unsynchronized version
         self._proc_info_handler = ProcInfoHandler()
 
+    @property
+    def proc_event(self):
+        return self._sync_lock
+
     def append(self, process_info):
         with self._sync_lock:
             self._proc_info_handler.append(process_info)
@@ -126,7 +130,6 @@ def proc_is_shutdown():
                     cmd_args=cmd_args,
                     strict_proc_matching=True
                 )[0]
-
                 process_event = self._proc_info_handler[resolved_process]
                 return isinstance(process_event, ProcessExited)
             except NoMatchingProcessException:

launch_testing/launch_testing/pytest/hooks.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from unittest import TestCase
+
 import pytest
 
 from ..loader import LoadTestsFromPythonModule
@@ -64,7 +66,7 @@ def repr_failure(self, excinfo):
                 )
                 for test_run, test_result in excinfo.value.results.items()
                 for test_case, _ in (test_result.errors + test_result.failures)
-                if not test_result.wasSuccessful()
+                if isinstance(test_case, TestCase) and not test_result.wasSuccessful()
             }) if excinfo.value.results else ''
         return super().repr_failure(excinfo)
 

launch_testing/launch_testing/test_runner.py

@@ -128,6 +128,11 @@ def run(self):
             RegisterEventHandler(
                 OnProcessStart(on_start=lambda info, unused: proc_info.append(info))
             ),
+            RegisterEventHandler(
+                OnProcessStart(
+                    on_start=lambda info, unused: proc_output.track(info.process_name)
+                )
+            ),
             RegisterEventHandler(
                 OnProcessExit(on_exit=lambda info, unused: proc_info.append(info))
             ),
@@ -148,7 +153,11 @@ def run(self):
         )
 
         self._test_tr.start()  # Run the tests on another thread
-        self._launch_service.run()  # This will block until the test thread stops it
+
+        # This will block until the test thread stops it
+        self._launch_service.run(
+            shutdown_when_idle=not self._test_run.markers.get('keep_alive', False)
+        )
 
         if not self._tests_completed.wait(timeout=0):
             # LaunchService.run returned before the tests completed.  This can be because the user

launch_testing/launch_testing/tools/__init__.py

@@ -12,13 +12,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from .launchers import launch_process
 from .output import basic_output_filter
+from .output import expect_output
 from .output import expected_output_from_file
+from .process import launch_process
+from .process import ProcessProxy
 
 
 __all__ = [
     'basic_output_filter',
+    'expect_output',
     'expected_output_from_file',
     'launch_process',
+    'ProcessProxy'
 ]