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