atest/atest_execution_info.py

# 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 execution info generator.
"""

from __future__ import print_function

import logging
import json
import os

_ARGS_KEY = 'args'
_STATUS_PASSED_KEY = 'PASSED'
_STATUS_FAILED_KEY = 'FAILED'
_STATUS_IGNORED_KEY = 'IGNORED'
_SUMMARY_KEY = 'summary'
_TOTAL_SUMMARY_KEY = 'total_summary'
_TEST_RUNNER_KEY = 'test_runner'
_TEST_NAME_KEY = 'test_name'
_TEST_TIME_KEY = 'test_time'
_TEST_DETAILS_KEY = 'details'
_TEST_RESULT_NAME = 'test_result'

_SUMMARY_MAP_TEMPLATE = {_STATUS_PASSED_KEY : 0,
                         _STATUS_FAILED_KEY : 0,
                         _STATUS_IGNORED_KEY : 0,}


class AtestExecutionInfo(object):
    """Class that stores the whole test progress information in JSON format.

    ----
    For example, running command
        atest hello_world_test HelloWorldTest

    will result in storing the execution detail in JSON:
    {
      "args": "hello_world_test HelloWorldTest",
      "test_runner": {
          "AtestTradefedTestRunner": {
              "hello_world_test": {
                  "FAILED": [
                      {"test_time": "(5ms)",
                       "details": "Hello, Wor...",
                       "test_name": "HelloWorldTest#PrintHelloWorld"}
                      ],
                  "summary": {"FAILED": 1, "PASSED": 0, "IGNORED": 0}
              },
              "HelloWorldTests": {
                  "PASSED": [
                      {"test_time": "(27ms)",
                       "details": null,
                       "test_name": "...HelloWorldTest#testHalloWelt"},
                      {"test_time": "(1ms)",
                       "details": null,
                       "test_name": "....HelloWorldTest#testHelloWorld"}
                      ],
                  "summary": {"FAILED": 0, "PASSED": 2, "IGNORED": 0}
              }
          }
      },
      "total_summary": {"FAILED": 1, "PASSED": 2, "IGNORED": 0}
    }
    """

    result_reporters = []

    def __init__(self, args, work_dir):
        """Initialise an AtestExecutionInfo instance.

        Args:
            args: Command line parameters.
            work_dir : The directory for saving information.

        Returns:
               A json format string.
        """
        self.args = args
        self.work_dir = work_dir
        self.result_file = None

    def __enter__(self):
        """Create and return information file object."""
        full_file_name = os.path.join(self.work_dir, _TEST_RESULT_NAME)
        try:
            self.result_file = open(full_file_name, 'w')
        except IOError:
            logging.error('Cannot open file %s', full_file_name)
        return self.result_file

    def __exit__(self, exit_type, value, traceback):
        """Write execution information and close information file."""
        if self.result_file:
            self.result_file.write(AtestExecutionInfo.
                                   _generate_execution_detail(self.args))
            self.result_file.close()

    @staticmethod
    def _generate_execution_detail(args):
        """Generate execution detail.

        Args:
            args: Command line parameters that you want to save.

        Returns:
            A json format string.
        """
        info_dict = {_ARGS_KEY: ' '.join(args)}
        try:
            AtestExecutionInfo._arrange_test_result(
                info_dict,
                AtestExecutionInfo.result_reporters)
            return json.dumps(info_dict)
        except ValueError as err:
            logging.warn('Parsing test result failed due to : %s', err)

    @staticmethod
    def _arrange_test_result(info_dict, reporters):
        """Append test result information in given dict.

        Arrange test information to below
        "test_runner": {
            "test runner name": {
                "test name": {
                    "FAILED": [
                        {"test time": "",
                         "details": "",
                         "test name": ""}
                    ],
                "summary": {"FAILED": 0, "PASSED": 0, "IGNORED": 0}
                },
            },
        "total_summary": {"FAILED": 0, "PASSED": 0, "IGNORED": 0}

        Args:
            info_dict: A dict you want to add result information in.
            reporters: A list of result_reporter.

        Returns:
            A dict contains test result information data.
        """
        info_dict[_TEST_RUNNER_KEY] = {}
        for reporter in reporters:
            for test in reporter.all_test_results:
                runner = info_dict[_TEST_RUNNER_KEY].setdefault(test.runner_name, {})
                group = runner.setdefault(test.group_name, {})
                result_dict = {_TEST_NAME_KEY : test.test_name,
                               _TEST_TIME_KEY : test.test_time,
                               _TEST_DETAILS_KEY : test.details}
                group.setdefault(test.status, []).append(result_dict)

        total_test_group_summary = _SUMMARY_MAP_TEMPLATE.copy()
        for runner in info_dict[_TEST_RUNNER_KEY]:
            for group in info_dict[_TEST_RUNNER_KEY][runner]:
                group_summary = _SUMMARY_MAP_TEMPLATE.copy()
                for status in info_dict[_TEST_RUNNER_KEY][runner][group]:
                    count = len(info_dict[_TEST_RUNNER_KEY][runner][group][status])
                    if _SUMMARY_MAP_TEMPLATE.has_key(status):
                        group_summary[status] = count
                        total_test_group_summary[status] += count
                info_dict[_TEST_RUNNER_KEY][runner][group][_SUMMARY_KEY] = group_summary
        info_dict[_TOTAL_SUMMARY_KEY] = total_test_group_summary
        return info_dict