#
# Copyright (C) 2017 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.
#
import importlib
import json
import logging
import os
import sys
import time
import yaml
from vts.runners.host import asserts
from vts.runners.host import base_test
from vts.runners.host import config_parser
from vts.runners.host import keys
from vts.runners.host import records
from vts.runners.host import test_runner
from vts.utils.python.io import capture_printout
from vts.utils.python.io import file_util
from mobly import test_runner as mobly_test_runner
LIST_TEST_OUTPUT_START = '==========> '
LIST_TEST_OUTPUT_END = ' <=========='
# Temp directory inside python log path. The name is required to be
# the set value for tradefed to skip reading contents as logs.
TEMP_DIR_NAME = 'temp'
CONFIG_FILE_NAME = 'test_config.yaml'
MOBLY_RESULT_JSON_FILE_NAME = 'test_run_summary.json'
MOBLY_RESULT_YAML_FILE_NAME = 'test_summary.yaml'
MOBLY_CONFIG_TEXT = '''TestBeds:
- Name: {module_name}
Controllers:
AndroidDevice:
- serial: {serial1}
- serial: {serial2}
MoblyParams:
LogPath: {log_path}
'''
#TODO(yuexima):
# 1. make DEVICES_REQUIRED configurable
# 2. add include filter function
DEVICES_REQUIRED = 2
RESULT_KEY_TYPE = 'Type'
RESULT_TYPE_SUMMARY = 'Summary'
RESULT_TYPE_RECORD = 'Record'
RESULT_TYPE_TEST_NAME_LIST = 'TestNameList'
RESULT_TYPE_CONTROLLER_INFO = 'ControllerInfo'
class MoblyTest(base_test.BaseTestClass):
'''Template class for running mobly test cases.
Attributes:
mobly_dir: string, mobly test temp directory for mobly runner
mobly_config_file_path: string, mobly test config file path
result_handlers: dict, map of result type and handler functions
'''
def setUpClass(self):
asserts.assertEqual(
len(self.android_devices), DEVICES_REQUIRED,
'Exactly %s devices are required for this test.' % DEVICES_REQUIRED
)
for ad in self.android_devices:
logging.debug('Android device serial: %s' % ad.serial)
logging.debug('Test cases: %s' % self.ListTestCases())
self.mobly_dir = os.path.join(logging.log_path, TEMP_DIR_NAME,
'mobly', str(time.time()))
file_util.Makedirs(self.mobly_dir)
logging.debug('mobly log path: %s' % self.mobly_dir)
self.result_handlers = {
RESULT_TYPE_SUMMARY: self.HandleSimplePrint,
RESULT_TYPE_RECORD: self.HandleRecord,
RESULT_TYPE_TEST_NAME_LIST: self.HandleSimplePrint,
RESULT_TYPE_CONTROLLER_INFO: self.HandleSimplePrint,
}
def tearDownClass(self):
''' Clear the mobly directory.'''
file_util.Rmdirs(self.mobly_dir, ignore_errors=True)
def PrepareConfigFile(self):
'''Prepare mobly config file for running test.'''
self.mobly_config_file_path = os.path.join(self.mobly_dir,
CONFIG_FILE_NAME)
config_text = MOBLY_CONFIG_TEXT.format(
module_name=self.test_module_name,
serial1=self.android_devices[0].serial,
serial2=self.android_devices[1].serial,
log_path=self.mobly_dir
)
with open(self.mobly_config_file_path, 'w') as f:
f.write(config_text)
def ListTestCases(self):
'''List test cases.
Returns:
List of string, test names.
'''
classes = mobly_test_runner._find_test_class()
with capture_printout.CaptureStdout() as output:
mobly_test_runner._print_test_names(classes)
test_names = []
for line in output:
if (not line.startswith(LIST_TEST_OUTPUT_START)
and line.endswith(LIST_TEST_OUTPUT_END)):
test_names.append(line)
tr_record = records.TestResultRecord(line, self.test_module_name)
self.results.requested.append(tr_record)
return test_names
def RunMoblyModule(self):
'''Execute mobly test module.'''
# Because mobly and vts uses a similar runner, both will modify
# log_path from python logging. The following step is to preserve
# log path after mobly test finishes.
# An alternative way is to start a new python process through shell
# command. In that case, test print out needs to be piped.
# This will also help avoid log overlapping
logger = logging.getLogger()
logger_path = logger.log_path
logging_path = logging.log_path
try:
mobly_test_runner.main(argv=['-c', self.mobly_config_file_path])
finally:
logger.log_path = logger_path
logging.log_path = logging_path
def GetMoblyResults(self):
'''Get mobly module run results and put in vts results.'''
file_handlers = (
(MOBLY_RESULT_YAML_FILE_NAME, self.ParseYamlResults),
(MOBLY_RESULT_JSON_FILE_NAME, self.ParseJsonResults),
)
for pair in file_handlers:
file_path = file_util.FindFile(self.mobly_dir, pair[0])
if file_path:
logging.debug('Mobly test yaml result path: %s', file_path)
pair[1](file_path)
return
asserts.fail('Mobly test result file not found.')
def generateAllTests(self):
'''Run the mobly test module and parse results.'''
#TODO(yuexima): report test names
self.PrepareConfigFile()
self.RunMoblyModule()
#TODO(yuexima): check whether DEBUG logs from mobly run are included
self.GetMoblyResults()
def ParseJsonResults(self, result_path):
'''Parse mobly test json result.
Args:
result_path: string, result json file path.
'''
with open(path, 'r') as f:
mobly_summary = json.load(f)
mobly_results = mobly_summary['Results']
for result in mobly_results:
logging.debug('Adding result for %s' % result[records.TestResultEnums.RECORD_NAME])
record = records.TestResultRecord(result[records.TestResultEnums.RECORD_NAME])
record.test_class = result[records.TestResultEnums.RECORD_CLASS]
record.begin_time = result[records.TestResultEnums.RECORD_BEGIN_TIME]
record.end_time = result[records.TestResultEnums.RECORD_END_TIME]
record.result = result[records.TestResultEnums.RECORD_RESULT]
record.uid = result[records.TestResultEnums.RECORD_UID]
record.extras = result[records.TestResultEnums.RECORD_EXTRAS]
record.details = result[records.TestResultEnums.RECORD_DETAILS]
record.extra_errors = result[records.TestResultEnums.RECORD_EXTRA_ERRORS]
self.results.addRecord(record)
def ParseYamlResults(self, result_path):
'''Parse mobly test yaml result.
Args:
result_path: string, result yaml file path.
'''
with open(result_path, 'r') as stream:
try:
docs = yaml.load_all(stream)
for doc in docs:
type = doc.get(RESULT_KEY_TYPE)
if type is None:
logging.warn(
'Mobly result document type unrecognized: %s', doc)
continue
logging.debug('Parsing result type: %s', type)
handler = self.result_handlers.get(type)
if handler is None:
logging.debug('Unknown result type: %s', type)
handler = self.HandleSimplePrint
handler(doc)
except yaml.YAMLError as exc:
print(exc)
def HandleRecord(self, doc):
'''Handle record result document type.
Args:
doc: dict, result document item
'''
logging.debug('Adding result for %s' % doc.get(records.TestResultEnums.RECORD_NAME))
record = records.TestResultRecord(doc.get(records.TestResultEnums.RECORD_NAME))
record.test_class = doc.get(records.TestResultEnums.RECORD_CLASS)
record.begin_time = doc.get(records.TestResultEnums.RECORD_BEGIN_TIME)
record.end_time = doc.get(records.TestResultEnums.RECORD_END_TIME)
record.result = doc.get(records.TestResultEnums.RECORD_RESULT)
record.uid = doc.get(records.TestResultEnums.RECORD_UID)
record.extras = doc.get(records.TestResultEnums.RECORD_EXTRAS)
record.details = doc.get(records.TestResultEnums.RECORD_DETAILS)
record.extra_errors = doc.get(records.TestResultEnums.RECORD_EXTRA_ERRORS)
# 'Stacktrace' in yaml result is ignored. 'Stacktrace' is a more
# detailed version of record.details when exception is emitted.
self.results.addRecord(record)
def HandleSimplePrint(self, doc):
'''Simply print result document to log.
Args:
doc: dict, result document item
'''
for k, v in doc.items():
logging.debug(str(k) + ": " + str(v))
def GetTestModuleNames():
'''Returns a list of mobly test module specified in test configuration.'''
configs = config_parser.load_test_config_file(sys.argv[1])
reduce_func = lambda x, y: x + y.get(keys.ConfigKeys.MOBLY_TEST_MODULE, [])
return reduce(reduce_func, configs, [])
def ImportTestModules():
'''Dynamically import mobly test modules.'''
for module_name in GetTestModuleNames():
module, cls = module_name.rsplit('.', 1)
sys.modules['__main__'].__dict__[cls] = getattr(
importlib.import_module(module), cls)
if __name__ == "__main__":
ImportTestModules()
test_runner.main()