xref: /aosp_15_r20/tools/asuite/atest/test_runners/event_handler.py (revision c2e18aaa1096c836b086f94603d04f4eb9cf37f5)

# Copyright 2019, The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Atest test event handler class."""

from __future__ import print_function

from collections import deque
from datetime import timedelta
import logging
import time

from atest import atest_execution_info
from atest import atest_utils
from atest import result_reporter
from atest.test_runners import test_runner_base


EVENT_NAMES = {
    'module_started': 'TEST_MODULE_STARTED',
    'module_ended': 'TEST_MODULE_ENDED',
    'run_started': 'TEST_RUN_STARTED',
    'run_ended': 'TEST_RUN_ENDED',
    # Next three are test-level events
    'test_started': 'TEST_STARTED',
    'test_failed': 'TEST_FAILED',
    'test_ended': 'TEST_ENDED',
    'invocation_ended': 'INVOCATION_ENDED',
    # Last two failures are runner-level, not test-level.
    # Invocation failure is broader than run failure.
    'run_failed': 'TEST_RUN_FAILED',
    'invocation_failed': 'INVOCATION_FAILED',
    'test_ignored': 'TEST_IGNORED',
    'test_assumption_failure': 'TEST_ASSUMPTION_FAILURE',
    'log_association': 'LOG_ASSOCIATION',
}

EVENT_PAIRS = {
    EVENT_NAMES['module_started']: EVENT_NAMES['module_ended'],
    EVENT_NAMES['run_started']: EVENT_NAMES['run_ended'],
    EVENT_NAMES['test_started']: EVENT_NAMES['test_ended'],
}
START_EVENTS = list(EVENT_PAIRS.keys())
END_EVENTS = list(EVENT_PAIRS.values())
TEST_NAME_TEMPLATE = '%s#%s'

# time in millisecond.
ONE_SECOND = 1000
ONE_MINUTE = 60000
ONE_HOUR = 3600000

CONNECTION_STATE = {
    'current_test': None,
    'test_run_name': None,
    'last_failed': None,
    'last_ignored': None,
    'last_assumption_failed': None,
    'current_group': None,
    'current_group_total': None,
    'test_count': 0,
    'test_start_time': None,
}


class EventHandleError(Exception):
  """Raised when handle event error."""


class EventHandler:
  """Test Event handle class."""

  def __init__(self, reporter, name):
    self.reporter = reporter
    self.runner_name = name
    self.state = CONNECTION_STATE.copy()
    self.event_stack = deque()
    self.log_associations = []
    self.run_num = 0

  def _module_started(self, event_data):
    if atest_execution_info.PREPARE_END_TIME is None:
      atest_execution_info.PREPARE_END_TIME = time.time()
    self.state['current_group'] = event_data['moduleName']
    self.state['last_failed'] = None
    self.state['current_test'] = None

  def _run_started(self, event_data):
    # Technically there can be more than one run per module.
    self.run_num = event_data.get('runAttempt', 0)
    self.state['test_run_name'] = event_data.setdefault('runName', '')
    self.state['current_group_total'] = event_data['testCount']
    self.state['test_count'] = 0
    self.state['last_failed'] = None
    self.state['current_test'] = None

  def _test_started(self, event_data):
    name = TEST_NAME_TEMPLATE % (
        event_data['className'],
        event_data['testName'],
    )
    self.state['current_test'] = name
    self.state['test_count'] += 1
    self.state['test_start_time'] = event_data['start_time']

  def _test_failed(self, event_data):
    self.state['last_failed'] = {
        'name': TEST_NAME_TEMPLATE % (
            event_data['className'],
            event_data['testName'],
        ),
        'trace': event_data['trace'],
    }

  def _test_ignored(self, event_data):
    name = TEST_NAME_TEMPLATE % (
        event_data['className'],
        event_data['testName'],
    )
    self.state['last_ignored'] = name

  def _test_assumption_failure(self, event_data):
    name = TEST_NAME_TEMPLATE % (
        event_data['className'],
        event_data['testName'],
    )
    self.state['last_assumption_failed'] = name

  def _run_failed(self, event_data):
    # Module and Test Run probably started, but failure occurred.
    self.reporter.process_test_result(
        test_runner_base.TestResult(
            runner_name=self.runner_name,
            group_name=self.state['current_group'],
            test_name=self.state['current_test'],
            status=test_runner_base.ERROR_STATUS,
            details=event_data['reason'],
            test_count=self.state['test_count'],
            test_time='',
            runner_total=None,
            group_total=self.state['current_group_total'],
            additional_info={},
            test_run_name=self.state['test_run_name'],
        )
    )

  def _invocation_failed(self, event_data):
    # Broadest possible failure. May not even start the module/test run.
    self.reporter.process_test_result(
        test_runner_base.TestResult(
            runner_name=self.runner_name,
            group_name=self.state['current_group'],
            test_name=self.state['current_test'],
            status=test_runner_base.ERROR_STATUS,
            details=event_data['cause'],
            test_count=self.state['test_count'],
            test_time='',
            runner_total=None,
            group_total=self.state['current_group_total'],
            additional_info={},
            test_run_name=self.state['test_run_name'],
        )
    )

  def _invocation_ended(self, event_data):
    self.reporter.device_count = event_data.get('device_count', None)

  # pylint: disable=unused-argument
  def _run_ended(self, event_data):
    # Renew ResultReport if is module level(reporter.silent=False)
    if not self.reporter.silent:
      self.reporter.set_current_iteration_summary(self.run_num)
      self.reporter = result_reporter.ResultReporter(silent=False)

  def _module_ended(self, event_data):
    pass

  def _test_ended(self, event_data):
    name = TEST_NAME_TEMPLATE % (
        event_data['className'],
        event_data['testName'],
    )
    test_time = ''
    if self.state['test_start_time']:
      test_time = self._calc_duration(
          event_data['end_time'] - self.state['test_start_time']
      )
    if self.state['last_failed'] and name == self.state['last_failed']['name']:
      status = test_runner_base.FAILED_STATUS
      trace = self.state['last_failed']['trace']
      self.state['last_failed'] = None
    elif (
        self.state['last_assumption_failed']
        and name == self.state['last_assumption_failed']
    ):
      status = test_runner_base.ASSUMPTION_FAILED
      self.state['last_assumption_failed'] = None
      trace = None
    elif self.state['last_ignored'] and name == self.state['last_ignored']:
      status = test_runner_base.IGNORED_STATUS
      self.state['last_ignored'] = None
      trace = None
    else:
      status = test_runner_base.PASSED_STATUS
      trace = None

    default_event_keys = ['className', 'end_time', 'testName']
    additional_info = {}
    for event_key in event_data.keys():
      if event_key not in default_event_keys:
        additional_info[event_key] = event_data.get(event_key, None)

    self.reporter.process_test_result(
        test_runner_base.TestResult(
            runner_name=self.runner_name,
            group_name=self.state['current_group'],
            test_name=name,
            status=status,
            details=trace,
            test_count=self.state['test_count'],
            test_time=test_time,
            runner_total=None,
            additional_info=additional_info,
            group_total=self.state['current_group_total'],
            test_run_name=self.state['test_run_name'],
        )
    )

  def _log_association(self, event_data):
    event_data.setdefault('time', time.time())
    self.log_associations.append(event_data)

  switch_handler = {
      EVENT_NAMES['module_started']: _module_started,
      EVENT_NAMES['run_started']: _run_started,
      EVENT_NAMES['test_started']: _test_started,
      EVENT_NAMES['test_failed']: _test_failed,
      EVENT_NAMES['test_ignored']: _test_ignored,
      EVENT_NAMES['test_assumption_failure']: _test_assumption_failure,
      EVENT_NAMES['run_failed']: _run_failed,
      EVENT_NAMES['invocation_failed']: _invocation_failed,
      EVENT_NAMES['invocation_ended']: _invocation_ended,
      EVENT_NAMES['test_ended']: _test_ended,
      EVENT_NAMES['run_ended']: _run_ended,
      EVENT_NAMES['module_ended']: _module_ended,
      EVENT_NAMES['log_association']: _log_association,
  }

  def process_event(self, event_name, event_data):
    """Process the events of the test run and call reporter with results.

    Args:
        event_name: A string of the event name.
        event_data: A dict of event data.
    """
    logging.debug('Processing %s %s', event_name, event_data)
    if event_name in START_EVENTS:
      self.event_stack.append(event_name)
    elif event_name in END_EVENTS:
      self._check_events_are_balanced(event_name, self.reporter, event_data)
    if event_name in self.switch_handler:
      self.switch_handler[event_name](self, event_data)
    else:
      # TODO(b/128875503): Implement the mechanism to inform not handled
      # TF event.
      logging.debug('Event[%s] is not processable.', event_name)

  def _check_events_are_balanced(
      self, end_event_name, reporter, end_event_data
  ):
    """Check whether the Start events and End events are balanced.

    When imbalance events are detected, and we understand the case of imbalance,
    the events will handled without throwing error; otherwise EventHandleError
    will raise.

    Args:
        end_event_name: A string of the event name.
        reporter: A ResultReporter instance.
        end_event_data: A dict of event data.

    Raises:
        EventHandleError if we can't handle the imbalance of START/END events.
    """
    start_event = self.event_stack.pop() if self.event_stack else None

    def _handle_crashed_test(message):
      atest_utils.print_and_log_error(message)

      self.reporter.process_test_result(
          test_runner_base.TestResult(
              runner_name=self.runner_name,
              group_name=self.state['current_group'],
              test_name=self.state['current_test'],
              status=test_runner_base.ERROR_STATUS,
              details=message,
              test_count=self.state['test_count'],
              test_time='',
              runner_total=None,
              group_total=self.state['current_group_total'],
              additional_info={},
              test_run_name=self.state['test_run_name'],
          )
      )

    if not start_event or EVENT_PAIRS[start_event] != end_event_name:
      # Here bubble up the failed trace in the situation having
      # TEST_FAILED but never receiving TEST_ENDED.
      if self.state['last_failed'] and (
          start_event == EVENT_NAMES['test_started']
      ):
        self.reporter.process_test_result(
            test_runner_base.TestResult(
                runner_name=self.runner_name,
                group_name=self.state['current_group'],
                test_name=self.state['last_failed']['name'],
                status=test_runner_base.FAILED_STATUS,
                details=self.state['last_failed']['trace'],
                test_count=self.state['test_count'],
                test_time='',
                runner_total=None,
                group_total=self.state['current_group_total'],
                additional_info={},
                test_run_name=self.state['test_run_name'],
            )
        )
        # Even though we have proceessed the test result here, we still consider
        # this case unhandled as we don't have a full understanding about the cause.
        # So we don't return here.
        raise EventHandleError(
            'Error: Test failed without receiving a test end event'
        )
      elif (
          end_event_name == EVENT_NAMES['module_ended']
          and start_event == EVENT_NAMES['run_started']
      ):
        _handle_crashed_test(
            'Test run started but did not end. This often happens when the test'
            ' binary/app such as android instrumentation app process died.'
            ' Test count might be inaccurate.'
        )
        return
      elif (
          end_event_name == EVENT_NAMES['run_ended']
          and start_event == EVENT_NAMES['test_started']
      ):
        _handle_crashed_test(
            'Test started but did not end. This often happens when the test'
            ' binary/app such as android instrumentation app process died.'
            ' Test count might be inaccurate.'
        )
        return
      else:
        raise EventHandleError(
            'Error: Saw %s Start event and %s End event. These should be equal!'
            % (start_event, end_event_name)
        )

  @staticmethod
  def _calc_duration(duration):
    """Convert duration from ms to 3h2m43.034s.

    Args:
        duration: millisecond

    Returns:
        string in h:m:s, m:s, s or millis, depends on the duration.
    """
    delta = timedelta(milliseconds=duration)
    timestamp = str(delta).split(':')  # hh:mm:microsec

    if duration < ONE_SECOND:
      return '({}ms)'.format(duration)
    if duration < ONE_MINUTE:
      return '({:.3f}s)'.format(float(timestamp[2]))
    if duration < ONE_HOUR:
      return '({0}m{1:.3f}s)'.format(timestamp[1], float(timestamp[2]))
    return '({0}h{1}m{2:.3f}s)'.format(
        timestamp[0], timestamp[1], float(timestamp[2])
    )