aboutsummaryrefslogtreecommitdiff
path: root/lava_android_test/commands.py
diff options
context:
space:
mode:
Diffstat (limited to 'lava_android_test/commands.py')
-rw-r--r--lava_android_test/commands.py935
1 files changed, 935 insertions, 0 deletions
diff --git a/lava_android_test/commands.py b/lava_android_test/commands.py
new file mode 100644
index 0000000..eb46c78
--- /dev/null
+++ b/lava_android_test/commands.py
@@ -0,0 +1,935 @@
+# Copyright (c) 2011, 2012 Linaro
+#
+# Author: Linaro Validation Team <linaro-dev@lists.linaro.org>
+#
+# This file is part of LAVA Android Test.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+import base64
+import os
+import re
+import urlparse
+import versiontools
+import zipfile
+
+from tempfile import mkdtemp
+from uuid import uuid4
+
+from lava_tool.interface import Command as LAVACommand
+from lava_tool.interface import LavaCommandError
+from linaro_dashboard_bundle.io import DocumentIO
+
+from lava_android_test.adb import ADB
+from lava_android_test.config import get_config
+from lava_android_test.provider import TestProvider
+from lava_android_test.repository import GitRepository
+from lava_android_test.testdef import AndroidTest
+from lava_android_test.testdef import AndroidTestRunner, \
+ AndroidTestInstaller, \
+ AndroidTestParser
+from lava_android_test import utils
+
+
+class Command(LAVACommand):
+
+ def __init__(self, parser, args):
+ super(Command, self).__init__(parser, args)
+ self.config = get_config()
+# self._test_loader = TestLoader(self._config)
+
+ @classmethod
+ def register_arguments(cls, parser):
+ parser.add_argument(
+ "-q", "--quiet",
+ action="store_true",
+ default=False,
+ help="Be less verbose about undertaken actions")
+ parser.add_argument(
+ "-Q", "--quiet-subcommands",
+ action="store_true",
+ default=False,
+ help="Hide the output of all sub-commands (including tests)")
+
+ def say(self, text, *args, **kwargs):
+ print "LAVA:", text.format(*args, **kwargs)
+
+ def say_begin(self, text, *args, **kwargs):
+ print "LAVA: --Start Operation: ", text.format(*args, **kwargs)
+
+ def say_end(self, text, *args, **kwargs):
+ print "LAVA: --End Operation: ", text.format(*args, **kwargs),
+
+ def display_subprocess_output(self, stream_name, line):
+ if self.args.quiet_subcommands:
+ return
+ if stream_name == 'stdout':
+ self.say('(stdout) {0}', line.rstrip())
+ elif stream_name == 'stderr':
+ self.say('(stderr) {0}', line.rstrip())
+
+
+class extract_attachments(Command):
+ """
+ Extract the attachment files from the result bundle file
+ generated by the -o option of run command
+ or from the output of parse command
+ """
+ @classmethod
+ def register_arguments(self, parser):
+ super(extract_attachments, self).register_arguments(parser)
+ parser.add_argument("result_file",
+ help="The result bundle json file")
+ group = parser.add_argument_group("specify the output directory")
+ group.add_argument("-d", "--directory",
+ default=None,
+ metavar="directory",
+ help=("specify the directory where all the "
+ "attachment files will be put"))
+
+ def invoke(self):
+
+ if not os.path.exists(self.args.result_file):
+ raise LavaCommandError("The specified result file(%s) "
+ "does not exist." % self.args.result_file)
+ msg = "extract attachment file from result bundle file(%s)" % (
+ self.args.result_file)
+ self.say_begin(msg)
+ badchars = "[^a-zA-Z0-9\._-]"
+ with open(self.args.result_file) as stream:
+ jobdata = stream.read()
+ result_data = DocumentIO.loads(jobdata)[1]
+ test_runs = result_data.get('test_runs')
+ if not self.args.directory:
+ attachment_dir = mkdtemp(prefix='attachments-',
+ dir=os.path.curdir)
+ elif not os.path.exists(self.args.directory):
+ os.makedirs(self.args.directory)
+ attachment_dir = self.args.directory
+ elif not os.path.isdir(self.args.directory):
+ raise LavaCommandError(
+ "The specified path(%s) is not a directory."
+ % self.args.directory)
+ else:
+ attachment_dir = self.args.directory
+
+ for test in test_runs:
+ test_id = test.get('test_id').replace(" ", "_")
+ test_id = re.sub(badchars, "_", test_id)
+ target_dir = mkdtemp(prefix='%s' % test_id, dir=attachment_dir)
+ print "The test id is: %s" % test_id
+ attachments = test.get('attachments', [])
+ for attach in attachments:
+ pathname = attach.get('pathname')
+ file_name = os.path.basename(pathname)
+ content_decoded = base64.standard_b64decode(
+ attach.get("content"))
+ with open(os.path.join(target_dir, file_name), 'w') as fd:
+ fd.write(content_decoded)
+ self.say("All attachment files are put under directory(%s)" %
+ (attachment_dir))
+ self.say_end(msg)
+
+
+class list_devices(Command):
+ """
+ List available devices
+ program::lava-android-test list-devices
+ """
+
+ def invoke(self):
+
+ self.adb = ADB()
+ try:
+ output = self.adb.devices()[1]
+ if output is not None:
+ for line in output:
+ print line.strip()
+ else:
+ print "No device attached"
+ except OSError:
+ print "No device attached"
+
+
+class list_tests(Command):
+ """
+ List available tests
+ program:: lava-android-test list-tests
+ """
+
+ def invoke(self):
+ self.say("Known tests:")
+ for provider in TestProvider().get_test_provider_list():
+ for test in provider().list_test():
+ self.say(" - {test_id}", test_id=test)
+
+
+class version(Command):
+ """
+ Show LAVA Test version
+ """
+
+ def invoke(self):
+ self.say("version details:")
+ for framework in self._get_frameworks():
+ self.say(" - {framework}: {version}",
+ framework=framework.__name__,
+ version=versiontools.format_version(
+ framework.__version__, framework))
+
+ def _get_frameworks(self):
+ import lava_tool
+ import lava_android_test
+ import linaro_dashboard_bundle
+ return [
+ lava_android_test,
+ lava_tool,
+ linaro_dashboard_bundle]
+
+
+class AndroidCommand(Command):
+
+ @classmethod
+ def register_arguments(self, parser):
+ super(AndroidCommand, self).register_arguments(parser)
+ group = parser.add_argument_group("specify device serial number")
+ group.add_argument("-s", "--serial",
+ default=None,
+ metavar="serial",
+ help=("specify the device with serial number"
+ "that this command will be run on"))
+
+ def test_installed(self, test_id):
+ if self.adb is None:
+ self.adb = ADB()
+ test_dir = os.path.join(self.config.installdir_android, test_id)
+ return self.adb.exists(test_dir)
+
+ def get_device_serial(self):
+ return ADB(self.args.serial).get_serial()
+
+ def assertDeviceIsConnected(self):
+ if not self.adb.isDeviceConnected():
+ if self.adb.serial:
+ raise Exception("Device '%s' is not connected" %
+ self.adb.serial)
+ else:
+ raise Exception("No device found")
+
+ def invoke(self):
+ serial = self.get_device_serial()
+ if not serial:
+ raise LavaCommandError("No device attached")
+ self.serial = serial
+ self.adb = ADB(self.serial)
+
+ try:
+ self.assertDeviceIsConnected()
+ except Exception as err:
+ raise LavaCommandError(err)
+
+ self.invoke_sub()
+
+ def invoke_sub(self):
+ raise NotImplementedError
+
+
+class AndroidTestCommand(AndroidCommand):
+ @classmethod
+ def register_arguments(self, parser):
+ super(AndroidTestCommand, self).register_arguments(parser)
+ parser.add_argument("test_id",
+ help="Test identifier")
+
+ def get_tip_msg(self, text):
+ if self.args.serial:
+ tip_msg = "%s (%s) on device(%s)" % (text,
+ self.args.test_id,
+ self.args.serial)
+ else:
+ tip_msg = "%s (%s)" % (text, self.args.test_id)
+ return tip_msg
+
+
+class AndroidResultCommand(AndroidCommand):
+ @classmethod
+ def register_arguments(self, parser):
+ super(AndroidResultCommand, self).register_arguments(parser)
+ parser.add_argument("result_id",
+ help="Test result identifier")
+
+
+class AndroidResultsCommand(AndroidCommand):
+ @classmethod
+ def register_arguments(self, parser):
+ super(AndroidResultsCommand, self).register_arguments(parser)
+ parser.add_argument("result_id", nargs="+",
+ help="One or more result identifiers")
+
+
+class list_installed(AndroidCommand):
+ """
+ List installed tests for specified device.
+ program:: lava-android-test list-tests
+ program:: lava-android-test list-tests -s device_serial
+ """
+ def invoke_sub(self):
+
+ self.say("Installed tests:")
+ try:
+ output = self.adb.listdir(self.config.installdir_android)[1]
+ if output is not None:
+ for dir_name in output:
+ self.say(" - {test_id}", test_id=dir_name.strip())
+ else:
+ self.say("No tests installed")
+ except OSError:
+ self.say("No tests installed")
+
+
+class list_results(AndroidCommand):
+ """
+ List results of tests that has been run on the specified device.
+ program:: lava-android-test list-results
+ program:: lava-android-test list-results -s device_serial
+ """
+ def invoke_sub(self):
+ self.say("Saved results:")
+ try:
+ (ret_code, output) = self.adb.listdir(
+ self.config.resultsdir_android)
+ if ret_code != 0:
+ raise OSError()
+ for dir_name in output:
+ self.say(" - {result_id}", result_id=dir_name.strip())
+ except OSError:
+ self.say("No results found")
+
+
+class install(AndroidTestCommand):
+ """
+ Install test to the specified device.
+ program:: lava-android-test install test-id
+ program:: lava-android-test install test-id -s device_serial
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(cls, install).register_arguments(parser)
+ parser.add_argument('-o', '--install-option')
+
+ def invoke_sub(self):
+ tip_msg = self.get_tip_msg("Install test")
+ self.say_begin(tip_msg)
+
+ if self.test_installed(self.args.test_id):
+ raise LavaCommandError("The test (%s) has already installed." %
+ self.args.test_id)
+ test = TestProvider().load_test(self.args.test_id, self.args.serial)
+ try:
+ test.install(self.args.install_option)
+ except Exception as strerror:
+ raise LavaCommandError("Test installation error: %s" % strerror)
+
+ self.say_end(tip_msg)
+
+
+class uninstall(AndroidTestCommand):
+ """
+ Unistall test of the specified device.
+ program:: lava-android-test uninstall test-id
+ program:: lava-android-test uninstall test-id -s device_serial
+ """
+ def invoke_sub(self):
+ tip_msg = self.get_tip_msg("Uninstall test")
+ self.say_begin(tip_msg)
+
+ test = TestProvider().load_test(self.args.test_id, self.args.serial)
+ try:
+ test.uninstall()
+ except Exception as strerror:
+ raise LavaCommandError("Test uninstall error: %s" % strerror)
+ self.say_end(tip_msg)
+
+
+class run(AndroidTestCommand):
+ """
+ Run a previously installed test program on the specified device
+ program:: lava-android-test run test-id
+ program:: lava-android-test run test-id -s device_serial
+ program:: lava-android-test run test-id -s device_serial -o outputfile
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(run, cls).register_arguments(parser)
+ parser.add_argument('-O', '--run-option',
+ help=("Specified in the job file for using in "
+ "the real test action, so that we can customize"
+ " some test when need"))
+ group = parser.add_argument_group("specify the bundle output file")
+ group.add_argument("-o", "--output",
+ default=None,
+ metavar="FILE",
+ help=("After running the test parse the result"
+ " artefacts, fuse them with the initial"
+ " bundle and finally save the complete bundle"
+ " to the specified FILE."))
+
+ def invoke_sub(self):
+ tip_msg = self.get_tip_msg("Run test")
+ self.say_begin(tip_msg)
+
+ if not self.test_installed(self.args.test_id):
+ raise LavaCommandError(
+ "The test (%s) has not been installed yet." %
+ self.args.test_id)
+ test = TestProvider().load_test(self.args.test_id, self.args.serial)
+
+ if not self.test_installed(test.testname):
+ raise LavaCommandError(
+ "The test (%s) has not been installed yet."
+ % self.args.test_id)
+
+ try:
+ result_id = test.run(quiet=self.args.quiet,
+ run_options=self.args.run_option)
+ if self.args.output:
+ output_dir = os.path.dirname(self.args.output)
+ if output_dir and (not os.path.exists(output_dir)):
+ os.makedirs(output_dir)
+ bundle = generate_bundle(self.args.serial, result_id)
+ with open(self.args.output, "wt") as stream:
+ DocumentIO.dump(stream, bundle)
+
+ except Exception as strerror:
+ raise LavaCommandError("Test execution error: %s" % strerror)
+
+ self.say_end(tip_msg)
+
+
+class run_custom(AndroidCommand):
+ """
+ Run the command(s) that specified by the -c option in the command line
+ program:: lava-android-test run-custom -c 'cm1' -c 'cmd2' -p 'parse-regex1'
+ program:: lava-android-test run test-id -s device_serial
+ program:: lava-android-test run test-id -s device_serial -o outputfile
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(run_custom, cls).register_arguments(parser)
+ group = parser.add_mutually_exclusive_group(required=True)
+ group.add_argument('-c', '--android-command', action='append',
+ help=("Specified in the job file for using"
+ " in the real test action, so that "
+ "we can customize some test when need"))
+ group.add_argument('-f', '--command-file',
+ help=("Specified the command file that will be "
+ "pushed into android and run."))
+ parser.add_argument('-p', '--parse-regex',
+ help=("Specified the regular expression used"
+ " for analyzing command output"))
+ group = parser.add_argument_group("specify the bundle output file")
+ group.add_argument("-o", "--output",
+ default=None,
+ metavar="FILE",
+ help=("After running the test parse the result"
+ " artefacts, fuse them with the initial"
+ " bundle and finally save the complete bundle"
+ " to the specified FILE."))
+
+ def invoke_sub(self):
+
+ test_name = 'custom'
+ ADB_SHELL_STEPS = []
+ STEPS_HOST_PRE = []
+ STEPS_ADB_PRE = []
+ file_name = None
+ if self.args.android_command:
+ ADB_SHELL_STEPS = self.args.android_command
+ cmds_str = ','.join(ADB_SHELL_STEPS)
+ if len(cmds_str) > 40:
+ cmds_str = '%s...' % (cmds_str[:40])
+ test_name_suffix = 'command=[%s]' % (cmds_str)
+ elif self.args.command_file:
+ file_url = self.args.command_file
+ urlpath = urlparse.urlsplit(file_url).path
+ file_name = os.path.basename(urlpath)
+ target_path = os.path.join(self.config.installdir_android,
+ test_name, file_name)
+ STEPS_HOST_PRE = ["wget %s -O %s" % (file_url, file_name)]
+ STEPS_ADB_PRE = ["push %s %s" % (file_name, target_path)]
+ ADB_SHELL_STEPS = ["chmod 777 %s" % target_path,
+ target_path]
+ file_name_str = file_name
+ if len(file_name_str) > 40:
+ file_name_str = '%s...' % (cmds_str[:40])
+ test_name_suffix = 'command_file=%s' % (file_name_str)
+
+ PATTERN = None
+ if self.args.parse_regex:
+ PATTERN = self.args.parse_regex
+
+ tip_msg = ''
+ if self.args.serial:
+ tip_msg = ("Run following custom test(s) on device(%s):"
+ "\n\tcommands=%s"
+ "\n\tcommand-file=%s\n") % (
+ self.args.serial,
+ '\n\t\t'.join(ADB_SHELL_STEPS),
+ file_name)
+ else:
+ tip_msg = ("Run following custom test(s):"
+ "\n\t\tcommands=%s"
+ "\n\tcommand-file=%s\n") % (
+ '\n\t\t'.join(ADB_SHELL_STEPS),
+ file_name)
+
+ self.say_begin(tip_msg)
+
+ inst = AndroidTestInstaller()
+
+ run = AndroidTestRunner(steps_host_pre=STEPS_HOST_PRE,
+ steps_adb_pre=STEPS_ADB_PRE,
+ adbshell_steps=ADB_SHELL_STEPS)
+ parser = AndroidTestParser(pattern=PATTERN)
+ test = AndroidTest(testname=test_name,
+ installer=inst, runner=run, parser=parser)
+ test.parser.results = {'test_results': []}
+ test.setadb(self.adb)
+
+ if not self.test_installed(test.testname):
+ test.install()
+
+ try:
+ result_id = test.run(quiet=self.args.quiet)
+ if self.args.output:
+ output_dir = os.path.dirname(self.args.output)
+ if output_dir and (not os.path.exists(output_dir)):
+ os.makedirs(output_dir)
+ bundle = generate_bundle(self.args.serial,
+ result_id, test=test,
+ test_id='%s(%s)' % (test_name, test_name_suffix))
+ with open(self.args.output, "wt") as stream:
+ DocumentIO.dump(stream, bundle)
+
+ except Exception as strerror:
+ raise LavaCommandError("Test execution error: %s" % strerror)
+ self.say_end(tip_msg)
+
+
+class run_monkeyrunner(AndroidCommand):
+ """
+ Run the monkeyrunner scripts that stored in the specified git repository
+ program:: lava-android-test run-monkeyrunner -g giturl -r resultfilelist
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(run_monkeyrunner, cls).register_arguments(parser)
+ parser.add_argument("url",
+ help="The repository url of the test scripts")
+ parser.add_argument('-t', '--repo-type',
+ default='git',
+ help=("Specify the type of the repository"))
+ group = parser.add_argument_group("specify the bundle output file")
+ group.add_argument("-o", "--output",
+ default=None,
+ metavar="FILE",
+ help=("After running the test parse the result"
+ " artefacts, fuse them with the initial"
+ " bundle and finally save the complete bundle"
+ " to the specified FILE."))
+
+ def invoke_sub(self):
+
+ if not utils.check_command_exist('monkeyrunner'):
+ raise LavaCommandError('The command monkeyrunner can not be found')
+
+ if self.args.repo_type == 'git':
+ target_dir = mkdtemp(prefix='git_repo',
+ dir=self.config.tempdir_host)
+ os.chmod(target_dir, 0755)
+ GitRepository(self.args.url).checkout(target_dir)
+ else:
+ raise LavaCommandError("The repository type(%s) is not supported"
+ % self.args.repo_type)
+
+ script_list = utils.find_files(target_dir, '.py')
+
+ test_id = self.args.url
+ if len(test_id) > 40:
+ test_id = '%s...' % (test_id[:40])
+ test_id = 'monkeyrunner_%s' % test_id
+
+ tip_msg = ("Run monkeyrunner scripts in following url on device(%s):"
+ "\n\turl=%s") % (
+ self.serial,
+ self.args.url)
+
+ self.say_begin(tip_msg)
+ bundles = []
+ for script in script_list:
+ if "monkeycommon.py" == os.path.basename(script):
+ continue
+ sub_bundle = {}
+ from datetime import datetime
+ starttime = datetime.utcnow()
+ test_case_id = script.replace('%s/' % target_dir, '')
+ if len(test_case_id) > 50:
+ test_case_id = '%s...' % (test_case_id[:50])
+ try:
+ sub_bundle = self.run_monkeyrunner_test(script, self.serial,
+ test_case_id)
+ test_result = {"test_case_id": test_case_id,
+ "result": 'pass'}
+ if sub_bundle:
+ sub_bundle['test_runs'][0]['test_results'].append(
+ test_result)
+ except Exception as strerror:
+ self.say('Failed to run script(%s) with error:\n%s' % (
+ script,
+ strerror))
+
+ test_result = {"test_case_id": test_case_id,
+ "result": 'fail'}
+ TIMEFORMAT = '%Y-%m-%dT%H:%M:%SZ'
+ sub_bundle['test_runs'] = [{'test_results': [test_result],
+ 'test_id': 'monkeyrunner(%s)' % test_case_id,
+ 'time_check_performed': False,
+ 'analyzer_assigned_uuid': str(uuid4()),
+ 'analyzer_assigned_date': starttime.strftime(TIMEFORMAT)}]
+ if sub_bundle:
+ bundles.append(sub_bundle)
+
+ if self.args.output:
+ output_dir = os.path.dirname(self.args.output)
+ if output_dir and (not os.path.exists(output_dir)):
+ os.makedirs(output_dir)
+ with open(self.args.output, "wt") as stream:
+ DocumentIO.dump(stream, merge_bundles(bundles))
+
+ self.say_end(tip_msg)
+
+ def run_monkeyrunner_test(self, script, serial, test_case_id=None):
+
+ inst = AndroidTestInstaller()
+ run = AndroidTestRunner(steps_host_pre=[
+ 'monkeyrunner %s %s' % (script, serial)])
+ parser = MonkeyrunnerTestParser()
+ parser.monkeyrunner_result = os.path.join(os.path.dirname(script),
+ 'results.txt')
+ test = AndroidTest(testname='monkeyrunner',
+ installer=inst, runner=run, parser=parser)
+ test.parser.results = {'test_results': []}
+ test.setadb(self.adb)
+
+ ##By calling the install function, we will create the directory
+ ##on the target, and the the output file and error file
+ ##will be pushed there
+ if not self.test_installed(test.testname):
+ test.install()
+
+ ##The png files here are generated to the host by the monkeyrunner
+ ##monkeyrunner is run on host, not on the target
+ bundle = {}
+ org_png_file_list = utils.find_files(self.config.tempdir_host,
+ '.%s' % 'png')
+ result_id = test.run(quiet=self.args.quiet)
+ if self.args.output:
+ cur_all_png_list = utils.find_files(self.config.tempdir_host,
+ '.%s' % 'png')
+ new_png_list = set(cur_all_png_list).difference(org_png_file_list)
+ test_id = 'monkeyrunner(%s)' % (test_case_id)
+ bundle = generate_bundle(self.args.serial,
+ result_id, test=test,
+ test_id=test_id,
+ attachments=list(new_png_list))
+ utils.delete_files(new_png_list)
+
+ return bundle
+
+
+class MonkeyrunnerTestParser(AndroidTestParser):
+ '''
+ Before this method is called, self.monkeyrunner_result must be set
+ to the right value
+ '''
+
+
+ def real_parse(self, result_filename=None, output_filename=None,
+ test_name=''):
+ self.res_pattern = ("^\s*(?P<test_case_id>.*?)\s*="
+ "\s*(?P<result>(true|false))\s*$")
+ self.measurement_pattern = ("^\s*(?P<test_case_id>.*?)\s*="
+ "\s*(?P<measurement>[\.\d]+)\s*$")
+ self.measurement_units_pattern = ("^\s*(?P<test_case_id>.*?)\s*="
+ "\s*(?P<measurement>[\.\d]+)\s+(?P<units>\S+)\s*$")
+
+ res_pat = re.compile(self.res_pattern)
+ measurement_pat = re.compile(self.measurement_pattern)
+ measurement_units_pat = re.compile(self.measurement_units_pattern)
+
+
+ if not os.path.exists(self.monkeyrunner_result):
+ return
+ with open(self.monkeyrunner_result) as stream:
+ for lineno, line in enumerate(stream, 1):
+ match = res_pat.search(line)
+ if not match:
+ match = measurement_pat.search(line)
+ if not match:
+ match = measurement_units_pat.search(line)
+ if not match:
+ continue
+ data = match.groupdict()
+ data["log_filename"] = result_filename
+ data["log_lineno"] = lineno
+ if data.get('result') is None:
+ data['result'] = 'pass'
+
+ self.results['test_results'].append(data)
+
+
+class parse(AndroidResultsCommand):
+ """
+ Parse the results of previous test that run on the specified device
+ program:: lava-android-test parse test-result-id
+ """
+ def invoke_sub(self):
+ bundle = generate_combined_bundle(self.args.serial,
+ self.args.result_id)
+ try:
+ print DocumentIO.dumps(bundle)
+ except IOError:
+ pass
+
+
+class parse_custom(AndroidResultsCommand):
+ """
+ Parse the results of previous test that run with run-custom command
+ on the specified device
+ program:: lava-android-test parse-custom test-result-id -P
+ """
+ @classmethod
+ def register_arguments(cls, parser):
+ super(parse_custom, cls).register_arguments(parser)
+ parser.add_argument('-p', '--parse-regex',
+ help=("Specified the regular expression used"
+ " for analyzing command output"))
+
+ def invoke_sub(self):
+ PATTERN = None
+ if self.args.parse_regex:
+ PATTERN = self.args.parse_regex
+ test_name = 'custom'
+ inst = AndroidTestInstaller()
+ run = AndroidTestRunner()
+ parser = AndroidTestParser(pattern=PATTERN)
+ test = AndroidTest(testname=test_name, installer=inst,
+ runner=run, parser=parser)
+ test.parser.results = {'test_results': []}
+ test.setadb(self.adb)
+
+ bundle = generate_combined_bundle(self.args.serial,
+ self.args.result_id, test=test)
+ try:
+ print DocumentIO.dumps(bundle)
+ except IOError:
+ pass
+
+
+def generate_combined_bundle(serial=None, result_ids=None, test=None,
+ test_id=None):
+ if result_ids is None:
+ return {}
+
+ bundle = None
+
+ for rid in result_ids:
+ b = generate_bundle(serial, rid, test, test_id)
+ if rid == result_ids[0]:
+ bundle = b
+ else:
+ bundle['test_runs'].append(b['test_runs'][0])
+
+ return bundle
+
+
+def merge_bundles(bundles=[]):
+ config = get_config()
+ merged_bundles = {"format": config.bundle_format,
+ 'test_runs': []}
+ for bundle in bundles:
+ if bundle['test_runs']:
+ merged_bundles['test_runs'].append(bundle['test_runs'][0])
+ return merged_bundles
+
+
+def generate_bundle(serial=None, result_id=None, test=None,
+ test_id=None, attachments=[]):
+ if result_id is None:
+ return {}
+ config = get_config()
+ adb = ADB(serial)
+ resultdir = os.path.join(config.resultsdir_android, result_id)
+ if not adb.exists(resultdir):
+ raise Exception("The result (%s) is not existed." % result_id)
+
+ bundle_text = adb.read_file(os.path.join(resultdir, "testdata.json"))
+ bundle = DocumentIO.loads(bundle_text)[1]
+ test_tmp = None
+ if test:
+ test_tmp = test
+ else:
+ test_tmp = TestProvider().load_test(bundle['test_runs'][0]['test_id'],
+ serial)
+ if test_id:
+ bundle['test_runs'][0]['test_id'] = test_id
+ else:
+ attrs = bundle['test_runs'][0].get('attributes')
+ if attrs:
+ run_options = attrs.get('run_options')
+ if run_options:
+ test_id = '%s(%s)' % (bundle['test_runs'][0]['test_id'],
+ run_options)
+ bundle['test_runs'][0]['test_id'] = test_id
+
+ test_tmp.parse(result_id)
+ stdout_text = adb.read_file(os.path.join(resultdir,
+ os.path.basename(test_tmp.org_ouput_file)))
+ if stdout_text is None:
+ stdout_text = ''
+ stderr_text = adb.read_file(os.path.join(resultdir, 'stderr.log'))
+ if stderr_text is None:
+ stderr_text = ''
+ bundle['test_runs'][0]["test_results"] = test_tmp.parser.results[
+ "test_results"]
+
+ ## following part is used for generating the attachment for normal test
+ attachment_bundles = []
+ for attachment in test_tmp.attachments:
+ data_bundle = attachment.generate_bundle(adb=adb, resultsdir=resultdir)
+ if data_bundle:
+ attachment_bundles.append(data_bundle)
+
+ bundle['test_runs'][0]["attachments"] = attachment_bundles
+
+ ##following used for the attachment for monkeyrunner test
+ for attach in attachments:
+ if os.path.exists(attach):
+ with open(attach, 'rb') as stream:
+ data = stream.read()
+ if data:
+ bundle['test_runs'][0]["attachments"].append({
+ "pathname": os.path.basename(attach),
+ "mime_type": 'image/png',
+ "content": base64.standard_b64encode(data)})
+ return bundle
+
+class show(AndroidResultCommand):
+ """
+ Display the output from a previous test that run on the specified device
+ program:: lava-android-test show result-id
+ program:: lava-android-test show result-id -s device_serial
+ """
+ def invoke_sub(self):
+ resultsdir = os.path.join(self.config.resultsdir_android,
+ self.args.result_id)
+ if not self.adb.exists(resultsdir):
+ raise LavaCommandError(
+ "The result (%s) is not existed." % self.args.result_id)
+
+ stdout = os.path.join(resultsdir, "stdout.log")
+ if not self.adb.exists(stdout):
+ self.say("No result found for '%s'" % self.args.result_id)
+ return
+ try:
+ output = self.adb.get_shellcmdoutput('cat %s' % stdout)[1]
+ if output is not None:
+ for line in output:
+ self.display_subprocess_output('stdout', line)
+ except IOError:
+ pass
+
+ stderr = os.path.join(resultsdir, "stderr.log")
+ if not self.adb.exists(stderr):
+ return
+ try:
+ output = self.adb.get_shellcmdoutput('cat %s' % stderr)[1]
+ if output is not None:
+ for line in output:
+ self.display_subprocess_output('stderr', line)
+ except IOError:
+ pass
+
+
+class rename(AndroidResultCommand):
+ """
+ Rename the result's id of a previous test that run on the specified device
+ program:: lava-android-test rename result-id result-id-new
+ program:: lava-android-test remove result-id result-id-new -s device_serial
+ """
+
+ @classmethod
+ def register_arguments(self, parser):
+ super(rename, self).register_arguments(parser)
+ parser.add_argument("result_id_new",
+ help="New test result identifier")
+
+ def invoke_sub(self):
+ srcdir = os.path.join(self.config.resultsdir_android,
+ self.args.result_id)
+ destdir = os.path.join(self.config.resultsdir_android,
+ self.args.result_id_new)
+
+ if not self.adb.exists(srcdir):
+ self.say("Result (%s) not found" % self.args.result_id)
+ return
+ if self.adb.exists(destdir):
+ self.say("Destination result name already exists")
+ self.adb.move(srcdir, destdir)
+
+
+class remove(AndroidResultsCommand):
+ """
+ Remove the result of a previous test that run on the specified device
+ program:: lava-android-test remove result-id
+ program:: lava-android-test remove result-id0 result-id1
+ program:: lava-android-test remove result-id -s device_serial
+ """
+
+ @classmethod
+ def register_arguments(self, parser):
+ super(remove, self).register_arguments(parser)
+ group = parser.add_argument_group("force to remove")
+ group.add_argument("-f", "--force",
+ action="store_true",
+ help=("give an interactive question about remove"))
+
+ def remove(self, rid):
+ resultsdir = os.path.join(self.config.resultsdir_android, rid)
+ if not self.adb.exists(resultsdir):
+ self.say("No result found for '%s'" % rid)
+ return
+ if not self.args.force:
+ self.say("Remove result '%s' for good? [Y/N]" % rid)
+ response = raw_input()
+ if response[0].upper() != 'Y':
+ return
+ self.adb.rmtree(resultsdir)
+
+ def invoke_sub(self):
+ for rid in self.args.result_id:
+ self.remove(rid)