Skip to content

Commit

Permalink
Migrate launch tests to new launch_testing features & API (#318)
Browse files Browse the repository at this point in the history 
* Update after launch_testing features becoming legacy.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Migrate demos tests to new launch_testing API.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Stop using injected attributes in launch tests.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Add launch_testing_ros as test dependency where necessary.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Fix bad process cmd on logging demo tests.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Fix bad output filter in logging demo tests.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Address peer review comments.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Fix failing demo_nodes_cpp launch tests

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Patch failing demo_nodes_cpp launch tests.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Alpha ordered some package.xml files.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

* Extended tests for parameter events.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>
  • Loading branch information
@hidmic
hidmic authored Apr 30, 2019
1 parent d2a934b commit 09b6291
Show file tree
Hide file tree
Showing 31 changed files with 639 additions and 646 deletions.
7 changes: 5 additions & 2 deletions composition/CMakeLists.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,7 @@ if(BUILD_TESTING)
ament_lint_auto_find_test_dependencies()

find_package(ament_cmake_pytest REQUIRED)
find_package(launch_testing_ament_cmake REQUIRED)
find_package(rmw_implementation_cmake REQUIRED)

file(GENERATE
Expand Down Expand Up @@ -165,12 +166,14 @@ if(BUILD_TESTING)
OUTPUT "${CMAKE_CURRENT_BINARY_DIR}/${test_name}${target_suffix}_$<CONFIG>.py"
INPUT "${CMAKE_CURRENT_BINARY_DIR}/${test_name}${target_suffix}.py.genexp"
)
ament_add_pytest_test(${test_name}${target_suffix}
add_launch_test(
"${CMAKE_CURRENT_BINARY_DIR}/${test_name}${target_suffix}_$<CONFIG>.py"
TARGET ${test_name}${target_suffix}
ENV RMW_IMPLEMENTATION=${rmw_implementation}
APPEND_ENV AMENT_PREFIX_PATH=${CMAKE_CURRENT_BINARY_DIR}/test_ament_index/$<CONFIG>
APPEND_LIBRARY_DIRS "${append_library_dirs}"
TIMEOUT 60)
TIMEOUT 60
)
list(
APPEND generated_python_files
"${CMAKE_CURRENT_BINARY_DIR}/${test_name}${target_suffix}_$<CONFIG>.py")
Expand Down
2 changes: 2 additions & 0 deletions composition/package.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
<test_depend>ament_lint_common</test_depend>
<test_depend>launch</test_depend>
<test_depend>launch_testing</test_depend>
<test_depend>launch_testing_ros</test_depend>
<test_depend>launch_testing_ament_cmake</test_depend>
<test_depend>rmw_implementation_cmake</test_depend>

<export>
Expand Down
93 changes: 42 additions & 51 deletions composition/test/test_api_pubsub_composition.py.in
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,67 +12,58 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import unittest

from launch import LaunchDescription
from launch import LaunchService
from launch.actions import ExecuteProcess
from launch_testing import LaunchTestService
from launch_testing.output import create_output_lines_filter
from launch_testing.output import create_output_test_from_file

from launch.actions import OpaqueFunction

def test_api_pubsub_composition():
name = 'test_api_composition'
executable = '@API_COMPOSITION_EXECUTABLE@'
additional_processes = [
{
'cmd': [
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Talker'],
'name': 'load_talker_component',
},
{
'cmd': [
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Listener'],
'name': 'load_listener_component',
}
]
output_file = '@EXPECTED_OUTPUT_PUBSUB@'
launch(name, [executable], output_file, additional_processes=additional_processes)
import launch_testing
import launch_testing.asserts
import launch_testing_ros


def launch(name, cmd, output_file, additional_processes=None):
launch_test = LaunchTestService()
def generate_test_description(ready_fn):
launch_description = LaunchDescription()

action = launch_test.add_fixture_action(
launch_description, ExecuteProcess(
cmd=cmd,
name=name,
output='screen'
)
process_under_test = ExecuteProcess(
cmd=['@API_COMPOSITION_EXECUTABLE@'],
name='test_api_composition',
output='screen'
)
rmw_implementation = '@rmw_implementation@'
launch_test.add_output_test(
launch_description, action,
output_test=create_output_test_from_file(output_file),
output_filter=create_output_lines_filter(
filtered_rmw_implementation=rmw_implementation
launch_description.add_action(process_under_test)
launch_description.add_action(
ExecuteProcess(
cmd=[
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Talker'
],
name='load_talker_component'
)
)
for additional_process in (additional_processes or []):
launch_test.add_fixture_action(
launch_description, ExecuteProcess(**additional_process)
launch_description.add_action(
ExecuteProcess(
cmd=[
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Listener'
],
name='load_listener_component'
)
)
launch_description.add_action(
OpaqueFunction(function=lambda context: ready_fn())
)
return launch_description, locals()

launch_service = LaunchService()
launch_service.include_launch_description(launch_description)
rc = launch_test.run(launch_service)

assert rc == 0, \
"The launch file failed with exit code '" + str(rc) + "'. " \
'Maybe the client did not receive any messages?'

class TestComposition(unittest.TestCase):

if __name__ == '__main__':
test_api_pubsub_composition()
def test_api_pubsub_composition(self, proc_output, process_under_test):
"""Test process' output against expectations."""
output_filter = launch_testing_ros.tools.basic_output_filter(
filtered_rmw_implementation='@rmw_implementation@'
)
proc_output.assertWaitFor(
expected_output=launch_testing.tools.expected_output_from_file(
path='@EXPECTED_OUTPUT_PUBSUB@'
), process=process_under_test, output_filter=output_filter, timeout=10
)
93 changes: 42 additions & 51 deletions composition/test/test_api_srv_composition.py.in
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,67 +12,58 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import unittest

from launch import LaunchDescription
from launch import LaunchService
from launch.actions import ExecuteProcess
from launch_testing import LaunchTestService
from launch_testing.output import create_output_lines_filter
from launch_testing.output import create_output_test_from_file

from launch.actions import OpaqueFunction

def test_api_srv_composition():
name = 'test_api_composition'
executable = '@API_COMPOSITION_EXECUTABLE@'
additional_processes = [
{
'cmd': [
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Server'],
'name': 'load_server_component',
},
{
'cmd': [
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Client'],
'name': 'load_client_component',
},
]
output_file = '@EXPECTED_OUTPUT_SRV@'
launch(name, [executable], output_file, additional_processes=additional_processes)
import launch_testing
import launch_testing.asserts
import launch_testing_ros


def launch(name, cmd, output_file, additional_processes=None):
launch_test = LaunchTestService()
def generate_test_description(ready_fn):
launch_description = LaunchDescription()

action = launch_test.add_fixture_action(
launch_description, ExecuteProcess(
cmd=cmd,
name=name,
output='screen'
)
process_under_test = ExecuteProcess(
cmd=['@API_COMPOSITION_EXECUTABLE@'],
name='test_api_composition',
output='screen'
)
rmw_implementation = '@rmw_implementation@'
launch_test.add_output_test(
launch_description, action,
output_test=create_output_test_from_file(output_file),
output_filter=create_output_lines_filter(
filtered_rmw_implementation=rmw_implementation
launch_description.add_action(process_under_test)
launch_description.add_action(
ExecuteProcess(
cmd=[
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Server'
],
name='load_server_component'
)
)
for additional_process in (additional_processes or []):
launch_test.add_fixture_action(
launch_description, ExecuteProcess(**additional_process)
launch_description.add_action(
ExecuteProcess(
cmd=[
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Client'
],
name='load_client_component'
)
)
launch_description.add_action(
OpaqueFunction(function=lambda context: ready_fn())
)
return launch_description, locals()

launch_service = LaunchService()
launch_service.include_launch_description(launch_description)
rc = launch_test.run(launch_service)

assert rc == 0, \
"The launch file failed with exit code '" + str(rc) + "'. " \
'Maybe the client did not receive any messages?'

class TestComposition(unittest.TestCase):

if __name__ == '__main__':
test_api_srv_composition()
def test_api_srv_composition(self, proc_output, process_under_test):
"""Test process' output against expectations."""
output_filter = launch_testing_ros.tools.basic_output_filter(
filtered_rmw_implementation='@rmw_implementation@'
)
proc_output.assertWaitFor(
expected_output=launch_testing.tools.expected_output_from_file(
path='@EXPECTED_OUTPUT_SRV@'
), process=process_under_test, output_filter=output_filter, timeout=10
)
96 changes: 43 additions & 53 deletions composition/test/test_api_srv_composition_client_first.py.in
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,69 +12,59 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import unittest

from launch import LaunchDescription
from launch import LaunchService
from launch.actions import ExecuteProcess
from launch_testing import LaunchTestService
from launch_testing.output import create_output_lines_filter
from launch_testing.output import create_output_test_from_file

from launch.actions import OpaqueFunction

def test_api_srv_composition_client_first():
# This is a regression test to ensure that the client starting first and waiting for a service
# does not block the executor from processing requests to load other components.
name = 'test_api_composition'
executable = '@API_COMPOSITION_EXECUTABLE@'
additional_processes = [
{
'cmd': [
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Client'],
'name': 'load_client_component',
},
{
'cmd': [
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Server', '--delay', '5000'],
'name': 'load_server_component',
},
]
output_file = '@EXPECTED_OUTPUT_SRV@'
launch(name, [executable], output_file, additional_processes=additional_processes)
import launch_testing
import launch_testing.asserts
import launch_testing_ros


def launch(name, cmd, output_file, additional_processes=None):
launch_test = LaunchTestService()
def generate_test_description(ready_fn):
launch_description = LaunchDescription()

action = launch_test.add_fixture_action(
launch_description, ExecuteProcess(
cmd=cmd,
name=name,
output='screen'
)
process_under_test = ExecuteProcess(
cmd=['@API_COMPOSITION_EXECUTABLE@'],
name='test_api_composition',
output='screen'
)
rmw_implementation = '@rmw_implementation@'
launch_test.add_output_test(
launch_description, action,
output_test=create_output_test_from_file(output_file),
output_filter=create_output_lines_filter(
filtered_rmw_implementation=rmw_implementation
launch_description.add_action(process_under_test)
launch_description.add_action(
ExecuteProcess(
cmd=[
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Client'
],
name='load_client_component'
)
)
for additional_process in (additional_processes or []):
launch_test.add_fixture_action(
launch_description, ExecuteProcess(**additional_process)
launch_description.add_action(
ExecuteProcess(
cmd=[
'@API_COMPOSITION_CLI_EXECUTABLE@',
'composition', 'composition::Server',
'--delay', '5000'
],
name='load_server_component'
)
)
launch_description.add_action(
OpaqueFunction(function=lambda context: ready_fn())
)
return launch_description, locals()

launch_service = LaunchService()
launch_service.include_launch_description(launch_description)
rc = launch_test.run(launch_service)

assert rc == 0, \
"The launch file failed with exit code '" + str(rc) + "'. " \
'Maybe the client did not receive any messages?'

class TestComposition(unittest.TestCase):

if __name__ == '__main__':
test_api_srv_composition_client_first()
def test_api_srv_composition_client_first(self, proc_output, process_under_test):
"""Test process' output against expectations."""
output_filter = launch_testing_ros.tools.basic_output_filter(
filtered_rmw_implementation='@rmw_implementation@'
)
proc_output.assertWaitFor(
expected_output=launch_testing.tools.expected_output_from_file(
path='@EXPECTED_OUTPUT_SRV@'
), process=process_under_test, output_filter=output_filter, timeout=15
)
Loading

0 comments on commit 09b6291

Please sign in to comment.