tko/status_lib_unittest.py

   1 #!/usr/bin/python
   2
   3 import unittest
   4 import common
   5 from autotest_lib.tko import status_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 = status_lib.parser(0)
 177             self.assertNotEqual(p, None)
 178
 179
 180 if __name__ == "__main__":
 181     unittest.main()