00001 import unittest
00002
00003 from nasa_common_logging.logging_utils import parse_log_buffer
00004
00005 class TestParseSyslogLine(unittest.TestCase):
00006
00007 def setUp(self):
00008 pass
00009
00010 def tearDown(self):
00011 pass
00012
00013 def test_parse_common_logging_basic(self):
00014 timestamp = "2016-01-01 00:00:00"
00015 hostname = "hostname"
00016 source_name = "source_name"
00017 level = "INFO"
00018 category = "category.test"
00019 message = "test message!"
00020
00021 buffer = "{} {} {}: [{}] [{}] {}".format(timestamp, hostname, source_name, level, category, message)
00022 t, host, src, lvl, cat, msg = parse_log_buffer(buffer)[0]
00023
00024 self.assertEqual(timestamp, t)
00025 self.assertEqual(hostname, host)
00026 self.assertEqual(source_name, src)
00027 self.assertEqual(level, lvl)
00028 self.assertEqual(category, cat)
00029 self.assertEqual(message, msg)
00030
00031 def test_parse_common_logging_space_in_names(self):
00032 timestamp = "2016-09-01 15:53:23"
00033 hostname = "hydrophobe"
00034 source_name = "ncl_py"
00035 level = "INFO"
00036 category = "root"
00037 message = "test"
00038
00039 buffer = "{} {} {}: [{} ] [{} ] {}\n".format(timestamp, hostname, source_name, level, category, message)
00040 t, host, src, lvl, cat, msg = parse_log_buffer(buffer)[0]
00041
00042 self.assertEqual(timestamp, t)
00043 self.assertEqual(hostname, host)
00044 self.assertEqual(source_name, src)
00045 self.assertEqual(level, lvl)
00046 self.assertEqual(category, cat)
00047 self.assertEqual(message, msg)
00048
00049
00050 def test_parse_common_logging_multiline(self):
00051 timestamp = "2016-01-01 00:00:00"
00052 hostname = "hostname"
00053 source_name = "source_name"
00054 level = "INFO"
00055 category = "category.test"
00056 message = "This is a long test message.\nIt has two lines."
00057
00058 buffer = "{} {} {}: [{}] [{}] {}".format(timestamp, hostname, source_name, level, category, message)
00059 t, host, src, lvl, cat, msg = parse_log_buffer(buffer)[0]
00060
00061 self.assertEqual(timestamp, t)
00062 self.assertEqual(hostname, host)
00063 self.assertEqual(source_name, src)
00064 self.assertEqual(level, lvl)
00065 self.assertEqual(category, cat)
00066 self.assertEqual(message, msg)
00067
00068
00069 def test_parse_common_logging_two_messages(self):
00070 timestamp1 = "2016-01-01 00:00:00"
00071 hostname1 = "hostname"
00072 source_name1 = "source_name"
00073 level1 = "INFO"
00074 category1 = "category.test"
00075 message1 = "test message!"
00076 line1 = "{} {} {}: [{}] [{}] {}".format(timestamp1, hostname1, source_name1, level1, category1, message1)
00077
00078 timestamp2 = "2016-02-02 00:00:02"
00079 hostname2 = "hostname2"
00080 source_name2 = "source_name2"
00081 level2 = "INFO2"
00082 category2 = "category.test2"
00083 message2 = "test message2!"
00084 line2 = "{} {} {}: [{}] [{}] {}".format(timestamp2, hostname2, source_name2, level2, category2, message2)
00085
00086 buffer = line1 + '\n' + line2
00087
00088 log_entries = parse_log_buffer(buffer)
00089
00090 log_entry1 = log_entries[0]
00091 self.assertEqual(timestamp1,log_entry1[0])
00092 self.assertEqual(hostname1,log_entry1[1])
00093 self.assertEqual(source_name1,log_entry1[2])
00094 self.assertEqual(level1,log_entry1[3])
00095 self.assertEqual(category1,log_entry1[4])
00096 self.assertEqual(message1,log_entry1[5])
00097
00098 log_entry2 = log_entries[1]
00099 self.assertEqual(timestamp2,log_entry2[0])
00100 self.assertEqual(hostname2,log_entry2[1])
00101 self.assertEqual(source_name2,log_entry2[2])
00102 self.assertEqual(level2,log_entry2[3])
00103 self.assertEqual(category2,log_entry2[4])
00104 self.assertEqual(message2,log_entry2[5])
00105
00106
00107 def test_parse_common_logging_multiline_messages(self):
00108 timestamp1 = "2016-01-01 00:00:00"
00109 hostname1 = "hostname"
00110 source_name1 = "source_name"
00111 level1 = "INFO"
00112 category1 = "category.test"
00113 message1 = "test message line1\nAnd line2\nand line3"
00114 line1 = "{} {} {}: [{}] [{}] {}".format(timestamp1, hostname1, source_name1, level1, category1, message1)
00115
00116 timestamp2 = "2016-02-02 00:00:02"
00117 hostname2 = "hostname2"
00118 source_name2 = "source_name2"
00119 level2 = "INFO2"
00120 category2 = "category.test2"
00121 message2 = "test message2"
00122 line2 = "{} {} {}: [{}] [{}] {}".format(timestamp2, hostname2, source_name2, level2, category2, message2)
00123
00124 buffer = line1 + '\n' + line2
00125
00126 log_entries = parse_log_buffer(buffer)
00127
00128 log_entry1 = log_entries[0]
00129 self.assertEqual(timestamp1, log_entry1[0])
00130 self.assertEqual(hostname1, log_entry1[1])
00131 self.assertEqual(source_name1, log_entry1[2])
00132 self.assertEqual(level1, log_entry1[3])
00133 self.assertEqual(category1, log_entry1[4])
00134 self.assertEqual(message1, log_entry1[5])
00135
00136 log_entry2 = log_entries[1]
00137 self.assertEqual(timestamp2, log_entry2[0])
00138 self.assertEqual(hostname2, log_entry2[1])
00139 self.assertEqual(source_name2, log_entry2[2])
00140 self.assertEqual(level2, log_entry2[3])
00141 self.assertEqual(category2, log_entry2[4])
00142 self.assertEqual(message2, log_entry2[5])
00143
00144
00145 def test_parse_syslog_basic(self):
00146 timestamp = "Jan 23 00:00:00"
00147 hostname = "hostname"
00148 source_name = "source_name"
00149 level = "INFO"
00150 category = "category.test"
00151 message = "test message!"
00152
00153 buffer = "{} {} {}: [{}] [{}] {}".format(timestamp, hostname, source_name, level, category, message)
00154 t, host, src, lvl, cat, msg = parse_log_buffer(buffer)[0]
00155
00156 self.assertEqual(timestamp, t)
00157 self.assertEqual(hostname, host)
00158 self.assertEqual(source_name, src)
00159 self.assertEqual(level, lvl)
00160 self.assertEqual(category, cat)
00161 self.assertEqual(message, msg)
00162
00163
00164 def test_parse_syslog_basic_single_digit_day(self):
00165 timestamp = "Jan 1 00:00:00"
00166 hostname = "hostname"
00167 source_name = "source_name"
00168 level = "INFO"
00169 category = "category.test"
00170 message = "test message!"
00171
00172 buffer = "{} {} {}: [{}] [{}] {}".format(timestamp, hostname, source_name, level, category, message)
00173 t, host, src, lvl, cat, msg = parse_log_buffer(buffer)[0]
00174
00175 self.assertEqual(timestamp, t)
00176 self.assertEqual(hostname, host)
00177 self.assertEqual(source_name, src)
00178 self.assertEqual(level, lvl)
00179 self.assertEqual(category, cat)
00180 self.assertEqual(message, msg)
00181
00182
00183
00184 def test_parse_syslog_basic_single_warn(self):
00185 timestamp = "Jan 1 00:00:00"
00186 hostname = "hostname"
00187 source_name = "source_name"
00188 level = "WARN"
00189 category = "category.test"
00190 message = "test message!"
00191
00192 buffer = "{} {} {}: [{}] [{}] {}".format(timestamp, hostname, source_name, level, category, message)
00193 t, host, src, lvl, cat, msg = parse_log_buffer(buffer)[0]
00194
00195 self.assertEqual(timestamp, t)
00196 self.assertEqual(hostname, host)
00197 self.assertEqual(source_name, src)
00198 self.assertEqual(level, lvl)
00199 self.assertEqual(category, cat)
00200 self.assertEqual(message, msg)
00201
00202
00203 def test_mix_log_types(self):
00204 timestamp1 = "Jan 01 01:01:01"
00205 hostname1 = "hostname"
00206 source_name1 = "source_name"
00207 level1 = "INFO"
00208 category1 = "category.test"
00209 message1 = "test message!"
00210 line1 = "{} {} {}: [{}] [{}] {}".format(timestamp1, hostname1, source_name1, level1, category1, message1)
00211
00212 timestamp2 = "2016-02-02 00:00:02"
00213 hostname2 = "hostname2"
00214 source_name2 = "source_name2"
00215 level2 = "INFO2"
00216 category2 = "category.test2"
00217 message2 = "test message2!"
00218 line2 = "{} {} {}: [{}] [{}] {}".format(timestamp2, hostname2, source_name2, level2, category2, message2)
00219
00220 buffer = line1 + '\n' + line2
00221
00222 log_entries = parse_log_buffer(buffer)
00223
00224 log_entry1 = log_entries[0]
00225 self.assertEqual(timestamp1, log_entry1[0])
00226 self.assertEqual(hostname1, log_entry1[1])
00227 self.assertEqual(source_name1, log_entry1[2])
00228 self.assertEqual(level1, log_entry1[3])
00229 self.assertEqual(category1, log_entry1[4])
00230 self.assertEqual(message1, log_entry1[5])
00231
00232 log_entry2 = log_entries[1]
00233 self.assertEqual(timestamp2, log_entry2[0])
00234 self.assertEqual(hostname2, log_entry2[1])
00235 self.assertEqual(source_name2, log_entry2[2])
00236 self.assertEqual(level2, log_entry2[3])
00237 self.assertEqual(category2, log_entry2[4])
00238 self.assertEqual(message2, log_entry2[5])
00239
00240 def test_parse_generic_logging_basic(self):
00241 timestamp = "2016-01-01 00:00:00"
00242 hostname = "hostname"
00243 source_name = "source_name"
00244 message = "test message!"
00245
00246 buffer = "{} {} {}: {}".format(timestamp, hostname, source_name, message)
00247 t, host, src, lvl, cat, msg = parse_log_buffer(buffer)[0]
00248
00249 self.assertEqual(timestamp, t)
00250 self.assertEqual(hostname, host)
00251 self.assertEqual(source_name, src)
00252 self.assertEqual(None, lvl)
00253 self.assertEqual(None, cat)
00254 self.assertEqual(message, msg)
00255
00256 def test_parse_generic_logging_multi_line(self):
00257 timestamp = "2016-01-01 00:00:00"
00258 hostname = "hostname"
00259 source_name = "source_name"
00260 message = "test message!\nTwo lines long"
00261
00262 buffer = "{} {} {}: {}".format(timestamp, hostname, source_name, message)
00263 t, host, src, lvl, cat, msg = parse_log_buffer(buffer)[0]
00264
00265 self.assertEqual(timestamp, t)
00266 self.assertEqual(hostname, host)
00267 self.assertEqual(source_name, src)
00268 self.assertEqual(None, lvl)
00269 self.assertEqual(None, cat)
00270 self.assertEqual(message, msg)
00271
00272 if __name__ == "__main__":
00273 unittest.main()