00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035 import roslib
00036 roslib.load_manifest('joy')
00037
00038 import sys
00039 import struct
00040
00041 import unittest
00042
00043 import rostest
00044 import rosbag
00045 import rosbagmigration
00046
00047 import re
00048 from cStringIO import StringIO
00049 import os
00050
00051 import rospy
00052
00053
00054
00055 migrator = rosbagmigration.MessageMigrator()
00056
00057
00058 def repack(x):
00059 return struct.unpack('<f',struct.pack('<f',x))[0]
00060
00061 class TestJoyMsgsMigration(unittest.TestCase):
00062
00063
00064
00065
00066
00067
00068 def get_old_joy(self):
00069 joy_classes = self.load_saved_classes('Joy.saved')
00070 joy = joy_classes['joy/Joy']
00071 return joy([0.1,0.2,0.3,0.4,0.5],[0,1,0,1,0])
00072
00073 def get_new_joy(self):
00074 from sensor_msgs.msg import Joy
00075 from roslib.msg import Header
00076 return Joy(Header(),[0.1,0.2,0.3,0.4,0.5],[0,1,0,1,0])
00077
00078
00079 def test_joy(self):
00080 self.do_test('joy', self.get_old_joy, self.get_new_joy)
00081
00082
00083
00084 def setUp(self):
00085 self.pkg_dir = roslib.packages.get_pkg_dir("joy")
00086
00087
00088 def load_saved_classes(self,saved_msg):
00089 f = open("%s/test/saved/%s"%(self.pkg_dir,saved_msg), 'r')
00090
00091 type_line = f.readline()
00092 pat = re.compile(r"\[(.*)]:")
00093 type_match = pat.match(type_line)
00094
00095 self.assertTrue(type_match is not None, "Full definition file malformed. First line should be: '[my_package/my_msg]:'")
00096
00097 saved_type = type_match.groups()[0]
00098 saved_full_text = f.read()
00099
00100 saved_classes = roslib.genpy.generate_dynamic(saved_type,saved_full_text)
00101
00102 self.assertTrue(saved_classes is not None, "Could not generate class from full definition file.")
00103 self.assertTrue(saved_classes.has_key(saved_type), "Could not generate class from full definition file.")
00104
00105 return saved_classes
00106
00107 def do_test(self, name, old_msg, new_msg):
00108
00109 oldbag = "%s/test/%s_old.bag"%(self.pkg_dir,name)
00110 newbag = "%s/test/%s_new.bag"%(self.pkg_dir,name)
00111
00112
00113 bag = rosbag.Bag(oldbag, 'w')
00114 bag.write("topic", old_msg(), roslib.rostime.Time())
00115 bag.close()
00116
00117
00118 res = rosbagmigration.checkbag(migrator, oldbag)
00119 self.assertTrue(not False in [m[1] == [] for m in res], 'Bag not ready to be migrated')
00120 res = rosbagmigration.fixbag(migrator, oldbag, newbag)
00121 self.assertTrue(res, 'Bag not converted successfully')
00122
00123
00124 topic, msg, t = rosbag.Bag(newbag).read_messages().next()
00125
00126
00127 m = new_msg()
00128 buff = StringIO()
00129 m.serialize(buff)
00130 m.deserialize(buff.getvalue())
00131
00132
00133 self.assertTrue(roslib.message.strify_message(msg) == roslib.message.strify_message(m))
00134
00135
00136
00137 os.remove(oldbag)
00138 os.remove(newbag)
00139
00140
00141 if __name__ == '__main__':
00142 rostest.unitrun('test_joy_msg', 'test_joy_msg_migration', TestJoyMsgsMigration, sys.argv)