AS: adding first phase of orientation

This commit is contained in:
Alexander Schaefer
2025-01-29 09:58:44 +00:00
parent 79001dc331
commit 45650caa1b
5106 changed files with 582827 additions and 0 deletions

View File

@@ -0,0 +1,44 @@
# Copyright 2021 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 unittest
import launch
import launch.actions
import launch_ros.actions
import launch_testing.actions
import launch_testing.markers
from launch_testing_ros import WaitForTopics
import pytest
from std_msgs.msg import String
@pytest.mark.launch_test
@launch_testing.markers.keep_alive
def generate_test_description():
return launch.LaunchDescription([
launch_ros.actions.Node(
executable='talker',
package='demo_nodes_cpp',
name='demo_node_1'
),
launch_testing.actions.ReadyToTest()
])
class TestFixture(unittest.TestCase):
def test_check_if_msgs_published(self):
with WaitForTopics([('chatter', String)], timeout=15.0):
print('Topic received messages !')

View File

@@ -0,0 +1,152 @@
# Copyright 2021 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 random
import string
import time
import unittest
import launch
import launch.actions
import launch_ros.actions
import launch_testing.actions
import launch_testing.markers
import pytest
import rclpy
from rclpy.node import Node
@pytest.mark.launch_test
@launch_testing.markers.keep_alive
def generate_test_description():
launch_actions = []
node_names = []
for i in range(3):
launch_actions.append(
launch_ros.actions.Node(
executable='talker',
package='demo_nodes_cpp',
name='demo_node_' + str(i)
)
)
node_names.append('demo_node_' + str(i))
launch_actions.append(launch_testing.actions.ReadyToTest())
return launch.LaunchDescription(launch_actions), {'node_list': node_names}
class CheckMultipleNodesLaunched(unittest.TestCase):
def test_nodes_successful(self, node_list):
"""Check if all the nodes were launched correctly."""
# Method 1
wait_for_nodes_1 = WaitForNodes(node_list, timeout=5.0)
assert wait_for_nodes_1.wait()
assert wait_for_nodes_1.get_nodes_not_found() == set()
wait_for_nodes_1.shutdown()
# Method 2
with WaitForNodes(node_list, timeout=5.0) as wait_for_nodes_2:
print('All nodes were found !')
assert wait_for_nodes_2.get_nodes_not_found() == set()
def test_node_does_not_exist(self, node_list):
"""Insert a invalid node name that should not exist."""
invalid_node_list = node_list + ['invalid_node']
# Method 1
wait_for_nodes_1 = WaitForNodes(invalid_node_list, timeout=1.0)
assert not wait_for_nodes_1.wait()
assert wait_for_nodes_1.get_nodes_not_found() == {'invalid_node'}
wait_for_nodes_1.shutdown()
# Method 2
with pytest.raises(RuntimeError):
with WaitForNodes(invalid_node_list, timeout=1.0):
pass
# TODO (adityapande-1995): Move WaitForNodes implementation to launch_testing_ros
# after https://github.com/ros2/rclpy/issues/831 is resolved
class WaitForNodes:
"""
Wait to discover supplied nodes.
Example usage:
--------------
# Method 1, calling wait() and shutdown() manually
def method_1():
node_list = ['foo', 'bar']
wait_for_nodes = WaitForNodes(node_list, timeout=5.0)
assert wait_for_nodes.wait()
print('Nodes found!')
assert wait_for_nodes.get_nodes_not_found() == set()
wait_for_nodes.shutdown()
# Method 2, using the 'with' keyword
def method_2():
with WaitForNodes(['foo', 'bar'], timeout=5.0) as wait_for_nodes:
assert wait_for_nodes.get_nodes_not_found() == set()
print('Nodes found!')
"""
def __init__(self, node_names, timeout=5.0):
self.node_names = node_names
self.timeout = timeout
self.__ros_context = rclpy.Context()
rclpy.init(context=self.__ros_context)
self._prepare_node()
self.__expected_nodes_set = set(node_names)
self.__nodes_found = None
def _prepare_node(self):
self.__node_name = '_test_node_' +\
''.join(random.choices(string.ascii_uppercase + string.digits, k=10))
self.__ros_node = Node(node_name=self.__node_name, context=self.__ros_context)
def wait(self):
start = time.time()
flag = False
print('Waiting for nodes')
while time.time() - start < self.timeout and not flag:
flag = all(name in self.__ros_node.get_node_names() for name in self.node_names)
time.sleep(0.3)
self.__nodes_found = set(self.__ros_node.get_node_names())
self.__nodes_found.remove(self.__node_name)
return flag
def shutdown(self):
self.__ros_node.destroy_node()
rclpy.shutdown(context=self.__ros_context)
def __enter__(self):
if not self.wait():
raise RuntimeError('Did not find all nodes !')
return self
def __exit__(self, exep_type, exep_value, trace):
if exep_type is not None:
raise Exception('Exception occured, value: ', exep_value)
self.shutdown()
def get_nodes_found(self):
return self.__nodes_found
def get_nodes_not_found(self):
return self.__expected_nodes_set - self.__nodes_found

View File

@@ -0,0 +1,61 @@
# Copyright 2021 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 time
import unittest
import launch
import launch.actions
import launch_ros.actions
import launch_testing.actions
import launch_testing.markers
import pytest
import rclpy
from rclpy.node import Node
@pytest.mark.launch_test
@launch_testing.markers.keep_alive
def generate_test_description():
return launch.LaunchDescription([
launch.actions.TimerAction(
period=5.0,
actions=[
launch_ros.actions.Node(
executable='talker',
package='demo_nodes_cpp',
name='demo_node_1'
),
]),
launch_testing.actions.ReadyToTest()
])
class TestFixture(unittest.TestCase):
def setUp(self):
rclpy.init()
self.node = Node('test_node')
def tearDown(self):
self.node.destroy_node()
rclpy.shutdown()
def test_node_start(self, proc_output):
start = time.time()
found = False
while time.time() - start < 20.0 and not found:
found = 'demo_node_1' in self.node.get_node_names()
time.sleep(0.1)
assert found, 'Node not found !'

View File

@@ -0,0 +1,58 @@
# Copyright 2021 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 unittest
import launch
import launch.actions
import launch_testing.actions
import launch_testing.markers
import pytest
# This function specifies the processes to be run for our test
@pytest.mark.launch_test
@launch_testing.markers.keep_alive
def generate_test_description():
"""Launch a simple process to print 'hello_world'."""
return launch.LaunchDescription([
# Launch a process to test
launch.actions.ExecuteProcess(
cmd=['echo', 'hello_world'],
shell=True
),
# Tell launch to start the test
launch_testing.actions.ReadyToTest()
])
# This is our test fixture. Each method is a test case.
# These run alongside the processes specified in generate_test_description()
class TestHelloWorldProcess(unittest.TestCase):
def test_read_stdout(self, proc_output):
"""Check if 'hello_world' was found in the stdout."""
# 'proc_output' is an object added automatically by the launch_testing framework.
# It captures the outputs of the processes launched in generate_test_description()
# Refer to the documentation for further details.
proc_output.assertWaitFor('hello_world', timeout=10, stream='stdout')
# These tests are run after the processes in generate_test_description() have shutdown.
@launch_testing.post_shutdown_test()
class TestHelloWorldShutdown(unittest.TestCase):
def test_exit_codes(self, proc_info):
"""Check if the processes exited normally."""
launch_testing.asserts.assertExitCodes(proc_info)

View File

@@ -0,0 +1,84 @@
# Copyright 2021 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 os
import shutil
import tempfile
import time
import unittest
import launch
import launch.actions
import launch_ros.actions
import launch_testing.actions
import launch_testing.markers
import pytest
import yaml
@pytest.mark.launch_test
@launch_testing.markers.keep_alive
def generate_test_description():
rosbag_dir = os.path.join(tempfile.mkdtemp(), 'test_bag')
node_list = [
launch_ros.actions.Node(
executable='talker',
package='demo_nodes_cpp',
name='demo_node'
),
launch.actions.ExecuteProcess(
cmd=['ros2', 'bag', 'record', '-a', '-o', rosbag_dir],
output='screen'
),
launch_testing.actions.ReadyToTest()
]
return launch.LaunchDescription(node_list), {'rosbag_dir': rosbag_dir}
class DelayShutdown(unittest.TestCase):
def test_delay(self):
"""Delay the shutdown of processes so that rosbag can record some messages."""
time.sleep(3)
# TODO : Test fails on windows, to be fixed
# https://github.com/ros2/rosbag2/issues/926
if os.name != 'nt':
@launch_testing.post_shutdown_test()
class TestFixtureAfterShutdown(unittest.TestCase):
rosbag_dir = None
def test_rosbag_record(self, rosbag_dir):
"""Check if the rosbag2 recording was successful."""
with open(os.path.join(rosbag_dir, 'metadata.yaml'), 'r') as file:
metadata = yaml.safe_load(file)
assert metadata['rosbag2_bagfile_information']['message_count'] > 0
print('The following topics received messages:')
for item in metadata['rosbag2_bagfile_information']['topics_with_message_count']:
print(item['topic_metadata']['name'], 'recieved ', item['message_count'],
' messages')
TestFixtureAfterShutdown.rosbag_dir = rosbag_dir
@classmethod
def tearDownClass(cls):
"""Delete the rosbag directory."""
print('Deleting ', cls.rosbag_dir)
shutil.rmtree(cls.rosbag_dir.replace('test_bag', ''))

View File

@@ -0,0 +1,68 @@
# Copyright 2021 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 os
import unittest
import launch
import launch.actions
import launch_ros.actions
import launch_testing.actions
import launch_testing.markers
import pytest
from rcl_interfaces.srv import SetParameters
import rclpy
from rclpy.node import Node
@pytest.mark.launch_test
@launch_testing.markers.keep_alive
def generate_test_description():
return launch.LaunchDescription([
launch_ros.actions.Node(
executable='parameter_blackboard',
package='demo_nodes_cpp',
name='demo_node_1'
),
launch_testing.actions.ReadyToTest()
])
# TODO: Fix windows failures for this test
if os.name != 'nt':
class TestFixture(unittest.TestCase):
def setUp(self):
rclpy.init()
self.node = Node('test_node')
def tearDown(self):
self.node.destroy_node()
rclpy.shutdown()
def test_set_parameter(self, proc_output):
parameters = [rclpy.Parameter('demo_parameter_1', value=True).to_parameter_msg()]
client = self.node.create_client(SetParameters, 'demo_node_1/set_parameters')
ready = client.wait_for_service(timeout_sec=15.0)
if not ready:
raise RuntimeError('Wait for service timed out')
request = SetParameters.Request()
request.parameters = parameters
future = client.call_async(request)
rclpy.spin_until_future_complete(self.node, future, timeout_sec=15.0)
response = future.result()
assert response.results[0].successful, 'Could not set parameter!'

View File

@@ -0,0 +1 @@
0

View File

@@ -0,0 +1 @@
# generated from colcon_core/shell/template/command_prefix.sh.em

View File

@@ -0,0 +1,20 @@
AMENT_PREFIX_PATH=/opt/ros/humble
COLCON=1
HOME=/root
HOSTNAME=a41c1d4d4b48
LANG=C.UTF-8
LC_ALL=C.UTF-8
LD_LIBRARY_PATH=/opt/ros/humble/opt/rviz_ogre_vendor/lib:/opt/ros/humble/lib/x86_64-linux-gnu:/opt/ros/humble/lib
LS_COLORS=rs=0:di=01;34:ln=01;36:mh=00:pi=40;33:so=01;35:do=01;35:bd=40;33;01:cd=40;33;01:or=40;31;01:mi=00:su=37;41:sg=30;43:ca=30;41:tw=30;42:ow=34;42:st=37;44:ex=01;32:*.tar=01;31:*.tgz=01;31:*.arc=01;31:*.arj=01;31:*.taz=01;31:*.lha=01;31:*.lz4=01;31:*.lzh=01;31:*.lzma=01;31:*.tlz=01;31:*.txz=01;31:*.tzo=01;31:*.t7z=01;31:*.zip=01;31:*.z=01;31:*.dz=01;31:*.gz=01;31:*.lrz=01;31:*.lz=01;31:*.lzo=01;31:*.xz=01;31:*.zst=01;31:*.tzst=01;31:*.bz2=01;31:*.bz=01;31:*.tbz=01;31:*.tbz2=01;31:*.tz=01;31:*.deb=01;31:*.rpm=01;31:*.jar=01;31:*.war=01;31:*.ear=01;31:*.sar=01;31:*.rar=01;31:*.alz=01;31:*.ace=01;31:*.zoo=01;31:*.cpio=01;31:*.7z=01;31:*.rz=01;31:*.cab=01;31:*.wim=01;31:*.swm=01;31:*.dwm=01;31:*.esd=01;31:*.jpg=01;35:*.jpeg=01;35:*.mjpg=01;35:*.mjpeg=01;35:*.gif=01;35:*.bmp=01;35:*.pbm=01;35:*.pgm=01;35:*.ppm=01;35:*.tga=01;35:*.xbm=01;35:*.xpm=01;35:*.tif=01;35:*.tiff=01;35:*.png=01;35:*.svg=01;35:*.svgz=01;35:*.mng=01;35:*.pcx=01;35:*.mov=01;35:*.mpg=01;35:*.mpeg=01;35:*.m2v=01;35:*.mkv=01;35:*.webm=01;35:*.webp=01;35:*.ogm=01;35:*.mp4=01;35:*.m4v=01;35:*.mp4v=01;35:*.vob=01;35:*.qt=01;35:*.nuv=01;35:*.wmv=01;35:*.asf=01;35:*.rm=01;35:*.rmvb=01;35:*.flc=01;35:*.avi=01;35:*.fli=01;35:*.flv=01;35:*.gl=01;35:*.dl=01;35:*.xcf=01;35:*.xwd=01;35:*.yuv=01;35:*.cgm=01;35:*.emf=01;35:*.ogv=01;35:*.ogx=01;35:*.aac=00;36:*.au=00;36:*.flac=00;36:*.m4a=00;36:*.mid=00;36:*.midi=00;36:*.mka=00;36:*.mp3=00;36:*.mpc=00;36:*.ogg=00;36:*.ra=00;36:*.wav=00;36:*.oga=00;36:*.opus=00;36:*.spx=00;36:*.xspf=00;36:
OLDPWD=/root/ros2_ws/src
PATH=/opt/ros/humble/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
PWD=/root/ros2_ws/build/launch_testing_examples
PYTHONPATH=/opt/ros/humble/lib/python3.10/site-packages:/opt/ros/humble/local/lib/python3.10/dist-packages
ROS_DISTRO=humble
ROS_LOCALHOST_ONLY=0
ROS_PYTHON_VERSION=3
ROS_VERSION=2
SHLVL=1
TERM=xterm
_=/usr/bin/colcon
_colcon_cd_root=/opt/ros/humble/

View File

@@ -0,0 +1,23 @@
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/record_rosbag_launch_test.py
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/__init__.py
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/set_param_launch_test.py
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/hello_world_launch_test.py
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/check_msgs_launch_test.py
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/check_multiple_nodes_launch_test.py
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/check_node_launch_test.py
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/__pycache__/record_rosbag_launch_test.cpython-310.pyc
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/__pycache__/__init__.cpython-310.pyc
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/__pycache__/set_param_launch_test.cpython-310.pyc
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/__pycache__/hello_world_launch_test.cpython-310.pyc
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/__pycache__/check_msgs_launch_test.cpython-310.pyc
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/__pycache__/check_multiple_nodes_launch_test.cpython-310.pyc
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples/__pycache__/check_node_launch_test.cpython-310.pyc
/root/ros2_ws/install/launch_testing_examples/share/ament_index/resource_index/packages/launch_testing_examples
/root/ros2_ws/install/launch_testing_examples/share/launch_testing_examples/package.xml
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples-0.15.3-py3.10.egg-info/zip-safe
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples-0.15.3-py3.10.egg-info/dependency_links.txt
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples-0.15.3-py3.10.egg-info/top_level.txt
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples-0.15.3-py3.10.egg-info/requires.txt
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples-0.15.3-py3.10.egg-info/entry_points.txt
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples-0.15.3-py3.10.egg-info/PKG-INFO
/root/ros2_ws/install/launch_testing_examples/lib/python3.10/site-packages/launch_testing_examples-0.15.3-py3.10.egg-info/SOURCES.txt

View File

@@ -0,0 +1 @@
/root/ros2_ws/src/examples/launch_testing/launch_testing_examples/launch_testing_examples

View File

@@ -0,0 +1,12 @@
Metadata-Version: 2.1
Name: launch-testing-examples
Version: 0.15.3
Summary: Examples of simple launch tests
Home-page: UNKNOWN
Maintainer: Aditya Pande, Shane Loretz
Maintainer-email: aditya.pande@openrobotics.org, shane@openrobotics.org
License: Apache License 2.0
Platform: UNKNOWN
UNKNOWN

View File

@@ -0,0 +1,22 @@
README.md
package.xml
setup.cfg
setup.py
../../../../build/launch_testing_examples/launch_testing_examples.egg-info/PKG-INFO
../../../../build/launch_testing_examples/launch_testing_examples.egg-info/SOURCES.txt
../../../../build/launch_testing_examples/launch_testing_examples.egg-info/dependency_links.txt
../../../../build/launch_testing_examples/launch_testing_examples.egg-info/entry_points.txt
../../../../build/launch_testing_examples/launch_testing_examples.egg-info/requires.txt
../../../../build/launch_testing_examples/launch_testing_examples.egg-info/top_level.txt
../../../../build/launch_testing_examples/launch_testing_examples.egg-info/zip-safe
launch_testing_examples/__init__.py
launch_testing_examples/check_msgs_launch_test.py
launch_testing_examples/check_multiple_nodes_launch_test.py
launch_testing_examples/check_node_launch_test.py
launch_testing_examples/hello_world_launch_test.py
launch_testing_examples/record_rosbag_launch_test.py
launch_testing_examples/set_param_launch_test.py
resource/launch_testing_examples
test/test_copyright.py
test/test_flake8.py
test/test_pep257.py

View File

@@ -0,0 +1,3 @@
[console_scripts]

View File

@@ -0,0 +1 @@
launch_testing_examples

View File

@@ -0,0 +1 @@
/root/ros2_ws/src/examples/launch_testing/launch_testing_examples/package.xml

View File

@@ -0,0 +1,4 @@
import sys
if sys.prefix == '/usr':
sys.real_prefix = sys.prefix
sys.prefix = sys.exec_prefix = '/root/ros2_ws/install/launch_testing_examples'

View File

@@ -0,0 +1 @@
/root/ros2_ws/src/examples/launch_testing/launch_testing_examples/resource/launch_testing_examples

View File

@@ -0,0 +1 @@
/root/ros2_ws/src/examples/launch_testing/launch_testing_examples/setup.cfg

View File

@@ -0,0 +1 @@
prepend-non-duplicate;PYTHONPATH;/root/ros2_ws/build/launch_testing_examples

View File

@@ -0,0 +1,3 @@
# generated from colcon_powershell/shell/template/hook_prepend_value.ps1.em
colcon_prepend_unique_value PYTHONPATH "$env:COLCON_CURRENT_PREFIX\/root/ros2_ws/build/launch_testing_examples"

View File

@@ -0,0 +1,3 @@
# generated from colcon_core/shell/template/hook_prepend_value.sh.em
_colcon_prepend_unique_value PYTHONPATH "/root/ros2_ws/build/launch_testing_examples"