xref: /aosp_15_r20/external/autotest/frontend/tko/tko_rpc_utils.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

from autotest_lib.frontend.afe import rpc_utils
from autotest_lib.client.common_lib import kernel_versions
from autotest_lib.frontend.tko import models

class TooManyRowsError(Exception):
    """
    Raised when a database query returns too many rows.
    """


class KernelString(str):
    """
    Custom string class that uses correct kernel version comparisons.
    """
    def _map(self):
        return kernel_versions.version_encode(self)


    def __hash__(self):
        return hash(self._map())


    def __eq__(self, other):
        return self._map() == other._map()


    def __ne__(self, other):
        return self._map() != other._map()


    def __lt__(self, other):
        return self._map() < other._map()


    def __lte__(self, other):
        return self._map() <= other._map()


    def __gt__(self, other):
        return self._map() > other._map()


    def __gte__(self, other):
        return self._map() >= other._map()


# SQL expression to compute passed test count for test groups
_PASS_COUNT_NAME = 'pass_count'
_COMPLETE_COUNT_NAME = 'complete_count'
_INCOMPLETE_COUNT_NAME = 'incomplete_count'
# Using COUNT instead of SUM here ensures the resulting row has the right type
# (i.e. numeric, not string).  I don't know why.
_PASS_COUNT_SQL = 'COUNT(IF(status="GOOD", 1, NULL))'
_COMPLETE_COUNT_SQL = ('COUNT(IF(status NOT IN ("TEST_NA", "RUNNING", '
                                               '"NOSTATUS"), 1, NULL))')
_INCOMPLETE_COUNT_SQL = 'COUNT(IF(status="RUNNING", 1, NULL))'
STATUS_FIELDS = {_PASS_COUNT_NAME : _PASS_COUNT_SQL,
                 _COMPLETE_COUNT_NAME : _COMPLETE_COUNT_SQL,
                 _INCOMPLETE_COUNT_NAME : _INCOMPLETE_COUNT_SQL}
_INVALID_STATUSES = ('TEST_NA', 'NOSTATUS')


def add_status_counts(group_dict, status):
    pass_count = complete_count = incomplete_count = 0
    if status == 'GOOD':
        pass_count = complete_count = 1
    elif status == 'RUNNING':
        incomplete_count = 1
    else:
        complete_count = 1
    group_dict[_PASS_COUNT_NAME] = pass_count
    group_dict[_COMPLETE_COUNT_NAME] = complete_count
    group_dict[_INCOMPLETE_COUNT_NAME] = incomplete_count
    group_dict[models.TestView.objects._GROUP_COUNT_NAME] = 1


def _construct_machine_label_header_sql(machine_labels):
    """
    Example result for machine_labels=['Index', 'Diskful']:
    CONCAT_WS(",",
              IF(FIND_IN_SET("Diskful", tko_test_attributes_host_labels.value),
                 "Diskful", NULL),
              IF(FIND_IN_SET("Index", tko_test_attributes_host_labels.value),
                 "Index", NULL))

    This would result in field values "Diskful,Index", "Diskful", "Index", NULL.
    """
    machine_labels = sorted(machine_labels)
    if_clauses = []
    for label in machine_labels:
        if_clauses.append(
            'IF(FIND_IN_SET("%s", tko_test_attributes_host_labels.value), '
               '"%s", NULL)' % (label, label))
    return 'CONCAT_WS(",", %s)' % ', '.join(if_clauses)


class GroupDataProcessor(object):
    _MAX_GROUP_RESULTS = 80000

    def __init__(self, query, group_by, header_groups, fixed_headers):
        self._query = query
        self._group_by = self.uniqify(group_by)
        self._header_groups = header_groups
        self._fixed_headers = dict((field, set(values))
                                   for field, values
                                   in fixed_headers.iteritems())

        self._num_group_fields = len(group_by)
        self._header_value_sets = [set() for i
                                   in xrange(len(header_groups))]
        self._group_dicts = []


    @staticmethod
    def uniqify(values):
        return list(set(values))


    def _restrict_header_values(self):
        for header_field, values in self._fixed_headers.iteritems():
            self._query = self._query.filter(**{header_field + '__in' : values})


    def _fetch_data(self):
        self._restrict_header_values()
        self._group_dicts = models.TestView.objects.execute_group_query(
            self._query, self._group_by)


    @staticmethod
    def _get_field(group_dict, field):
        """
        Use special objects for certain fields to achieve custom sorting.
        -Wrap kernel versions with a KernelString
        -Replace null dates with special values
        """
        value = group_dict[field]
        if field == 'kernel':
            return KernelString(value)
        if value is None: # handle null dates as later than everything else
            if field.startswith('DATE('):
                return rpc_utils.NULL_DATE
            if field.endswith('_time'):
                return rpc_utils.NULL_DATETIME
        return value


    def _process_group_dict(self, group_dict):
        # compute and aggregate header groups
        for i, group in enumerate(self._header_groups):
            header = tuple(self._get_field(group_dict, field)
                           for field in group)
            self._header_value_sets[i].add(header)
            group_dict.setdefault('header_values', []).append(header)

        # frontend's SelectionManager needs a unique ID
        group_values = [group_dict[field] for field in self._group_by]
        group_dict['id'] = str(group_values)
        return group_dict


    def _find_header_value_set(self, field):
        for i, group in enumerate(self._header_groups):
            if [field] == group:
                return self._header_value_sets[i]
        raise RuntimeError('Field %s not found in header groups %s' %
                           (field, self._header_groups))


    def _add_fixed_headers(self):
        for field, extra_values in self._fixed_headers.iteritems():
            header_value_set = self._find_header_value_set(field)
            for value in extra_values:
                header_value_set.add((value,))


    def _get_sorted_header_values(self):
        self._add_fixed_headers()
        sorted_header_values = [sorted(value_set)
                                for value_set in self._header_value_sets]
        # construct dicts mapping headers to their indices, for use in
        # replace_headers_with_indices()
        self._header_index_maps = []
        for value_list in sorted_header_values:
            index_map = dict((value, i) for i, value in enumerate(value_list))
            self._header_index_maps.append(index_map)

        return sorted_header_values


    def _replace_headers_with_indices(self, group_dict):
        group_dict['header_indices'] = [index_map[header_value]
                                        for index_map, header_value
                                        in zip(self._header_index_maps,
                                               group_dict['header_values'])]
        for field in self._group_by + ['header_values']:
            del group_dict[field]


    def process_group_dicts(self):
        self._fetch_data()
        if len(self._group_dicts) > self._MAX_GROUP_RESULTS:
            raise TooManyRowsError(
                'Query yielded %d rows, exceeding maximum %d' % (
                len(self._group_dicts), self._MAX_GROUP_RESULTS))

        for group_dict in self._group_dicts:
            self._process_group_dict(group_dict)
        self._header_values = self._get_sorted_header_values()
        if self._header_groups:
            for group_dict in self._group_dicts:
                self._replace_headers_with_indices(group_dict)


    def get_info_dict(self):
        return {'groups' : self._group_dicts,
                'header_values' : self._header_values}