blob: 72c8418097757f9a1927449054a6c9016af8e6d3 [file] [log] [blame]
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_BLOCKLIST = ('\r\x00',)
def clean_raw_line(raw_line, blocklist=DEFAULT_BLOCKLIST):
"""Strip blocklisted characters from raw_line."""
return re.sub('|'.join(blocklist), '', 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)