blob: c4b4c51baf4f0da98ac60d031f5882e8084cbf40 [file] [log] [blame]
# Copyright 2017 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import glob
import json
import os
import logging
from autotest_lib.site_utils.sponge_lib import autotest_job_info
class ACTSSummaryEnums(object):
"""A class contains the attribute names used in a ACTS summary."""
Requested = 'Requested'
Failed = 'Failed'
Unknown = 'Unknown'
class ACTSRecordEnums(object):
"""A class contains the attribute names used in an ACTS record."""
BeginTime = 'Begin Time'
Details = 'Details'
EndTime = 'End Time'
Extras = 'Extras'
ExtraErrors = 'Extra Errors'
Result = 'Result'
TestClass = 'Test Class'
TestName = 'Test Name'
class ACTSTaskInfo(autotest_job_info.AutotestTaskInfo):
"""Task info for an ACTS test."""
tags = autotest_job_info.AutotestTaskInfo.tags + ['acts', 'testtracker']
logs = autotest_job_info.AutotestTaskInfo.logs + ['results']
def __init__(self, test, job):
@param test: The autotest test for this ACTS test.
@param job: The job info that is the parent ot this task.
super(ACTSTaskInfo, self).__init__(test, job)
summary_location = os.path.join(
self.results_dir, 'results/latest/test_run_summary.json')
build_info_location = os.path.join(self.results_dir,
build_info_files = glob.iglob(build_info_location)
build_info_file = next(build_info_files)'Using build info file: %s', build_info_file)
with open(build_info_file) as fd:
self.build_info = json.load(fd)
except Exception as e:
logging.error('Bad build info file.')
self.build_info = {}
build_prop_str = self.build_info['build_prop']
prop_dict = {}
self.build_info['build_prop'] = prop_dict
lines = build_prop_str.splitlines()
for line in lines:
parts = line.split('=')
if len(parts) != 2:
prop_dict[parts[0]] = parts[1]
except Exception as e:
logging.error('Bad build prop data, using default empty dict')
self.build_info['build_prop'] = {}
with open(summary_location) as fd:
self._acts_summary = json.load(fd)
self._summary_block = self._acts_summary['Summary']
record_block = self._acts_summary['Results']
self._records = list(ACTSRecord(record) for record in record_block)
self.is_valid = True
except Exception as e:
logging.error('Bad acts data, reverting to autotest only.')
self.is_valid = False
self.tags = autotest_job_info.AutotestTaskInfo.tags
def test_case_count(self):
"""The number of test cases run."""
return self._summary_block[ACTSSummaryEnums.Requested]
def failed_case_count(self):
"""The number of failed test cases."""
return self._summary_block[ACTSSummaryEnums.Failed]
def error_case_count(self):
"""The number of errored test cases."""
return self._summary_block[ACTSSummaryEnums.Unknown]
def records(self):
"""All records of test cases in the ACTS tests."""
return self._records
def owner(self):
"""The owner of the task."""
if 'param-testtracker_owner' in self.keyvals:
return self.keyvals['param-testtracker_owner'].strip("'").strip('"')
elif 'param-test_tracker_owner' in self.keyvals:
return self.keyvals['param-testtracker_owner'].strip("'").strip('"')
return self._job.user.strip("'").strip('"')
def effort_name(self):
"""The test tracker effort name."""
build_id = self.build_info.get('build_prop', {}).get('')
if build_id and any(c.isdigit() for c in build_id):
return build_id
build_version = self.build_info.get('build_prop', {}).get(
return build_version
def project_id(self):
"""The test tracker project id."""
if 'param-testtracker_project_id' in self.keyvals:
return self.keyvals.get('param-testtracker_project_id')
return self.keyvals.get('param-test_tracker_project_id')
def environment(self):
"""The name of the enviroment for test tracker."""
build_props = self.build_info.get('build_prop', {})
if 'ro.product.board' in build_props:
board = build_props['ro.product.board']
elif '' in build_props:
board = build_props['']
return board
def extra_environment(self):
"""Extra environment info about the task."""
if 'param-testtracker_extra_env' in self.keyvals:
extra = self.keyvals.get('param-testtracker_extra_env', [])
extra = self.keyvals.get('param-test_tracker_extra_env', [])
if not isinstance(extra, list):
extra = [extra]
return extra
class ACTSRecord(object):
"""A single record of a test case in an ACTS test."""
tags = ['acts', 'testtracker']
def __init__(self, json_record):
@param json_record: The json info for this record
self._json_record = json_record
def test_class(self):
"""The test class that was run."""
return self._json_record[ACTSRecordEnums.TestClass]
def test_case(self):
"""The test case that was run. None implies all in the class."""
return self._json_record.get(ACTSRecordEnums.TestName)
def uid(self):
"""The uid of the test case."""
return self._json_record.get(ACTSRecordEnums.UID)
def status(self):
"""The status of the test case."""
return self._json_record[ACTSRecordEnums.Result]
def start_time(self):
"""The start time of the test case."""
return self._json_record[ACTSRecordEnums.BeginTime] / 1000.0
def end_time(self):
"""The end time of the test case."""
return self._json_record[ACTSRecordEnums.EndTime] / 1000.0
def details(self):
"""Details about the test case."""
return self._json_record.get(ACTSRecordEnums.Details)
def extras(self):
"""Extra info about the test case."""
return self._json_record.get(ACTSRecordEnums.Extras)
def extra_errors(self):
"""Extra errors about the test case."""
return self._json_record.get(ACTSRecordEnums.ExtraErrors)
def extra_environment(self):
"""Extra details about the environment for this test."""
extras = self.extras
if not extras:
return None
test_tracker_info = self.extras.get('test_tracker_info')
if not test_tracker_info:
return self.extras.get('test_tracker_environment_info')
return test_tracker_info.get('extra_environment')
def uuid(self):
"""The test tracker uuid of the test case."""
extras = self.extras
if not extras:
return None
test_tracker_info = self.extras.get('test_tracker_info')
if not test_tracker_info:
return self.extras.get('test_tracker_uuid')
return test_tracker_info.get('test_tracker_uuid')