Add auto_shutdown, shutdown_when_idle options

Signed-off-by: Ivan Santiago Paunovic <[email protected]>

Author: ivanpauno

launch_testing/launch_testing/pytest/fixture.py

@@ -27,8 +27,9 @@ def scope_gt(scope1, scope2):
     from _pytest.fixtures import scopemismatch as scope_gt
 
 
-def finalize_launch_service(launch_service, eprefix=''):
-    launch_service.shutdown(force_sync=True)
+def finalize_launch_service(launch_service, eprefix='', auto_shutdown=True):
+    if auto_shutdown:
+        launch_service.shutdown(force_sync=True)
     loop = launch_service.event_loop
     if loop is not None and not loop.is_closed():
         rc = loop.run_until_complete(launch_service.task)
@@ -65,12 +66,26 @@ def event_loop():
     return event_loop
 
 
-def fixture(decorated = None, *args, **kwargs):
+def fixture(
+    decorated = None,
+    *args,
+    shutdown_when_idle = True,
+    auto_shutdown = True,
+    **kwargs
+):
     """
     Decorate launch_test fixtures.
 
-    For documentation on the supported arguments, see
+    See also
     https://docs.pytest.org/en/latest/reference/reference.html#pytest-fixture.
+
+    :param decorated: object to be decorated.
+    :param \*args: extra posicional arguments to be passed to pytest.fixture().
+    :param shutdown_when_idle: when true, the launch service will shutdown when idle.
+    :param auto_shutdown: when true, the launch service will be shutdown automatically
+        after all pre-shutdown tests get run. If false, shutdown needs to be signaled in a
+        different way or the launch fixture should be self terminating.
+    :param \**kwargs: extra keyword arguments to be passed to pytest.fixture().
     """
     # Automagically override the event_loop and launch_testing fixtures
     # with a fixture of the correct scope.
@@ -97,6 +112,10 @@ def fixture(decorated = None, *args, **kwargs):
 
     def decorator(fixture_function):
         fixture_function._launch_pytest_fixture = True
+        fixture_function._launch_pytest_fixture_options = {
+            'shutdown_when_idle': shutdown_when_idle,
+            'auto_shutdown': auto_shutdown,
+        }
         return pytest.fixture(fixture_function, *args, **kwargs)
     if decorated is None:
         return decorator

launch_testing/launch_testing/pytest/plugin.py

@@ -96,6 +96,7 @@ def get_ready_to_test_action(launch_description):
 def pytest_fixture_setup(fixturedef, request):
     """Set up launch service for all launch_pytest fixtures."""
     if getattr(fixturedef.func, '_launch_pytest_fixture', False):
+        options = fixturedef.func._launch_pytest_fixture_options
         eprefix = f"When running launch_pytest fixture '{fixturedef.func.__name__}':"
         ls = request.getfixturevalue('launch_service')
         event_loop = request.getfixturevalue('event_loop')
@@ -114,14 +115,14 @@ def pytest_fixture_setup(fixturedef, request):
         ls.include_launch_description(ld)
         run_async_task = event_loop.create_task(ls.run_async(
             # TODO(ivanpauno): maybe this could be configurable (?)
-            shutdown_when_idle=True
+            shutdown_when_idle=options['shutdown_when_idle']
         ))
         ready = get_ready_to_test_action(ld)
         asyncio.set_event_loop(event_loop)
         event = asyncio.Event()
         ready._add_callback(lambda: event.set())
-
-        fixturedef.addfinalizer(functools.partial(finalize_launch_service, ls, eprefix=eprefix))
+        fixturedef.addfinalizer(functools.partial(
+            finalize_launch_service, ls, eprefix=eprefix, auto_shutdown=options['auto_shutdown']))
         run_until_complete(event_loop, event.wait())
         # this is guaranteed by the current run_async() implementation, let's check it just in case
         # it changes in the future
@@ -343,8 +344,12 @@ def pytest_pyfunc_call(pyfuncitem):
     scope = fixture._pytestfixturefunction.scope
     event_loop = pyfuncitem.funcargs['event_loop']
     ls = pyfuncitem.funcargs['launch_service']
+    auto_shutdown = fixture._launch_pytest_fixture_options['auto_shutdown']
     on_shutdown = functools.partial(
-        finalize_launch_service, ls, eprefix=f'When running test {func.__name__}')
+        finalize_launch_service,
+        ls,
+        eprefix=f'When running test {func.__name__}',
+        auto_shutdown=auto_shutdown)
     before_test = on_shutdown if shutdown_test else None
     if inspect.iscoroutinefunction(func):
         pyfuncitem.obj = wrap_coroutine(func, event_loop, before_test)