| #!/usr/bin/python3 |
| |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| import unittest |
| import common |
| from autotest_lib.tko import status_lib, parser_lib |
| from autotest_lib.client.common_lib import log |
| from six.moves import range |
| |
| |
| class clean_raw_line_test(unittest.TestCase): |
| def test_default(self): |
| raw_line_temp = 'this \r is a %s line \x00 yeah\n' |
| raw_line = raw_line_temp % status_lib.DEFAULT_BLOCKLIST[0] |
| cleaned = status_lib.clean_raw_line(raw_line) |
| self.assertEquals(cleaned, raw_line_temp % '') |
| |
| |
| def test_multi(self): |
| blocklist = ('\r\x00', 'FOOBAR', 'BLAh') |
| raw_line_temp = 'this \x00 FOO is BAR \r a %s line %s BL yeah %s ah\n' |
| raw_line = raw_line_temp % blocklist |
| cleaned = status_lib.clean_raw_line(raw_line, blocklist) |
| self.assertEquals( |
| cleaned, raw_line_temp % (('',) * len(blocklist))) |
| |
| |
| class line_buffer_test(unittest.TestCase): |
| def test_get_empty(self): |
| buf = status_lib.line_buffer() |
| self.assertRaises(IndexError, buf.get) |
| |
| |
| def test_get_single(self): |
| buf = status_lib.line_buffer() |
| buf.put("single line") |
| self.assertEquals(buf.get(), "single line") |
| self.assertRaises(IndexError, buf.get) |
| |
| |
| def test_is_fifo(self): |
| buf = status_lib.line_buffer() |
| lines = ["line #%d" for x in range(10)] |
| for line in lines: |
| buf.put(line) |
| results = [] |
| while buf.size(): |
| results.append(buf.get()) |
| self.assertEquals(lines, results) |
| |
| |
| def test_put_multiple_same_as_multiple_puts(self): |
| buf_put, buf_multi = [status_lib.line_buffer() |
| for x in range(2)] |
| lines = ["line #%d" % x for x in range(10)] |
| for line in lines: |
| buf_put.put(line) |
| buf_multi.put_multiple(lines) |
| counter = 0 |
| while buf_put.size(): |
| self.assertEquals(buf_put.size(), buf_multi.size()) |
| line = "line #%d" % counter |
| self.assertEquals(buf_put.get(), line) |
| self.assertEquals(buf_multi.get(), line) |
| counter += 1 |
| |
| |
| def test_put_back_is_lifo(self): |
| buf = status_lib.line_buffer() |
| lines = ["1", "2", "3"] |
| for line in lines: |
| buf.put(line) |
| results = [] |
| results.append(buf.get()) |
| buf.put_back("1") |
| buf.put_back("0") |
| while buf.size(): |
| results.append(buf.get()) |
| self.assertEquals(results, ["1", "0", "1", "2", "3"]) |
| |
| |
| def test_size_increased_by_put(self): |
| buf = status_lib.line_buffer() |
| self.assertEquals(buf.size(), 0) |
| buf.put("1") |
| buf.put("2") |
| self.assertEquals(buf.size(), 2) |
| buf.put("3") |
| self.assertEquals(buf.size(), 3) |
| |
| |
| def test_size_decreased_by_get(self): |
| buf = status_lib.line_buffer() |
| buf.put("1") |
| buf.put("2") |
| buf.put("3") |
| self.assertEquals(buf.size(), 3) |
| buf.get() |
| self.assertEquals(buf.size(), 2) |
| buf.get() |
| buf.get() |
| self.assertEquals(buf.size(), 0) |
| |
| |
| class status_stack_test(unittest.TestCase): |
| statuses = log.job_statuses |
| |
| def test_default_to_nostatus(self): |
| stack = status_lib.status_stack() |
| self.assertEquals(stack.current_status(), "NOSTATUS") |
| |
| |
| def test_default_on_start_to_nostatus(self): |
| stack = status_lib.status_stack() |
| stack.update("FAIL") |
| stack.start() |
| self.assertEquals(stack.current_status(), "NOSTATUS") |
| |
| |
| def test_size_always_at_least_zero(self): |
| stack = status_lib.status_stack() |
| self.assertEquals(stack.size(), 0) |
| stack.start() |
| stack.end() |
| self.assertEquals(stack.size(), 0) |
| stack.end() |
| self.assertEquals(stack.size(), 0) |
| |
| |
| def test_anything_overrides_nostatus(self): |
| for status in self.statuses: |
| stack = status_lib.status_stack() |
| stack.update(status) |
| self.assertEquals(stack.current_status(), status) |
| |
| |
| def test_worse_overrides_better(self): |
| for i in range(len(self.statuses)): |
| worse_status = self.statuses[i] |
| for j in range(i + 1, len(self.statuses)): |
| stack = status_lib.status_stack() |
| better_status = self.statuses[j] |
| stack.update(better_status) |
| stack.update(worse_status) |
| self.assertEquals(stack.current_status(), |
| worse_status) |
| |
| |
| def test_better_never_overrides_better(self): |
| for i in range(len(self.statuses)): |
| better_status = self.statuses[i] |
| for j in range(i): |
| stack = status_lib.status_stack() |
| worse_status = self.statuses[j] |
| stack.update(worse_status) |
| stack.update(better_status) |
| self.assertEquals(stack.current_status(), |
| worse_status) |
| |
| |
| def test_stack_is_lifo(self): |
| stack = status_lib.status_stack() |
| stack.update("GOOD") |
| stack.start() |
| stack.update("FAIL") |
| stack.start() |
| stack.update("WARN") |
| self.assertEquals(stack.end(), "WARN") |
| self.assertEquals(stack.end(), "FAIL") |
| self.assertEquals(stack.end(), "GOOD") |
| self.assertEquals(stack.end(), "NOSTATUS") |
| |
| |
| class parser_test(unittest.TestCase): |
| available_versions = [0, 1] |
| def test_can_import_available_versions(self): |
| for version in self.available_versions: |
| p = parser_lib.parser(0) |
| self.assertNotEqual(p, None) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |