<s> import sys <EOL> import rospy <EOL> import rostest <EOL> import unittest <EOL> from time import sleep <EOL> from rosbridge_library . internal . publishers import * <EOL> from rosbridge_library . internal . topics import * <EOL> from rosbridge_library . internal . message_conversion import FieldTypeMismatchException <EOL> from std_msgs . msg import String , Int32 <EOL> class TestPublisherManager ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> rospy . init_node ( "<STR_LIT>" ) <EOL> def is_topic_published ( self , topicname ) : <EOL> return topicname in dict ( rospy . get_published_topics ( ) ) . keys ( ) <EOL> def test_register_publisher ( self ) : <EOL> """<STR_LIT>""" <EOL> topic = "<STR_LIT>" <EOL> msg_type = "<STR_LIT>" <EOL> client = "<STR_LIT>" <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> manager . register ( client , topic , msg_type ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> manager . unregister ( client , topic ) <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> def test_register_publisher_multiclient ( self ) : <EOL> topic = "<STR_LIT>" <EOL> msg_type = "<STR_LIT>" <EOL> client1 = "<STR_LIT>" <EOL> client2 = "<STR_LIT>" <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> manager . register ( client1 , topic , msg_type ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> manager . register ( client2 , topic , msg_type ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> manager . unregister ( client1 , topic ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> manager . unregister ( client2 , topic ) <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> def test_register_publisher_conflicting_types ( self ) : <EOL> topic = "<STR_LIT>" <EOL> msg_type = "<STR_LIT>" <EOL> msg_type_bad = "<STR_LIT>" <EOL> client = "<STR_LIT>" <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> manager . register ( client , topic , msg_type ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> self . assertRaises ( TypeConflictException , manager . register , "<STR_LIT>" , topic , msg_type_bad ) <EOL> def test_register_multiple_publishers ( self ) : <EOL> topic1 = "<STR_LIT>" <EOL> topic2 = "<STR_LIT>" <EOL> msg_type = "<STR_LIT>" <EOL> client = "<STR_LIT>" <EOL> self . assertFalse ( topic1 in manager . _publishers ) <EOL> self . assertFalse ( topic2 in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic1 ) ) <EOL> self . assertFalse ( self . is_topic_published ( topic2 ) ) <EOL> manager . register ( client , topic1 , msg_type ) <EOL> self . assertTrue ( topic1 in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic1 ) ) <EOL> self . assertFalse ( topic2 in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic2 ) ) <EOL> manager . register ( client , topic2 , msg_type ) <EOL> self . assertTrue ( topic1 in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic1 ) ) <EOL> self . assertTrue ( topic2 in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic2 ) ) <EOL> manager . unregister ( client , topic1 ) <EOL> self . assertFalse ( topic1 in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic1 ) ) <EOL> self . assertTrue ( topic2 in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic2 ) ) <EOL> manager . unregister ( client , topic2 ) <EOL> self . assertFalse ( topic1 in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic1 ) ) <EOL> self . assertFalse ( topic2 in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic2 ) ) <EOL> def test_register_no_msgtype ( self ) : <EOL> topic = "<STR_LIT>" <EOL> client = "<STR_LIT>" <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> self . assertRaises ( TopicNotEstablishedException , manager . register , client , topic ) <EOL> def test_register_infer_topictype ( self ) : <EOL> topic = "<STR_LIT>" <EOL> client = "<STR_LIT>" <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> rospy . Publisher ( topic , String ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> manager . register ( client , topic ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> manager . unregister ( client , topic ) <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> def test_register_multiple_notopictype ( self ) : <EOL> topic = "<STR_LIT>" <EOL> msg_type = "<STR_LIT>" <EOL> client1 = "<STR_LIT>" <EOL> client2 = "<STR_LIT>" <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> manager . register ( client1 , topic , msg_type ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> manager . register ( client2 , topic ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> manager . unregister ( client1 , topic ) <EOL> self . assertTrue ( topic in manager . _publishers ) <EOL> self . assertTrue ( self . is_topic_published ( topic ) ) <EOL> manager . unregister ( client2 , topic ) <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> def test_publish_not_registered ( self ) : <EOL> topic = "<STR_LIT>" <EOL> msg = { "<STR_LIT:data>" : "<STR_LIT>" } <EOL> client = "<STR_LIT>" <EOL> self . assertFalse ( topic in manager . _publishers ) <EOL> self . assertFalse ( self . is_topic_published ( topic ) ) <EOL> self . assertRaises ( TopicNotEstablishedException , manager . publish , client , topic , msg ) <EOL> def test_publisher_manager_publish ( self ) : <EOL> """<STR_LIT>""" <EOL> topic = "<STR_LIT>" <EOL> msg = { "<STR_LIT:data>" : "<STR_LIT>" } <EOL> client = "<STR_LIT>" <EOL> received = { "<STR_LIT>" : None } <EOL> def cb ( msg ) : <EOL> received [ "<STR_LIT>" ] = msg <EOL> rospy . Subscriber ( topic , String , cb ) <EOL> manager . publish ( client , topic , msg ) <EOL> sleep ( <NUM_LIT:0.5> ) <EOL> self . assertEqual ( received [ "<STR_LIT>" ] . data , msg [ "<STR_LIT:data>" ] ) <EOL> def test_publisher_manager_bad_publish ( self ) : <EOL> """<STR_LIT>""" <EOL> topic = "<STR_LIT>" <EOL> client = "<STR_LIT>" <EOL> msg_type = "<STR_LIT>" <EOL> msg = { "<STR_LIT:data>" : <NUM_LIT:3> } <EOL> manager . register ( client , topic , msg_type ) <EOL> self . assertRaises ( FieldTypeMismatchException , manager . publish , client , topic , msg ) <EOL> PKG = '<STR_LIT>' <EOL> NAME = '<STR_LIT>' <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> rostest . unitrun ( PKG , NAME , TestPublisherManager ) </s>
