blob: d80387e643047d846c27f820a33f40ef10cfcfae [file] [log] [blame]
#!/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()