basic.py

Go to the documentation of this file.

#!/usr/bin/env python
 
# Test driver for basic test
 
from __future__ import print_function
 
import sys
import unittest
import time
import os
import math
 
import rospy
import rospy.client
 
import rospkg
 
from std_msgs.msg import String
 
from rosmon_msgs.msg import State, NodeState
from rosmon_msgs.srv import StartStop, StartStopRequest
 
rospack = rospkg.RosPack()
 
class _WFM(object):
        def __init__(self):
                self.msg = None
        def cb(self, msg):
                if self.msg is None:
                        self.msg = msg
 
 
class BasicTest(unittest.TestCase):
 
        def get_param(self, name):
                try:
                        return rospy.get_param(name)
                except KeyError as e:
                        params = ', '.join([ '"' + name + '"' for name in sorted(rospy.get_param_names()) ])
                        raise KeyError(
                                'Caught KeyError("%s") while getting parameter. Known parameters: %s' % (e, params)
                        )
 
        def test_rosmon_running(self):
                try:
                        state = rospy.client.wait_for_message('/rosmon_uut/state', State, timeout=5.0)
                except rospy.ROSException:
                        self.fail('Did not get state msg on /rosmon_uut/state' + repr(rospy.client.get_published_topics()))
 
                self.assertEqual(len(state.nodes), 4)
 
                test1 = [ n for n in state.nodes if n.name == 'test1' ]
                self.assertEqual(len(test1), 1)
                self.assertEqual(test1[0].state, NodeState.RUNNING)
 
        def test_remapping(self):
                pub = rospy.Publisher('/test_input', String, queue_size=5, latch=True)
 
                wfm = _WFM()
                sub = rospy.Subscriber('/test_output', String, wfm.cb)
 
                time.sleep(1);
 
                self.assertGreater(pub.get_num_connections(), 0)
                self.assertGreater(sub.get_num_connections(), 0)
                pub.publish('Hello world!')
 
                timeout_t = time.time() + 5
                while wfm.msg is None:
                        rospy.rostime.wallsleep(0.01)
                        if time.time() >= timeout_t:
                                self.fail('No reply to test message')
 
                self.assertEqual(wfm.msg.data, 'Hello world!')
 
                sub.unregister()
                pub.unregister()
 
        def test_params(self):
                self.assertEqual(self.get_param("path_to_rosmon"), rospack.get_path('rosmon_core'))
                self.assertEqual(
                        self.get_param("path_to_launch_file"),
                        os.path.join(rospack.get_path('rosmon_core'), 'test/basic.launch')
                )
                executable = self.get_param("path_to_rosmon_executable")
                self.assert_(os.access(executable, os.X_OK), 'Invalid rosmon path: ' + executable)
 
                self.assertEqual(
                        self.get_param("dirname"),
                        os.path.join(rospack.get_path('rosmon_core'), 'test')
                )
 
                self.assertEqual(self.get_param("eval_simple"), True)
                self.assertEqual(self.get_param("eval_argexpr"), True)
                self.assertAlmostEqual(self.get_param("eval_radius_pi"), 0.5 * math.pi)
 
                self.assertEqual(self.get_param("/test1/private_param1"), "hello_world")
                self.assertEqual(self.get_param("/test1/private_param2"), "hello_world")
 
        def test_multiLine(self):
                self.assertEqual(self.get_param("multiple_lines1"), "first_line  second_line")
                self.assertEqual(self.get_param("multiple_lines2"), "first_line second_line")
 
        def test_yaml(self):
                self.assertAlmostEqual(self.get_param("yaml/radius"), 0.5)
 
        def test_arg_passing(self):
                self.assertEqual(self.get_param("test_argument"), 123)
 
        def test_global_nested_ns(self):
                self.assertEqual(self.get_param("/foo/bar"), True)
 
        def test_global_remapping(self):
                pub = rospy.Publisher('/remapped_test_input', String, queue_size=5, latch=True)
 
                wfm = _WFM()
                sub = rospy.Subscriber('/remapped_test_output', String, wfm.cb)
 
                time.sleep(1);
 
                self.assertGreater(pub.get_num_connections(), 0)
                self.assertGreater(sub.get_num_connections(), 0)
                pub.publish('Hello world!')
 
                timeout_t = time.time() + 5
                while wfm.msg is None:
                        rospy.rostime.wallsleep(0.01)
                        if time.time() >= timeout_t:
                                self.fail('No reply to test message')
 
                self.assertEqual(wfm.msg.data, 'Hello world!')
 
                sub.unregister()
                pub.unregister()
 
        def test_nested(self):
                self.assertEqual(rospy.get_param("/nested/nested_param"), "hello")
 
        def getNode(self, state, name):
                lst = list(filter(lambda n: n.name == name, state.nodes))
                self.assertEqual(len(lst), 1)
                return lst[0]
 
        def test_startstop_one(self):
                srv = rospy.ServiceProxy('/rosmon_uut/start_stop', StartStop)
 
                srv(node=r'test1', action=StartStopRequest.STOP)
 
                time.sleep(2)
 
                try:
                        state = rospy.client.wait_for_message('/rosmon_uut/state', State, timeout=5.0)
                except rospy.ROSException:
                        self.fail('Did not get state msg on /rosmon_uut/state' + repr(rospy.client.get_published_topics()))
 
                self.assertEqual(self.getNode(state, 'test1').state, NodeState.IDLE)
                self.assertEqual(self.getNode(state, 'test2').state, NodeState.RUNNING)
 
                srv(node=r'test1', action=StartStopRequest.START)
 
                time.sleep(2)
 
                try:
                        state = rospy.client.wait_for_message('/rosmon_uut/state', State, timeout=5.0)
                except rospy.ROSException:
                        self.fail('Did not get state msg on /rosmon_uut/state' + repr(rospy.client.get_published_topics()))
 
                self.assertEqual(self.getNode(state, 'test1').state, NodeState.RUNNING)
                self.assertEqual(self.getNode(state, 'test2').state, NodeState.RUNNING)
 
        def test_startstop_regex(self):
                srv = rospy.ServiceProxy('/rosmon_uut/start_stop', StartStop)
 
                srv(node=r'test\d+', action=StartStopRequest.STOP)
 
                time.sleep(2)
 
                try:
                        state = rospy.client.wait_for_message('/rosmon_uut/state', State, timeout=5.0)
                except rospy.ROSException:
                        self.fail('Did not get state msg on /rosmon_uut/state' + repr(rospy.client.get_published_topics()))
 
                self.assertEqual(self.getNode(state, 'test1').state, NodeState.IDLE)
                self.assertEqual(self.getNode(state, 'test2').state, NodeState.IDLE)
 
                srv(node=r'test\d+', action=StartStopRequest.START)
 
                time.sleep(2)
 
                try:
                        state = rospy.client.wait_for_message('/rosmon_uut/state', State, timeout=5.0)
                except rospy.ROSException:
                        self.fail('Did not get state msg on /rosmon_uut/state' + repr(rospy.client.get_published_topics()))
 
                self.assertEqual(self.getNode(state, 'test1').state, NodeState.RUNNING)
                self.assertEqual(self.getNode(state, 'test2').state, NodeState.RUNNING)
 
if __name__ == '__main__':
        rospy.init_node('basic_test')
 
        # Wait for rosmon to start up
        state = rospy.client.wait_for_message('/rosmon_uut/state', State, timeout=5.0)
 
        import rostest
        rostest.rosrun('rosmon_core', 'basic', BasicTest)
 
        time.sleep(1)