# 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