test_multi_unregistering.py

Go to the documentation of this file.

#!/usr/bin/env python
 
import os
import rospy
import rosunit
import unittest
 
from time import sleep
 
from rosbridge_library.internal.publishers import MultiPublisher
from rosbridge_library.internal import ros_loader
 
 
PKG = 'rosbridge_library'
NAME = 'test_multi_unregistering'
 
 
class TestMultiUnregistering(unittest.TestCase):
 
    def setUp(self):
        rospy.init_node(NAME, log_level=rospy.DEBUG)
 
    def test_publish_once(self):
        """ Make sure that publishing works """
        topic = "/test_publish_once"
        msg_type = "std_msgs/String"
        msg = {"data": "why halo thar"}
 
        received = {"msg": None}
 
        def cb(msg):
            received["msg"] = msg
 
        rospy.Subscriber(topic, ros_loader.get_message_class(msg_type), cb)
        p = MultiPublisher(topic, msg_type)
        p.publish(msg)
 
        sleep(1)  # Time to publish and receive
 
        self.assertEqual(received["msg"].data, msg["data"])
 
    def test_publish_twice(self):
        """ Make sure that publishing works """
        topic = "/test_publish_twice"
        msg_type = "std_msgs/String"
        msg = {"data": "why halo thar"}
 
        received = {"msg": None}
 
        def cb(msg):
            received["msg"] = msg
 
        rospy.Subscriber(topic, ros_loader.get_message_class(msg_type), cb)
        p = MultiPublisher(topic, msg_type)
        p.publish(msg)
 
        sleep(1)  # Time to publish and receive
 
        self.assertEqual(received["msg"].data, msg["data"])
 
        p.unregister()
        sleep(5)  # Time to unregister
 
        received["msg"] = None  # Reset received
        self.assertIsNone(received["msg"])
        p = MultiPublisher(topic, msg_type)
        p.publish(msg)
 
        sleep(1)   # Time to publish and receive
 
        self.assertEqual(received["msg"].data, msg["data"])
 
 
if __name__ == '__main__':
    rosunit.unitrun(PKG, NAME, TestMultiUnregistering)