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
00036 import roslib; roslib.load_manifest('test_rospy')
00037
00038 import os
00039 import sys
00040 import struct
00041 import unittest
00042 import time
00043
00044 class TestRospyTransport(unittest.TestCase):
00045
00046 def test_Transport(self):
00047 from rospy.impl.transport import Transport, INBOUND, OUTBOUND, BIDIRECTIONAL
00048 ids = []
00049 for d in [INBOUND, OUTBOUND, BIDIRECTIONAL]:
00050 t = Transport(d)
00051 self.assertEquals(d, t.direction)
00052 self.assertEquals("UNKNOWN", t.transport_type)
00053 self.failIf(t.done)
00054 self.assertEquals(None, t.cleanup_cb)
00055 self.assertEquals('', t.endpoint_id)
00056 self.assertEquals('unnamed', t.name)
00057 self.assertEquals(0, t.stat_bytes)
00058 self.assertEquals(0, t.stat_num_msg)
00059 self.failIf(t.id in ids)
00060 ids.append(t.id)
00061
00062 t = Transport(d, 'a name')
00063 self.assertEquals(d, t.direction)
00064 self.assertEquals('a name', t.name)
00065
00066
00067 t = Transport(INBOUND)
00068 t.close()
00069 self.assert_(t.done)
00070 t = Transport(INBOUND)
00071
00072 self.cleanup_obj = None
00073 def cleanup(obj):
00074 self.cleanup_obj = obj
00075
00076 t.set_cleanup_callback(cleanup)
00077 self.assertEquals(t.cleanup_cb, cleanup)
00078 t.close()
00079 self.assert_(t.done)
00080 self.assertEquals(self.cleanup_obj, t)
00081
00082 t = Transport(OUTBOUND)
00083 import traceback
00084 try:
00085 t.send_message('msg', 1)
00086 self.fail("send_message() should be abstract")
00087 except:
00088 traceback.print_exc()
00089 try:
00090 t.write_data('data')
00091 self.fail("write_data() should be abstract")
00092 except:
00093 traceback.print_exc()
00094 t = Transport(INBOUND)
00095 try:
00096 t.receive_once()
00097 self.fail("receive_once() should be abstract")
00098 except: pass
00099 t = Transport(INBOUND)
00100 try:
00101 t.receive_loop()
00102 self.fail("receive_loop() should be abstract")
00103 except: pass
00104
00105 def test_DeadTransport(self):
00106 from rospy.impl.transport import Transport, DeadTransport, INBOUND, OUTBOUND, BIDIRECTIONAL
00107 t = Transport(INBOUND, 'foo')
00108 t.stat_bytes = 1234
00109 t.stat_num_msg = 5678
00110 dead = DeadTransport(t)
00111 self.assertEquals(INBOUND, dead.direction)
00112 self.assertEquals('foo', dead.name)
00113 self.assertEquals(1234, dead.stat_bytes)
00114 self.assertEquals(5678, dead.stat_num_msg)
00115 self.assertEquals(True, dead.done)
00116 self.assertEquals('', dead.endpoint_id)
00117
00118 t = Transport(OUTBOUND, 'bar')
00119 t.endpoint_id = 'blah blah'
00120 t.close()
00121 dead = DeadTransport(t)
00122 self.assertEquals(OUTBOUND, dead.direction)
00123 self.assertEquals('bar', dead.name)
00124 self.assertEquals(True, dead.done)
00125 self.assertEquals(t.endpoint_id, dead.endpoint_id)
00126
00127 def test_ProtocolHandler(self):
00128
00129 from rospy.impl.transport import ProtocolHandler
00130 h = ProtocolHandler()
00131 self.failIf(h.supports('TCPROS'))
00132 self.assertEquals([], h.get_supported())
00133 try:
00134 h.create_connection("/topic", 'http://localhost:1234', ['TCPROS'])
00135 self.fail("create_connection should raise an exception")
00136 except: pass
00137 try:
00138 h.init_publisher("/topic", 'TCPROS')
00139 self.fail("init_publisher raise an exception")
00140 except: pass
00141 h.shutdown()
00142
00143 if __name__ == '__main__':
00144 import rostest
00145 rostest.unitrun('test_rospy', sys.argv[0], TestRospyTransport, coverage_packages=['rospy.impl.transport'])