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 from __future__ import with_statement
00037
00038 NAME = 'test_rostopic_command_line_offline'
00039 import roslib; roslib.load_manifest('test_rostopic')
00040
00041 import os
00042 import sys
00043 import unittest
00044 import cStringIO
00045 import time
00046
00047 import rostest
00048
00049 from subprocess import Popen, PIPE, check_call, call
00050
00051 class TestRostopicOffline(unittest.TestCase):
00052
00053 def setUp(self):
00054 pass
00055
00056
00057 def test_cmd_help(self):
00058 cmd = 'rostopic'
00059
00060 sub = ['bw', 'echo', 'hz', 'info', 'list', 'pub', 'type','find']
00061 output = Popen([cmd], stdout=PIPE).communicate()[0]
00062 self.assert_('Commands' in output)
00063 output = Popen([cmd, '-h'], stdout=PIPE).communicate()[0]
00064 self.assert_('Commands' in output)
00065
00066 for c in sub:
00067 self.assert_("%s %s"%(cmd, c) in output, output)
00068
00069 for c in sub:
00070 output = Popen([cmd, c, '-h'], stdout=PIPE, stderr=PIPE).communicate()
00071 self.assert_("Usage:" in output[0], output)
00072
00073 self.assert_("%s %s"%(cmd, c) in output[0], output)
00074
00075
00076 for c in ['bw', 'echo', 'hz', 'info', 'pub', 'type', 'find']:
00077 output = Popen([cmd, c], stdout=PIPE, stderr=PIPE).communicate()
00078 self.assert_("Usage:" in output[0] or "Usage:" in output[1], output)
00079
00080 self.assert_("%s %s"%(cmd, c) in output[1], output)
00081
00082 def test_offline(self):
00083 cmd = 'rostopic'
00084
00085
00086 env = os.environ.copy()
00087 env['ROS_MASTER_URI'] = 'http://localhost:11312'
00088 kwds = { 'env': env, 'stdout': PIPE, 'stderr': PIPE}
00089
00090 msg = "ERROR: Unable to communicate with master!\n"
00091
00092 output = Popen([cmd, 'bw', 'chatter'], **kwds).communicate()
00093 self.assertEquals(msg, output[1])
00094 output = Popen([cmd, 'echo', 'chatter'], **kwds).communicate()
00095 self.assertEquals(msg, output[1])
00096 output = Popen([cmd, 'hz', 'chatter'], **kwds).communicate()
00097 self.assertEquals(msg, output[1])
00098 output = Popen([cmd, 'list'], **kwds).communicate()
00099 self.assertEquals(msg, output[1])
00100 output = Popen([cmd, 'pub', 'chatter', 'std_msgs/String', 'hello'], **kwds).communicate()
00101 self.assertEquals(msg, output[1])
00102 output = Popen([cmd, 'type', 'chatter'], **kwds).communicate()
00103 self.assertEquals(msg, output[1])
00104 output = Popen([cmd, 'type', 'std_msgs/String'], **kwds).communicate()
00105 self.assertEquals(msg, output[1])
00106
00107 if __name__ == '__main__':
00108 rostest.unitrun('test_rostopic', NAME, TestRostopicOffline, sys.argv, coverage_packages=[])