Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Migrate launch tests to new launch_testing features & API #318

Merged
merged 11 commits into from Apr 30, 2019
Merged

Migrate launch tests to new launch_testing features & API #318

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
c39a58f
Update after launch_testing features becoming legacy.
hidmic Mar 25, 2019
13f9538
Migrate demos tests to new launch_testing API.
hidmic Apr 3, 2019
82eb361
Stop using injected attributes in launch tests.
hidmic Apr 12, 2019
a5b6d3c
Add launch_testing_ros as test dependency where necessary.
hidmic Apr 15, 2019
38fbbb7
Fix bad process cmd on logging demo tests.
hidmic Apr 15, 2019
02d08b9
Fix bad output filter in logging demo tests.
hidmic Apr 15, 2019
044d5fb
Address peer review comments.
hidmic Apr 26, 2019
4efe6a0
Fix failing demo_nodes_cpp launch tests
hidmic Apr 26, 2019
38eb873
Patch failing demo_nodes_cpp launch tests.
hidmic Apr 29, 2019
b7edf2a
Alpha ordered some package.xml files.
hidmic Apr 30, 2019
798105a
Extended tests for parameter events.
hidmic Apr 30, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
7 changes: 5 additions & 2 deletions composition/CMakeLists.txt
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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'
wjwwood marked this conversation as resolved.
Show resolved Hide resolved
],
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
)