1#!/usr/bin/python
2
3import datetime, time, unittest
4
5import common
6from autotest_lib.client.common_lib import utils
7from autotest_lib.tko.parsers import version_1
8
9
10class test_status_line(unittest.TestCase):
11    """Tests for status lines."""
12
13    statuses = ['GOOD', 'WARN', 'FAIL', 'ABORT']
14
15
16    def test_handles_start(self):
17        """Tests that START is handled properly."""
18        line = version_1.status_line(0, 'START', '----', 'test',
19                                     '', {})
20        self.assertEquals(line.type, 'START')
21        self.assertEquals(line.status, None)
22
23
24    def test_handles_info(self):
25        """Tests that INFO is handled properly."""
26        line = version_1.status_line(0, 'INFO', '----', '----',
27                                     '', {})
28        self.assertEquals(line.type, 'INFO')
29        self.assertEquals(line.status, None)
30
31
32    def test_handles_status(self):
33        """Tests that STATUS is handled properly."""
34        for stat in self.statuses:
35            line = version_1.status_line(0, stat, '----', 'test',
36                                         '', {})
37            self.assertEquals(line.type, 'STATUS')
38            self.assertEquals(line.status, stat)
39
40
41    def test_handles_endstatus(self):
42        """Tests that END is handled properly."""
43        for stat in self.statuses:
44            line = version_1.status_line(0, 'END ' + stat, '----',
45                                         'test', '', {})
46            self.assertEquals(line.type, 'END')
47            self.assertEquals(line.status, stat)
48
49
50    def test_fails_on_bad_status(self):
51        """Tests that an exception is raised on a bad status."""
52        for stat in self.statuses:
53            self.assertRaises(AssertionError,
54                              version_1.status_line, 0,
55                              'BAD ' + stat, '----', 'test',
56                              '', {})
57
58
59    def test_saves_all_fields(self):
60        """Tests that all fields are saved."""
61        line = version_1.status_line(5, 'GOOD', 'subdir_name',
62                                     'test_name', 'my reason here',
63                                     {'key1': 'value',
64                                      'key2': 'another value',
65                                      'key3': 'value3'})
66        self.assertEquals(line.indent, 5)
67        self.assertEquals(line.status, 'GOOD')
68        self.assertEquals(line.subdir, 'subdir_name')
69        self.assertEquals(line.testname, 'test_name')
70        self.assertEquals(line.reason, 'my reason here')
71        self.assertEquals(line.optional_fields,
72                          {'key1': 'value', 'key2': 'another value',
73                           'key3': 'value3'})
74
75
76    def test_parses_blank_subdir(self):
77        """Tests that a blank subdirectory is parsed properly."""
78        line = version_1.status_line(0, 'GOOD', '----', 'test',
79                                     '', {})
80        self.assertEquals(line.subdir, None)
81
82
83    def test_parses_blank_testname(self):
84        """Tests that a blank test name is parsed properly."""
85        line = version_1.status_line(0, 'GOOD', 'subdir', '----',
86                                     '', {})
87        self.assertEquals(line.testname, None)
88
89
90    def test_parse_line_smoketest(self):
91        """Runs a parse line smoke test."""
92        input_data = ('\t\t\tGOOD\t----\t----\t'
93                      'field1=val1\tfield2=val2\tTest Passed')
94        line = version_1.status_line.parse_line(input_data)
95        self.assertEquals(line.indent, 3)
96        self.assertEquals(line.type, 'STATUS')
97        self.assertEquals(line.status, 'GOOD')
98        self.assertEquals(line.subdir, None)
99        self.assertEquals(line.testname, None)
100        self.assertEquals(line.reason, 'Test Passed')
101        self.assertEquals(line.optional_fields,
102                          {'field1': 'val1', 'field2': 'val2'})
103
104    def test_parse_line_handles_newline(self):
105        """Tests that newlines are handled properly."""
106        input_data = ('\t\tGOOD\t----\t----\t'
107                      'field1=val1\tfield2=val2\tNo newline here!')
108        for suffix in ('', '\n'):
109            line = version_1.status_line.parse_line(input_data +
110                                                    suffix)
111            self.assertEquals(line.indent, 2)
112            self.assertEquals(line.type, 'STATUS')
113            self.assertEquals(line.status, 'GOOD')
114            self.assertEquals(line.subdir, None)
115            self.assertEquals(line.testname, None)
116            self.assertEquals(line.reason, 'No newline here!')
117            self.assertEquals(line.optional_fields,
118                              {'field1': 'val1',
119                               'field2': 'val2'})
120
121
122    def test_parse_line_fails_on_untabbed_lines(self):
123        """Tests that untabbed lines do not parse."""
124        input_data = '   GOOD\trandom\tfields\tof text'
125        line = version_1.status_line.parse_line(input_data)
126        self.assertEquals(line, None)
127        line = version_1.status_line.parse_line(input_data.lstrip())
128        self.assertEquals(line.indent, 0)
129        self.assertEquals(line.type, 'STATUS')
130        self.assertEquals(line.status, 'GOOD')
131        self.assertEquals(line.subdir, 'random')
132        self.assertEquals(line.testname, 'fields')
133        self.assertEquals(line.reason, 'of text')
134        self.assertEquals(line.optional_fields, {})
135
136
137    def test_parse_line_fails_on_incomplete_lines(self):
138        """Tests that incomplete lines do not parse."""
139        input_data = '\t\tGOOD\tfield\tsecond field'
140        complete_data = input_data + '\tneeded last field'
141        line = version_1.status_line.parse_line(input_data)
142        self.assertEquals(line, None)
143        line = version_1.status_line.parse_line(complete_data)
144        self.assertEquals(line.indent, 2)
145        self.assertEquals(line.type, 'STATUS')
146        self.assertEquals(line.status, 'GOOD')
147        self.assertEquals(line.subdir, 'field')
148        self.assertEquals(line.testname, 'second field')
149        self.assertEquals(line.reason, 'needed last field')
150        self.assertEquals(line.optional_fields, {})
151
152
153    def test_good_reboot_passes_success_test(self):
154        """Tests good reboot statuses."""
155        line = version_1.status_line(0, 'NOSTATUS', None, 'reboot',
156                                     'reboot success', {})
157        self.assertEquals(line.is_successful_reboot('GOOD'), True)
158        self.assertEquals(line.is_successful_reboot('WARN'), True)
159
160
161    def test_bad_reboot_passes_success_test(self):
162        """Tests bad reboot statuses."""
163        line = version_1.status_line(0, 'NOSTATUS', None, 'reboot',
164                                     'reboot success', {})
165        self.assertEquals(line.is_successful_reboot('FAIL'), False)
166        self.assertEquals(line.is_successful_reboot('ABORT'), False)
167
168
169    def test_get_kernel_returns_kernel_plus_patches(self):
170        """Tests that get_kernel returns the appropriate info."""
171        line = version_1.status_line(0, 'GOOD', 'subdir', 'testname',
172                                     'reason text',
173                                     {'kernel': '2.6.24-rc40',
174                                      'patch0': 'first_patch 0 0',
175                                      'patch1': 'another_patch 0 0'})
176        kern = line.get_kernel()
177        kernel_hash = utils.hash('md5', '2.6.24-rc40,0,0').hexdigest()
178        self.assertEquals(kern.base, '2.6.24-rc40')
179        self.assertEquals(kern.patches[0].spec, 'first_patch')
180        self.assertEquals(kern.patches[1].spec, 'another_patch')
181        self.assertEquals(len(kern.patches), 2)
182        self.assertEquals(kern.kernel_hash, kernel_hash)
183
184
185    def test_get_kernel_ignores_out_of_sequence_patches(self):
186        """Tests that get_kernel ignores patches that are out of sequence."""
187        line = version_1.status_line(0, 'GOOD', 'subdir', 'testname',
188                                     'reason text',
189                                     {'kernel': '2.6.24-rc40',
190                                      'patch0': 'first_patch 0 0',
191                                      'patch2': 'another_patch 0 0'})
192        kern = line.get_kernel()
193        kernel_hash = utils.hash('md5', '2.6.24-rc40,0').hexdigest()
194        self.assertEquals(kern.base, '2.6.24-rc40')
195        self.assertEquals(kern.patches[0].spec, 'first_patch')
196        self.assertEquals(len(kern.patches), 1)
197        self.assertEquals(kern.kernel_hash, kernel_hash)
198
199
200    def test_get_kernel_returns_unknown_with_no_kernel(self):
201        """Tests that a missing kernel is handled properly."""
202        line = version_1.status_line(0, 'GOOD', 'subdir', 'testname',
203                                     'reason text',
204                                     {'patch0': 'first_patch 0 0',
205                                      'patch2': 'another_patch 0 0'})
206        kern = line.get_kernel()
207        self.assertEquals(kern.base, 'UNKNOWN')
208        self.assertEquals(kern.patches, [])
209        self.assertEquals(kern.kernel_hash, 'UNKNOWN')
210
211
212    def test_get_timestamp_returns_timestamp_field(self):
213        """Tests that get_timestamp returns the expected info."""
214        timestamp = datetime.datetime(1970, 1, 1, 4, 30)
215        timestamp -= datetime.timedelta(seconds=time.timezone)
216        line = version_1.status_line(0, 'GOOD', 'subdir', 'testname',
217                                     'reason text',
218                                     {'timestamp': '16200'})
219        self.assertEquals(timestamp, line.get_timestamp())
220
221
222    def test_get_timestamp_returns_none_on_missing_field(self):
223        """Tests that get_timestamp returns None if no timestamp exists."""
224        line = version_1.status_line(0, 'GOOD', 'subdir', 'testname',
225                                     'reason text', {})
226        self.assertEquals(None, line.get_timestamp())
227
228
229class iteration_parse_line_into_dicts(unittest.TestCase):
230    """Tests for parsing iteration keyvals into dictionaries."""
231
232    def parse_line(self, line):
233        """
234        Invokes parse_line_into_dicts with two empty dictionaries.
235
236        @param line: The line to parse.
237
238        @return A 2-tuple representing the filled-in attr and perf dictionaries,
239            respectively.
240
241        """
242        attr, perf = {}, {}
243        version_1.iteration.parse_line_into_dicts(line, attr, perf)
244        return attr, perf
245
246
247    def test_perf_entry(self):
248        """Tests a basic perf keyval line."""
249        result = self.parse_line('perf-val{perf}=-173')
250        self.assertEqual(({}, {'perf-val': -173}), result)
251
252
253    def test_attr_entry(self):
254        """Tests a basic attr keyval line."""
255        result = self.parse_line('attr-val{attr}=173')
256        self.assertEqual(({'attr-val': '173'}, {}), result)
257
258
259    def test_untagged_is_perf(self):
260        """Tests that an untagged keyval is considered to be perf by default."""
261        result = self.parse_line('untagged=-678.5e5')
262        self.assertEqual(({}, {'untagged': -678.5e5}), result)
263
264
265    def test_invalid_tag_ignored(self):
266        """Tests that invalid tags are ignored."""
267        result = self.parse_line('bad-tag{invalid}=56')
268        self.assertEqual(({}, {}), result)
269
270
271    def test_non_numeric_perf_ignored(self):
272        """Tests that non-numeric perf values are ignored."""
273        result = self.parse_line('perf-val{perf}=FooBar')
274        self.assertEqual(({}, {}), result)
275
276
277    def test_non_numeric_untagged_ignored(self):
278        """Tests that non-numeric untagged keyvals are ignored."""
279        result = self.parse_line('untagged=FooBar')
280        self.assertEqual(({}, {}), result)
281
282
283class perf_value_iteration_parse_line_into_dict(unittest.TestCase):
284    """Tests for parsing perf value iterations into a dictionary."""
285
286    def parse_line(self, line):
287        """
288        Invokes parse_line_into_dict with a line to parse.
289
290        @param line: The string line to parse.
291
292        @return A dictionary containing the information parsed from the line.
293
294        """
295        return version_1.perf_value_iteration.parse_line_into_dict(line)
296
297    def test_invalid_json(self):
298        """Tests that a non-JSON line is handled properly."""
299        result = self.parse_line('{"invalid_json" "string"}')
300        self.assertEqual(result, {})
301
302    def test_single_value_int(self):
303        """Tests that a single integer value is parsed properly."""
304        result = self.parse_line('{"value": 7}')
305        self.assertEqual(result, {'value': 7, 'stddev': 0})
306
307    def test_single_value_float(self):
308        """Tests that a single float value is parsed properly."""
309        result = self.parse_line('{"value": 1.298}')
310        self.assertEqual(result, {'value': 1.298, 'stddev': 0})
311
312    def test_value_list_int(self):
313        """Tests that an integer list is parsed properly."""
314        result = self.parse_line('{"value": [10, 20, 30]}')
315        self.assertEqual(result, {'value': 20.0, 'stddev': 10.0})
316
317    def test_value_list_float(self):
318        """Tests that a float list is parsed properly."""
319        result = self.parse_line('{"value": [2.0, 3.0, 4.0]}')
320        self.assertEqual(result, {'value': 3.0, 'stddev': 1.0})
321
322
323class DummyAbortTestCase(unittest.TestCase):
324    """Tests for the make_dummy_abort function."""
325
326    def setUp(self):
327        self.indent = 3
328        self.testname = 'testname'
329        self.timestamp = 1220565792
330        self.reason = 'Job aborted unexpectedly'
331
332
333    def test_make_dummy_abort_with_timestamp(self):
334        """Tests make_dummy_abort with a timestamp specified."""
335        abort = version_1.parser.make_dummy_abort(
336            self.indent, None, self.testname, self.timestamp, self.reason)
337        self.assertEquals(
338            abort, '%sEND ABORT\t----\t%s\ttimestamp=%d\t%s' % (
339            '\t' * self.indent, self.testname, self.timestamp, self.reason))
340
341
342    def test_make_dummy_abort_no_timestamp(self):
343        """Tests make_dummy_abort with no timestamp specified."""
344        abort = version_1.parser.make_dummy_abort(
345            self.indent, None, self.testname, None, self.reason)
346        self.assertEquals(
347            abort, '%sEND ABORT\t----\t%s\t%s' % (
348            '\t' * self.indent, self.testname, self.reason))
349
350
351if __name__ == '__main__':
352    unittest.main()
353