xref: /aosp_15_r20/external/autotest/autotest_lib/tko/status_lib.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

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)