Hook up launch tests in pure Python packages

[hidmic]

Feature request

Feature description

As a follow-up of ros2/launch_ros#15 (comment), we currently lack a way to register launch tests to be run in pure Python packages. We need a way to do so.

Implementation considerations

As of #236, launch_test can now be run from Python code. We'd just be lacking a pytest plugin to deal with launch test files properly.

[dirk-thomas]

We'd just be lacking a pytest plugin to deal with launch test files properly.

Why can't the launch file be invoked like any other test in a Python package?

[hidmic]

Because a launch test is (currently) not just another pytest and thus it cannot be picked up by it as-is.

[hidmic]

Put in another way, the launch_test command replaces pytest entirely.

[dirk-thomas]

What would be the benefit over running the launch file as a normal unit test using pytest?

[cevans87]

Put in another way, the launch_test command replaces pytest entirely.

I was surprised to discover that launch_testing was an entirely different testing framework when I tried running demo_nodes_cpp tests. Pytest flags had no effect when I expected they would. For example, I wanted to run a specific test rather than all launch tests for a package
colcon test --python-testing pytest --packages-select demo_nodes_cpp --pytest-args -k <test_name>

I also tried generating an html report using pytest-html
colcon test --python-testing pytest --packages-select demo_nodes_cpp --pytest-args --html /results.html

I haven't been able to figure out why launch_testing needs to be in control of test execution rather than a helper library that can be used from within unittest.main or pytest. If there's a good rationale, do you think we should include it in the readme?

[ivanpauno]

I haven't been able to figure out why launch_testing needs to be in control of test execution rather than a helper library that can be used from within unittest.main or pytest. If there's a good rationale, do you think we should include it in the readme?

I think that the root cause is that a LaunchService can't be run from a thread different of the main thread (see #126). I think #210 was an attempt of working-around that limitation. Probably, @hidmic have a better answer.

[hidmic]

I think that the root cause is that a LaunchService can't be run from a thread different of the main thread

That's correct. Tests can't be run in the main thread by any of the existing testing frameworks as launch's asyncio loop has to, and IIRC at the time launch_testing codebase was written, significant changes to launch would've been necessary to allow for anything else. As @ivanpauno points out, #210 was a start in that direction.

But I'm sure @pbaughman can provide further (and better) insight.

[pbaughman]

The main reason was I used 'rostest` from ros1 as a starting point for what eventually became launch_testing. The problem statement I was working off of was "We want to write integration tests that use ros2 launch files" Personally, I also wanted more visibility into process exit codes because one of my frustrations with rostest at a previous company was a lot of times processes under test would die and restart and the test wouldn't necessarily notice. Other people have contributed now and the tool does more than that, but that was the starting ponit

There's probably a way to make launch_testing a pytest plugin, but I don't know how to do it off the top of my head, and my contributions to this tool was work done for a 3rd party on an hourly basis, so I didn't feel great about spending extra time figuring it out. The fact that we need to run the launch on the main thread and put the tests on a background thread is probably the biggest obstacle.

Maybe an easier way to make this work like you expect is to run pytests in another process along side the processes under test. There was a little bit of work done to have test actions that could run 'pytest' or 'gtest' concurrently, but in their own process. This would be a little closer to the way things worked in rostest - and we could probably plumb arguments into them so you can use the pytest args you're familiar with. You wouldn't be able to see the exit codes and stdout from the other peer processes, but you may not need that.

[cevans87]

@ivanpauno @hidmic @pbaughman Thanks for explaining.

I'm somewhat motivated to see these tests work with pytest or unittest so I'm digging into this. I just tried playing around with running a LaunchService from within a non-main thread in pytest to get a better idea what the issues might be. I ran into these problems, and possible solutions.


1. Instantiating a LaunchService() in any child thread fails with ValueError: signal only works in main thread if install_signal_handlers() isn’t called first from the main thread. Solution is to call install_signal_handlers() from the main thread before it can be called from an child thread.

2. Calling a LaunchService.run() in a child thread raises RuntimeError: There is no current event loop in thread. Solution is to call asyncio.set_event_loop(asyncio.new_event_loop()) in a child thread at the beginning of a test and call asyncio.set_event_loop(None) at the end of the test.
3. Instantiating a subprocess in a child thread raises RuntimeError: Cannot add child handler, the child watcher does not have a loop attached. Solution is to call asyncio.get_child_watcher() once from the main thread before calling LaunchService.run() in a child thread.

All of these solutions can be taken care of from a unittest or pytest setup/teardown and shouldn't look too dirty.

Here’s the example I was able to get working.

import asyncio
from asyncio import get_event_loop, new_event_loop, set_event_loop
import pytest

from launch import LaunchService
import launch_ros
import launch
from launch.utilities import install_signal_handlers


def generate_launch_description():
    server = launch_ros.actions.Node(
        package='demo_nodes_cpp', node_executable='add_two_ints_server', output='screen')
    client = launch_ros.actions.Node(
        package='demo_nodes_cpp', node_executable='add_two_ints_client', output='screen')
    return launch.LaunchDescription([
        server,
        client,
        # TODO(wjwwood): replace this with a `required=True|False` option on ExecuteProcess().
        # Shutdown launch when client exits.
        launch.actions.RegisterEventHandler(
            event_handler=launch.event_handlers.OnProcessExit(
                target_action=client,
                on_exit=[launch.actions.EmitEvent(event=launch.events.Shutdown())],
            )),
    ])


@pytest.mark.asyncio
async def test_launch_service_in_executor() -> None:
    from threading import current_thread, main_thread
    assert current_thread() is main_thread()

    # Must be called from the main thread first. Subsequent calls do nothing.
    install_signal_handlers()

    # Must be called from main thread before any child thread spawns an asyncio subprocess.
    asyncio.get_child_watcher()

    def launch_service_in_child_thread():
        assert current_thread() is not main_thread()

        # set up event loop
        set_event_loop(new_event_loop())

        launch_service = LaunchService(debug=True)
        launch_service.include_launch_description(generate_launch_description())
        launch_service.run()
        launch_service.shutdown()

        # destroy event loop
        set_event_loop(None)

    # Try running the whole thing a few times to make sure additional LaunchService setup
    # and teardown works.
    for _ in range(5):
        await get_event_loop().run_in_executor(None, launch_service_in_child_thread)

I think I can get the equivalent test to work in a non-async test function, but have an additional awkward asyncio-related exception to figure out there.

I’m not super familiar with ros2/launch yet, so I’m likely missing some additional problems. Can you take a look at this and see if you remember any other issues?

[pbaughman]

Off the top of my head, you would also need a way to:

• run test cases concurrently with the launch that can interact with the launched processes
• capture the information from a finished launch service and make that information available to other test cases that run after everything is shut down
• give individual test cases access to the launch actions in the launch description
• wait for output from launched processes
• wait for launched processes to terminate
• coordinate running the test cases with some action in the launch description
• Use parameterization to create multiple launch services (maybe you get this for free, depending on how this works as a pytest plugin)
• Stop a launch service that isn't set up to launch on its own after all the tests are done

Some things that we can't do now that we'd like to be able to do

• Give tests access to the launch context
• Be able to use a freshly launched launch service for each test case, or re-use the same launch service for multiple test cases (right now you can only do the 2nd one)

[pbaughman]

More background: when I imagined how a pytest plugin would look, it would be something like

def generate_launch_description_fn():
    return launch.LaunchDescription(
        action1,
        action2,
        action_that_signals_its_ok_to_start_test
    )

@launch_testing.launch_test(generate_launch_description_fn)
def test_case_1(test_context, test_context2, action1):  # <---Arguments automatically populated by framework
    assert(. . .)
    assert(. . .)

@launch_testing.after_shutdown_test(generate_launch_description_fn)
def test2(process_output, action1):
    assert "ok" in process_output[action1]

I think during test discovery, it could group all the tests that use the same 'generate_launch_description' function, and run them all on the same set of processes, or there could be an argument to the decorator that says 'I want a fresh launch for this test'