<s> import sys <EOL> import rospy <EOL> import rostest <EOL> import unittest <EOL> from json import loads , dumps <EOL> from StringIO import StringIO <EOL> from rosbridge_library . internal import message_conversion as c <EOL> from rosbridge_library . internal import ros_loader <EOL> from base64 import standard_b64encode , standard_b64decode <EOL> class TestMessageConversion ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> rospy . init_node ( "<STR_LIT>" ) <EOL> def validate_instance ( self , inst1 ) : <EOL> """<STR_LIT>""" <EOL> inst1 . _check_types ( ) <EOL> buff = StringIO ( ) <EOL> inst1 . serialize ( buff ) <EOL> inst2 = type ( inst1 ) ( ) <EOL> inst2 . deserialize ( buff . getvalue ( ) ) <EOL> self . assertEqual ( inst1 , inst2 ) <EOL> inst2 . _check_types ( ) <EOL> def msgs_equal ( self , msg1 , msg2 ) : <EOL> if type ( msg1 ) in [ str , unicode ] and type ( msg2 ) in [ str , unicode ] : <EOL> pass <EOL> else : <EOL> self . assertEqual ( type ( msg1 ) , type ( msg2 ) ) <EOL> if type ( msg1 ) in c . list_types : <EOL> for x , y in zip ( msg1 , msg2 ) : <EOL> self . msgs_equal ( x , y ) <EOL> elif type ( msg1 ) in c . primitive_types or type ( msg1 ) in c . string_types : <EOL> self . assertEqual ( msg1 , msg2 ) <EOL> else : <EOL> for x in msg1 : <EOL> self . assertTrue ( x in msg2 ) <EOL> for x in msg2 : <EOL> self . assertTrue ( x in msg1 ) <EOL> for x in msg1 : <EOL> self . msgs_equal ( msg1 [ x ] , msg2 [ x ] ) <EOL> def do_primitive_test ( self , data_value , msgtype ) : <EOL> for msg in [ { "<STR_LIT:data>" : data_value } , loads ( dumps ( { "<STR_LIT:data>" : data_value } ) ) ] : <EOL> inst = ros_loader . get_message_instance ( msgtype ) <EOL> c . populate_instance ( msg , inst ) <EOL> self . assertEqual ( inst . data , data_value ) <EOL> self . validate_instance ( inst ) <EOL> extracted = c . extract_values ( inst ) <EOL> for msg2 in [ extracted , loads ( dumps ( extracted ) ) ] : <EOL> self . msgs_equal ( msg , msg2 ) <EOL> self . assertEqual ( msg [ "<STR_LIT:data>" ] , msg2 [ "<STR_LIT:data>" ] ) <EOL> self . assertEqual ( msg2 [ "<STR_LIT:data>" ] , inst . data ) <EOL> def do_test ( self , orig_msg , msgtype ) : <EOL> for msg in [ orig_msg , loads ( dumps ( orig_msg ) ) ] : <EOL> inst = ros_loader . get_message_instance ( msgtype ) <EOL> c . populate_instance ( msg , inst ) <EOL> self . validate_instance ( inst ) <EOL> extracted = c . extract_values ( inst ) <EOL> for msg2 in [ extracted , loads ( dumps ( extracted ) ) ] : <EOL> self . msgs_equal ( msg , msg2 ) <EOL> def test_int_primitives ( self ) : <EOL> for msg in range ( - <NUM_LIT:100> , <NUM_LIT:100> ) : <EOL> for rostype in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> self . assertEqual ( c . _to_primitive_inst ( msg , rostype , rostype , [ ] ) , msg ) <EOL> self . assertEqual ( c . _to_inst ( msg , rostype , rostype ) , msg ) <EOL> for msg in range ( <NUM_LIT:0> , <NUM_LIT:200> ) : <EOL> for rostype in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> self . assertEqual ( c . _to_primitive_inst ( msg , rostype , rostype , [ ] ) , msg ) <EOL> self . assertEqual ( c . _to_inst ( msg , rostype , rostype ) , msg ) <EOL> def test_bool_primitives ( self ) : <EOL> self . assertTrue ( c . _to_primitive_inst ( True , "<STR_LIT:bool>" , "<STR_LIT:bool>" , [ ] ) ) <EOL> self . assertTrue ( c . _to_inst ( True , "<STR_LIT:bool>" , "<STR_LIT:bool>" ) ) <EOL> self . assertFalse ( c . _to_primitive_inst ( False , "<STR_LIT:bool>" , "<STR_LIT:bool>" , [ ] ) ) <EOL> self . assertFalse ( c . _to_inst ( False , "<STR_LIT:bool>" , "<STR_LIT:bool>" ) ) <EOL> def test_float_primitives ( self ) : <EOL> for msg in [ <NUM_LIT> + i for i in range ( - <NUM_LIT:100> , <NUM_LIT:100> ) ] : <EOL> for rostype in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> self . assertEqual ( c . _to_primitive_inst ( msg , rostype , rostype , [ ] ) , msg ) <EOL> self . assertEqual ( c . _to_inst ( msg , rostype , rostype ) , msg ) <EOL> c . _to_inst ( msg , rostype , rostype ) <EOL> def test_float_special_cases ( self ) : <EOL> for msg in [ <NUM_LIT> , - <NUM_LIT> , float ( '<STR_LIT>' ) ] : <EOL> for rostype in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> self . assertEqual ( c . _from_inst ( msg , rostype ) , None ) <EOL> self . assertEqual ( dumps ( { "<STR_LIT:data>" : c . _from_inst ( msg , rostype ) } ) , "<STR_LIT>" ) <EOL> def test_signed_int_base_msgs ( self ) : <EOL> int8s = range ( - <NUM_LIT> , <NUM_LIT> ) <EOL> for int8 in int8s : <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> int16s = [ - <NUM_LIT> , <NUM_LIT> ] <EOL> for int16 in int16s : <EOL> self . do_primitive_test ( int16 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int16 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int16 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int16 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int16 , "<STR_LIT>" ) <EOL> int32s = [ - <NUM_LIT> , <NUM_LIT> ] <EOL> for int32 in int32s : <EOL> self . do_primitive_test ( int32 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int32 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int32 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int32 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int32 , "<STR_LIT>" ) <EOL> int64s = [ - <NUM_LIT> , <NUM_LIT> ] <EOL> for int64 in int64s : <EOL> self . do_primitive_test ( int64 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int64 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int64 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int64 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int64 , "<STR_LIT>" ) <EOL> def test_unsigned_int_base_msgs ( self ) : <EOL> int8s = range ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> for int8 in int8s : <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int8 , "<STR_LIT>" ) <EOL> int16s = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] <EOL> for int16 in int16s : <EOL> self . do_primitive_test ( int16 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int16 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int16 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int16 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int16 , "<STR_LIT>" ) <EOL> int32s = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] <EOL> for int32 in int32s : <EOL> self . do_primitive_test ( int32 , "<STR_LIT>" ) <EOL> self . do_primitive_test ( int32 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int32 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int32 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int32 , "<STR_LIT>" ) <EOL> int64s = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <EOL> <NUM_LIT> ] <EOL> for int64 in int64s : <EOL> self . do_primitive_test ( int64 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int64 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int64 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int64 , "<STR_LIT>" ) <EOL> self . assertRaises ( Exception , self . do_primitive_test , int64 , "<STR_LIT>" ) <EOL> def test_bool_base_msg ( self ) : <EOL> self . do_primitive_test ( True , "<STR_LIT>" ) <EOL> self . do_primitive_test ( False , "<STR_LIT>" ) <EOL> def test_string_base_msg ( self ) : <EOL> for x in c . ros_primitive_types : <EOL> self . do_primitive_test ( x , "<STR_LIT>" ) <EOL> def test_time_msg ( self ) : <EOL> msg = { "<STR_LIT:data>" : { "<STR_LIT>" : <NUM_LIT:3> , "<STR_LIT>" : <NUM_LIT:5> } } <EOL> self . do_test ( msg , "<STR_LIT>" ) <EOL> msg = { "<STR_LIT>" : [ { "<STR_LIT>" : <NUM_LIT:3> , "<STR_LIT>" : <NUM_LIT:5> } , { "<STR_LIT>" : <NUM_LIT:2> , "<STR_LIT>" : <NUM_LIT:7> } ] } <EOL> self . do_test ( msg , "<STR_LIT>" ) <EOL> def test_time_msg_now ( self ) : <EOL> msg = { "<STR_LIT:data>" : "<STR_LIT>" } <EOL> msgtype = "<STR_LIT>" <EOL> inst = ros_loader . get_message_instance ( msgtype ) <EOL> c . populate_instance ( msg , inst ) <EOL> currenttime = rospy . get_rostime ( ) <EOL> self . validate_instance ( inst ) <EOL> extracted = c . extract_values ( inst ) <EOL> print extracted <EOL> self . assertIn ( "<STR_LIT:data>" , extracted ) <EOL> self . assertIn ( "<STR_LIT>" , extracted [ "<STR_LIT:data>" ] ) <EOL> self . assertIn ( "<STR_LIT>" , extracted [ "<STR_LIT:data>" ] ) <EOL> self . assertNotEqual ( extracted [ "<STR_LIT:data>" ] [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> self . assertLessEqual ( extracted [ "<STR_LIT:data>" ] [ "<STR_LIT>" ] , currenttime . secs ) <EOL> self . assertGreaterEqual ( currenttime . secs , extracted [ "<STR_LIT:data>" ] [ "<STR_LIT>" ] ) <EOL> def test_duration_msg ( self ) : <EOL> msg = { "<STR_LIT:data>" : { "<STR_LIT>" : <NUM_LIT:3> , "<STR_LIT>" : <NUM_LIT:5> } } <EOL> self . do_test ( msg , "<STR_LIT>" ) <EOL> msg = { "<STR_LIT>" : [ { "<STR_LIT>" : <NUM_LIT:3> , "<STR_LIT>" : <NUM_LIT:5> } , { "<STR_LIT>" : <NUM_LIT:2> , "<STR_LIT>" : <NUM_LIT:7> } ] } <EOL> self . do_test ( msg , "<STR_LIT>" ) <EOL> def test_header_msg ( self ) : <EOL> msg = { "<STR_LIT>" : <NUM_LIT:5> , "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT> , "<STR_LIT>" : <NUM_LIT> } , "<STR_LIT>" : "<STR_LIT>" } <EOL> self . do_test ( msg , "<STR_LIT>" ) <EOL> msg = { "<STR_LIT>" : msg } <EOL> self . do_test ( msg , "<STR_LIT>" ) <EOL> self . do_test ( msg , "<STR_LIT>" ) <EOL> msg = { "<STR_LIT>" : [ msg [ "<STR_LIT>" ] , msg [ "<STR_LIT>" ] , msg [ "<STR_LIT>" ] ] } <EOL> msg [ "<STR_LIT>" ] [ <NUM_LIT:1> ] [ "<STR_LIT>" ] = <NUM_LIT:6> <EOL> msg [ "<STR_LIT>" ] [ <NUM_LIT:2> ] [ "<STR_LIT>" ] = <NUM_LIT:7> <EOL> self . do_test ( msg , "<STR_LIT>" ) <EOL> def test_assorted_msgs ( self ) : <EOL> assortedmsgs = [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> for rostype in assortedmsgs : <EOL> inst = ros_loader . get_message_instance ( rostype ) <EOL> msg = c . extract_values ( inst ) <EOL> self . do_test ( msg , rostype ) <EOL> l = loads ( dumps ( msg ) ) <EOL> inst2 = ros_loader . get_message_instance ( rostype ) <EOL> c . populate_instance ( msg , inst2 ) <EOL> self . assertEqual ( inst , inst2 ) <EOL> def test_int8array ( self ) : <EOL> def test_int8_msg ( rostype , data ) : <EOL> msg = { "<STR_LIT:data>" : data } <EOL> inst = ros_loader . get_message_instance ( rostype ) <EOL> c . populate_instance ( msg , inst ) <EOL> self . validate_instance ( inst ) <EOL> return inst . data <EOL> for msgtype in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> rostype = "<STR_LIT>" + msgtype <EOL> int8s = range ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> ret = test_int8_msg ( rostype , int8s ) <EOL> self . assertEqual ( ret , str ( bytearray ( int8s ) ) ) <EOL> str_int8s = str ( bytearray ( int8s ) ) <EOL> b64str_int8s = standard_b64encode ( str_int8s ) <EOL> ret = test_int8_msg ( rostype , b64str_int8s ) <EOL> self . assertEqual ( ret , str_int8s ) <EOL> for msgtype in [ "<STR_LIT>" ] : <EOL> rostype = "<STR_LIT>" + msgtype <EOL> int8s = range ( <NUM_LIT:0> , <NUM_LIT:16> ) <EOL> ret = test_int8_msg ( rostype , int8s ) <EOL> self . assertEqual ( ret , str ( bytearray ( int8s ) ) ) <EOL> str_int8s = str ( bytearray ( int8s ) ) <EOL> b64str_int8s = standard_b64encode ( str_int8s ) <EOL> ret = test_int8_msg ( rostype , b64str_int8s ) <EOL> self . assertEqual ( ret , str_int8s ) <EOL> PKG = '<STR_LIT>' <EOL> NAME = '<STR_LIT>' <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> rostest . unitrun ( PKG , NAME , TestMessageConversion ) </s>
