1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
import collections, re
import common
from autotest_lib.client.common_lib import log
statuses = log.job_statuses
def is_worse_than(lhs, rhs):
""" Compare two statuses and return a boolean indicating if the LHS status
is worse than the RHS status."""
return (statuses.index(lhs) < statuses.index(rhs))
def is_worse_than_or_equal_to(lhs, rhs):
""" Compare two statuses and return a boolean indicating if the LHS status
is worse than or equal to the RHS status."""
if lhs == rhs:
return True
return is_worse_than(lhs, rhs)
DEFAULT_BLACKLIST = ('\r\x00',)
def clean_raw_line(raw_line, blacklist=DEFAULT_BLACKLIST):
"""Strip blacklisted characters from raw_line."""
return re.sub('|'.join(blacklist), '', raw_line)
class status_stack(object):
def __init__(self):
self.status_stack = [statuses[-1]]
def current_status(self):
return self.status_stack[-1]
def update(self, new_status):
if new_status not in statuses:
return
if is_worse_than(new_status, self.current_status()):
self.status_stack[-1] = new_status
def start(self):
self.status_stack.append(statuses[-1])
def end(self):
result = self.status_stack.pop()
if len(self.status_stack) == 0:
self.status_stack.append(statuses[-1])
return result
def size(self):
return len(self.status_stack) - 1
class line_buffer(object):
def __init__(self):
self.buffer = collections.deque()
def get(self):
return self.buffer.pop()
def put(self, line):
self.buffer.appendleft(line)
def put_multiple(self, lines):
self.buffer.extendleft(lines)
def put_back(self, line):
self.buffer.append(line)
def size(self):
return len(self.buffer)
def parser(version):
library = "autotest_lib.tko.parsers.version_%d" % version
module = __import__(library, globals(), locals(), ["parser"])
return module.parser()
|
