Skip to content

Commit

Permalink
Fix test_ros_loader.py
Browse files Browse the repository at this point in the history 
Signed-off-by: Kenji Miyake <[email protected]>
  • Loading branch information
Kenji Miyake committed Dec 10, 2021
1 parent 021ae06 commit 28e5a1f
Show file tree
Hide file tree
Showing 3 changed files with 38 additions and 112 deletions.
9 changes: 0 additions & 9 deletions rosbridge_library/src/rosbridge_library/internal/ros_loader.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,15 +53,6 @@ def __init__(self, typestring):
Exception.__init__(self, "%s is not a valid type string" % typestring)


class InvalidPackageException(Exception):
def __init__(self, package, original_exception):
Exception.__init__(
self,
"Unable to load the manifest for package %s. Caused by: %s"
% (package, str(original_exception)),
)


class InvalidModuleException(Exception):
def __init__(self, modname, subname, original_exception):
Exception.__init__(
Expand Down
1 change: 0 additions & 1 deletion rosbridge_library/test/capabilities/test_advertise.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,6 @@ def test_invalid_msg_package(self):
nonexistent = [
"wangle_msgs/Jam",
"whistleblower_msgs/Document",
"sexual_harrassment_msgs/UnwantedAdvance",
"coercion_msgs/Bribe",
"airconditioning_msgs/Cold",
"pr2thoughts_msgs/Escape",
Expand Down
140 changes: 38 additions & 102 deletions rosbridge_library/test/internal/test_ros_loader.py
View file
Original file line number Diff line number Diff line change
@@ -1,15 +1,11 @@
#!/usr/bin/env python
import unittest

import rospy
import rostest
from rosbridge_library.internal import ros_loader
from rosidl_runtime_py.utilities import get_message


class TestROSLoader(unittest.TestCase):
def setUp(self):
rospy.init_node("test_ros_loader")

def test_bad_msgnames(self):
bad = [
"",
Expand All @@ -20,9 +16,6 @@ def test_bad_msgnames(self):
"/////",
"bad",
"stillbad",
"not/better/still",
"not//better//still",
"not///better///still",
"better/",
"better//",
"better///",
Expand All @@ -37,9 +30,7 @@ def test_bad_msgnames(self):
ros_loader.InvalidTypeStringException, ros_loader.get_message_class, x
)
self.assertRaises(
ros_loader.InvalidTypeStringException,
ros_loader.get_message_instance,
x,
ros_loader.InvalidTypeStringException, ros_loader.get_message_instance, x
)

def test_irregular_msgnames(self):
Expand All @@ -64,7 +55,6 @@ def test_std_msgnames(self):
"std_msgs/Byte",
"std_msgs/ByteMultiArray",
"std_msgs/ColorRGBA",
"std_msgs/Duration",
"std_msgs/Empty",
"std_msgs/Float32",
"std_msgs/Float32MultiArray",
Expand All @@ -81,7 +71,6 @@ def test_std_msgnames(self):
"std_msgs/MultiArrayDimension",
"std_msgs/MultiArrayLayout",
"std_msgs/String",
"std_msgs/Time",
"std_msgs/UInt16",
"std_msgs/UInt16MultiArray",
"std_msgs/UInt32MultiArray",
Expand All @@ -95,15 +84,14 @@ def test_std_msgnames(self):
self.assertNotEqual(ros_loader.get_message_class(x), None)
inst = ros_loader.get_message_instance(x)
self.assertNotEqual(inst, None)
self.assertEqual(x, inst._type)
self.assertEqual(get_message(x), type(inst))

def test_msg_cache(self):
stdmsgs = [
"std_msgs/Bool",
"std_msgs/Byte",
"std_msgs/ByteMultiArray",
"std_msgs/ColorRGBA",
"std_msgs/Duration",
"std_msgs/Empty",
"std_msgs/Float32",
"std_msgs/Float32MultiArray",
Expand All @@ -120,7 +108,6 @@ def test_msg_cache(self):
"std_msgs/MultiArrayDimension",
"std_msgs/MultiArrayLayout",
"std_msgs/String",
"std_msgs/Time",
"std_msgs/UInt16",
"std_msgs/UInt16MultiArray",
"std_msgs/UInt32MultiArray",
Expand All @@ -134,7 +121,7 @@ def test_msg_cache(self):
self.assertNotEqual(ros_loader.get_message_class(x), None)
inst = ros_loader.get_message_instance(x)
self.assertNotEqual(inst, None)
self.assertEqual(x, inst._type)
self.assertEqual(get_message(x), type(inst))
self.assertTrue(x in ros_loader._loaded_msgs)

def test_assorted_msgnames(self):
Expand All @@ -156,7 +143,7 @@ def test_assorted_msgnames(self):
self.assertNotEqual(ros_loader.get_message_class(x), None)
inst = ros_loader.get_message_instance(x)
self.assertNotEqual(inst, None)
self.assertEqual(x, inst._type)
self.assertEqual(get_message(x), type(inst))

def test_invalid_msgnames_primitives(self):
invalid = [
Expand Down Expand Up @@ -189,37 +176,30 @@ def test_nonexistent_packagenames(self):
nonexistent = [
"wangle_msgs/Jam",
"whistleblower_msgs/Document",
"sexual_harrassment_msgs/UnwantedAdvance",
"coercion_msgs/Bribe",
"airconditioning_msgs/Cold",
"pr2thoughts_msgs/Escape",
]
for x in nonexistent:
self.assertRaises(ros_loader.InvalidPackageException, ros_loader.get_message_class, x)
self.assertRaises(
ros_loader.InvalidPackageException, ros_loader.get_message_instance, x
)
self.assertRaises(ros_loader.InvalidModuleException, ros_loader.get_message_class, x)
self.assertRaises(ros_loader.InvalidModuleException, ros_loader.get_message_instance, x)

def test_packages_without_msgs(self):
no_msgs = [
"roslib/Time",
"roslib/Duration",
"roslib/Header",
"std_srvs/ConflictedMsg",
"topic_tools/MessageMessage",
]
for x in no_msgs:
self.assertRaises(ros_loader.InvalidModuleException, ros_loader.get_message_class, x)
self.assertRaises(ros_loader.InvalidModuleException, ros_loader.get_message_instance, x)

def test_nonexistent_msg_classnames(self):
nonexistent = [
"roscpp/Time",
"roscpp/Duration",
"roscpp/Header",
"rospy/Time",
"rospy/Duration",
"rospy/Header",
"rcl_interfaces/Time",
"rcl_interfaces/Duration",
"rcl_interfaces/Header",
"std_msgs/Spool",
"geometry_msgs/Tetrahedron",
"sensor_msgs/TelepathyUnit",
Expand All @@ -238,9 +218,9 @@ def test_bad_servicenames(self):
"/////",
"bad",
"stillbad",
"not/better/still",
"not//better//still",
"not///better///still",
"not/better/even/still",
"not//better//even//still",
"not///better///even///still",
"better/",
"better//",
"better///",
Expand All @@ -255,77 +235,58 @@ def test_bad_servicenames(self):
ros_loader.InvalidTypeStringException, ros_loader.get_service_class, x
)
self.assertRaises(
ros_loader.InvalidTypeStringException,
ros_loader.get_service_instance,
x,
ros_loader.InvalidTypeStringException, ros_loader.get_service_request_instance, x
)
self.assertRaises(
ros_loader.InvalidTypeStringException,
ros_loader.get_service_request_instance,
x,
)
self.assertRaises(
ros_loader.InvalidTypeStringException,
ros_loader.get_service_response_instance,
x,
ros_loader.InvalidTypeStringException, ros_loader.get_service_response_instance, x
)

def test_irregular_servicenames(self):
irregular = [
"roscpp//GetLoggers",
"/roscpp/GetLoggers/",
"/roscpp/GetLoggers",
"//roscpp/GetLoggers",
"/roscpp//GetLoggers",
"roscpp/GetLoggers//",
"/roscpp/GetLoggers//",
"roscpp/GetLoggers/",
"roscpp//GetLoggers//",
"rcl_interfaces//GetParameters",
"/rcl_interfaces/GetParameters/",
"/rcl_interfaces/GetParameters",
"//rcl_interfaces/GetParameters",
"/rcl_interfaces//GetParameters",
"rcl_interfaces/GetParameters//",
"/rcl_interfaces/GetParameters//",
"rcl_interfaces/GetParameters/",
"rcl_interfaces//GetParameters//",
]
for x in irregular:
self.assertNotEqual(ros_loader.get_service_class(x), None)
self.assertNotEqual(ros_loader.get_service_instance(x), None)
self.assertNotEqual(ros_loader.get_service_request_instance(x), None)
self.assertNotEqual(ros_loader.get_service_response_instance(x), None)

def test_common_servicenames(self):
common = [
"roscpp/GetLoggers",
"roscpp/SetLoggerLevel",
"rcl_interfaces/GetParameters",
"rcl_interfaces/SetParameters",
"std_srvs/Empty",
"nav_msgs/GetMap",
"nav_msgs/GetPlan",
"sensor_msgs/SetCameraInfo",
"topic_tools/MuxAdd",
"topic_tools/MuxSelect",
"tf2_msgs/FrameGraph",
"rospy_tutorials/BadTwoInts",
"rospy_tutorials/AddTwoInts",
"example_interfaces/AddTwoInts",
]
for x in common:
self.assertNotEqual(ros_loader.get_service_class(x), None)
self.assertNotEqual(ros_loader.get_service_instance(x), None)
self.assertNotEqual(ros_loader.get_service_request_instance(x), None)
self.assertNotEqual(ros_loader.get_service_response_instance(x), None)
self.assertEqual(x, ros_loader.get_service_instance(x)._type)

def test_srv_cache(self):
common = [
"roscpp/GetLoggers",
"roscpp/SetLoggerLevel",
"rcl_interfaces/GetParameters",
"rcl_interfaces/SetParameters",
"std_srvs/Empty",
"nav_msgs/GetMap",
"nav_msgs/GetPlan",
"sensor_msgs/SetCameraInfo",
"topic_tools/MuxAdd",
"topic_tools/MuxSelect",
"tf2_msgs/FrameGraph",
"rospy_tutorials/BadTwoInts",
"rospy_tutorials/AddTwoInts",
"example_interfaces/AddTwoInts",
]
for x in common:
self.assertNotEqual(ros_loader.get_service_class(x), None)
self.assertNotEqual(ros_loader.get_service_instance(x), None)
self.assertNotEqual(ros_loader.get_service_request_instance(x), None)
self.assertNotEqual(ros_loader.get_service_response_instance(x), None)
self.assertTrue(x in ros_loader._loaded_srvs)
Expand All @@ -334,16 +295,11 @@ def test_packages_without_srvs(self):
no_msgs = ["roslib/A", "roslib/B", "roslib/C", "std_msgs/CuriousSrv"]
for x in no_msgs:
self.assertRaises(ros_loader.InvalidModuleException, ros_loader.get_service_class, x)
self.assertRaises(ros_loader.InvalidModuleException, ros_loader.get_service_instance, x)
self.assertRaises(
ros_loader.InvalidModuleException,
ros_loader.get_service_request_instance,
x,
ros_loader.InvalidModuleException, ros_loader.get_service_request_instance, x
)
self.assertRaises(
ros_loader.InvalidModuleException,
ros_loader.get_service_response_instance,
x,
ros_loader.InvalidModuleException, ros_loader.get_service_response_instance, x
)

def test_nonexistent_service_packagenames(self):
Expand All @@ -354,45 +310,25 @@ def test_nonexistent_service_packagenames(self):
"revenge_srvs/BackStab",
]
for x in nonexistent:
self.assertRaises(ros_loader.InvalidPackageException, ros_loader.get_service_class, x)
self.assertRaises(ros_loader.InvalidModuleException, ros_loader.get_service_class, x)
self.assertRaises(
ros_loader.InvalidPackageException, ros_loader.get_service_instance, x
ros_loader.InvalidModuleException, ros_loader.get_service_request_instance, x
)
self.assertRaises(
ros_loader.InvalidPackageException,
ros_loader.get_service_request_instance,
x,
)
self.assertRaises(
ros_loader.InvalidPackageException,
ros_loader.get_service_response_instance,
x,
ros_loader.InvalidModuleException, ros_loader.get_service_response_instance, x
)

def test_nonexistent_service_classnames(self):
nonexistent = [
"std_srvs/KillAllHumans",
"std_srvs/Reboot",
"std_srvs/Full",
"rospy_tutorials/SubtractTwoInts",
"nav_msgs/LoseMap",
"topic_tools/TellMeWhatThisTopicIsActuallyAbout",
]
for x in nonexistent:
self.assertRaises(ros_loader.InvalidClassException, ros_loader.get_service_class, x)
self.assertRaises(ros_loader.InvalidClassException, ros_loader.get_service_instance, x)
self.assertRaises(
ros_loader.InvalidClassException,
ros_loader.get_service_request_instance,
x,
ros_loader.InvalidClassException, ros_loader.get_service_request_instance, x
)
self.assertRaises(
ros_loader.InvalidClassException,
ros_loader.get_service_response_instance,
x,
ros_loader.InvalidClassException, ros_loader.get_service_response_instance, x
)


PKG = "rosbridge_library"
NAME = "test_ros_loader"
if __name__ == "__main__":
rostest.unitrun(PKG, NAME, TestROSLoader)

0 comments on commit 28e5a1f

Please sign in to comment.