Go to the documentation of this file.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
00037
00038
00039 from __future__ import with_statement
00040
00041 DURATION = 10
00042 PKG = 'test_diagnostic_aggregator'
00043 import roslib; roslib.load_manifest(PKG)
00044 import rospy, rostest, unittest
00045 from diagnostic_msgs.msg import DiagnosticArray, DiagnosticStatus, KeyValue
00046 from time import sleep
00047 import sys
00048 import threading
00049
00050 from optparse import OptionParser
00051
00052 MATCH_NAME = 'Match Item'
00053
00054 def get_raw_name(agg_name):
00055 return agg_name.split('/')[-1]
00056
00057 def get_header_name(agg_name):
00058 return '/'.join(agg_name.split('/')[1:-1])
00059
00060 class TestMatchAnalyze(unittest.TestCase):
00061 def __init__(self, *args):
00062 super(TestMatchAnalyze, self).__init__(*args)
00063
00064 parser = OptionParser(usage="usage ./%prog [options]", prog="match_analyze_test.py")
00065 parser.add_option('--header', action="store", default=None,
00066 dest="header", metavar="HEADER",
00067 help="Expected header that \"Match Item\" will be under")
00068
00069 parser.add_option('--gtest_output', action="store",
00070 dest="gtest")
00071
00072 options, args = parser.parse_args(rospy.myargv())
00073
00074 if not options.header:
00075 parser.error("Option --header is mandatory. Unable to parse args")
00076
00077 self.header = options.header
00078
00079 self._mutex = threading.Lock()
00080
00081 self.match_headers = []
00082
00083 rospy.init_node('test_match_analyze')
00084 self._starttime = rospy.get_time()
00085
00086 sub_agg = rospy.Subscriber("/diagnostics_agg", DiagnosticArray, self.diag_agg_cb)
00087
00088 def diag_agg_cb(self, msg):
00089 with self._mutex:
00090 for stat in msg.status:
00091 if stat.name.find(MATCH_NAME) > 0:
00092 if self.match_headers.count(get_header_name(stat.name)) == 0:
00093 self.match_headers.append(get_header_name(stat.name))
00094
00095 def test_match_analyze(self):
00096 while not rospy.is_shutdown():
00097 sleep(1.0)
00098 if rospy.get_time() - self._starttime > DURATION:
00099 break
00100
00101 self.assert_(not rospy.is_shutdown(), "Rospy shutdown!")
00102
00103 with self._mutex:
00104 self.assert_(self.header, "Header is none. Option --header not given")
00105 self.assert_(len(self.match_headers) == 1, "Multiple analyzers reported our item! Headers: %s" % self.match_headers)
00106 self.assert_(self.match_headers.count(self.header) > 0, "Didn't have item under header \"%s\". Header: \"%s\"" % (self.header, self.match_headers[0]))
00107
00108
00109 if __name__ == '__main__':
00110 if False:
00111 suite = unittest.TestSuite()
00112 suite.addTest(TestMatchAnalyze('test_match_analyze'))
00113 unittest.TextTestRunner(verbosity = 2).run(suite)
00114 else:
00115 rostest.run(PKG, sys.argv[0], TestMatchAnalyze, sys.argv)