Home | History | Annotate | Download | only in tko
      1 #!/usr/bin/python
      2 
      3 import unittest
      4 import common
      5 from autotest_lib.tko import status_lib, parser_lib
      6 from autotest_lib.client.common_lib import log
      7 
      8 
      9 class clean_raw_line_test(unittest.TestCase):
     10     def test_default(self):
     11         raw_line_temp = 'this \r is a %s line \x00 yeah\n'
     12         raw_line = raw_line_temp % status_lib.DEFAULT_BLACKLIST[0]
     13         cleaned = status_lib.clean_raw_line(raw_line)
     14         self.assertEquals(cleaned, raw_line_temp % '')
     15 
     16 
     17     def test_multi(self):
     18         blacklist = ('\r\x00', 'FOOBAR', 'BLAh')
     19         raw_line_temp = 'this \x00 FOO is BAR \r a %s line %s BL yeah %s ah\n'
     20         raw_line = raw_line_temp % blacklist
     21         cleaned = status_lib.clean_raw_line(raw_line, blacklist)
     22         self.assertEquals(
     23             cleaned, raw_line_temp % (('',) * len(blacklist)))
     24 
     25 
     26 class line_buffer_test(unittest.TestCase):
     27     def test_get_empty(self):
     28         buf = status_lib.line_buffer()
     29         self.assertRaises(IndexError, buf.get)
     30 
     31 
     32     def test_get_single(self):
     33         buf = status_lib.line_buffer()
     34         buf.put("single line")
     35         self.assertEquals(buf.get(), "single line")
     36         self.assertRaises(IndexError, buf.get)
     37 
     38 
     39     def test_is_fifo(self):
     40         buf = status_lib.line_buffer()
     41         lines = ["line #%d" for x in xrange(10)]
     42         for line in lines:
     43             buf.put(line)
     44         results = []
     45         while buf.size():
     46             results.append(buf.get())
     47         self.assertEquals(lines, results)
     48 
     49 
     50     def test_put_multiple_same_as_multiple_puts(self):
     51         buf_put, buf_multi = [status_lib.line_buffer()
     52                               for x in xrange(2)]
     53         lines = ["line #%d" % x for x in xrange(10)]
     54         for line in lines:
     55             buf_put.put(line)
     56         buf_multi.put_multiple(lines)
     57         counter = 0
     58         while buf_put.size():
     59             self.assertEquals(buf_put.size(), buf_multi.size())
     60             line = "line #%d" % counter
     61             self.assertEquals(buf_put.get(), line)
     62             self.assertEquals(buf_multi.get(), line)
     63             counter += 1
     64 
     65 
     66     def test_put_back_is_lifo(self):
     67         buf = status_lib.line_buffer()
     68         lines = ["1", "2", "3"]
     69         for line in lines:
     70             buf.put(line)
     71         results = []
     72         results.append(buf.get())
     73         buf.put_back("1")
     74         buf.put_back("0")
     75         while buf.size():
     76             results.append(buf.get())
     77         self.assertEquals(results, ["1", "0", "1", "2", "3"])
     78 
     79 
     80     def test_size_increased_by_put(self):
     81         buf = status_lib.line_buffer()
     82         self.assertEquals(buf.size(), 0)
     83         buf.put("1")
     84         buf.put("2")
     85         self.assertEquals(buf.size(), 2)
     86         buf.put("3")
     87         self.assertEquals(buf.size(), 3)
     88 
     89 
     90     def test_size_decreased_by_get(self):
     91         buf = status_lib.line_buffer()
     92         buf.put("1")
     93         buf.put("2")
     94         buf.put("3")
     95         self.assertEquals(buf.size(), 3)
     96         buf.get()
     97         self.assertEquals(buf.size(), 2)
     98         buf.get()
     99         buf.get()
    100         self.assertEquals(buf.size(), 0)
    101 
    102 
    103 class status_stack_test(unittest.TestCase):
    104     statuses = log.job_statuses
    105 
    106     def test_default_to_nostatus(self):
    107         stack = status_lib.status_stack()
    108         self.assertEquals(stack.current_status(), "NOSTATUS")
    109 
    110 
    111     def test_default_on_start_to_nostatus(self):
    112         stack = status_lib.status_stack()
    113         stack.update("FAIL")
    114         stack.start()
    115         self.assertEquals(stack.current_status(), "NOSTATUS")
    116 
    117 
    118     def test_size_always_at_least_zero(self):
    119         stack = status_lib.status_stack()
    120         self.assertEquals(stack.size(), 0)
    121         stack.start()
    122         stack.end()
    123         self.assertEquals(stack.size(), 0)
    124         stack.end()
    125         self.assertEquals(stack.size(), 0)
    126 
    127 
    128     def test_anything_overrides_nostatus(self):
    129         for status in self.statuses:
    130             stack = status_lib.status_stack()
    131             stack.update(status)
    132             self.assertEquals(stack.current_status(), status)
    133 
    134 
    135     def test_worse_overrides_better(self):
    136         for i in xrange(len(self.statuses)):
    137             worse_status = self.statuses[i]
    138             for j in xrange(i + 1, len(self.statuses)):
    139                 stack = status_lib.status_stack()
    140                 better_status = self.statuses[j]
    141                 stack.update(better_status)
    142                 stack.update(worse_status)
    143                 self.assertEquals(stack.current_status(),
    144                                   worse_status)
    145 
    146 
    147     def test_better_never_overrides_better(self):
    148         for i in xrange(len(self.statuses)):
    149             better_status = self.statuses[i]
    150             for j in xrange(i):
    151                 stack = status_lib.status_stack()
    152                 worse_status = self.statuses[j]
    153                 stack.update(worse_status)
    154                 stack.update(better_status)
    155                 self.assertEquals(stack.current_status(),
    156                                   worse_status)
    157 
    158 
    159     def test_stack_is_lifo(self):
    160         stack = status_lib.status_stack()
    161         stack.update("GOOD")
    162         stack.start()
    163         stack.update("FAIL")
    164         stack.start()
    165         stack.update("WARN")
    166         self.assertEquals(stack.end(), "WARN")
    167         self.assertEquals(stack.end(), "FAIL")
    168         self.assertEquals(stack.end(), "GOOD")
    169         self.assertEquals(stack.end(), "NOSTATUS")
    170 
    171 
    172 class parser_test(unittest.TestCase):
    173     available_versions = [0, 1]
    174     def test_can_import_available_versions(self):
    175         for version in self.available_versions:
    176             p = parser_lib.parser(0)
    177             self.assertNotEqual(p, None)
    178 
    179 
    180 if __name__ == "__main__":
    181     unittest.main()
    182