You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
558 lines
20 KiB
558 lines
20 KiB
#!/usr/bin/python2.4
|
|
#
|
|
#
|
|
# Copyright 2008, 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.
|
|
|
|
"""Provides an interface to communicate with the device via the adb command.
|
|
|
|
Assumes adb binary is currently on system path.
|
|
"""
|
|
# Python imports
|
|
import os
|
|
import string
|
|
import time
|
|
|
|
# local imports
|
|
import am_instrument_parser
|
|
import errors
|
|
import logger
|
|
import run_command
|
|
|
|
|
|
class AdbInterface:
|
|
"""Helper class for communicating with Android device via adb."""
|
|
|
|
# argument to pass to adb, to direct command to specific device
|
|
_target_arg = ""
|
|
|
|
DEVICE_TRACE_DIR = "/data/test_results/"
|
|
|
|
def SetEmulatorTarget(self):
|
|
"""Direct all future commands to the only running emulator."""
|
|
self._target_arg = "-e"
|
|
|
|
def SetDeviceTarget(self):
|
|
"""Direct all future commands to the only connected USB device."""
|
|
self._target_arg = "-d"
|
|
|
|
def SetTargetSerial(self, serial):
|
|
"""Direct all future commands to Android target with the given serial."""
|
|
self._target_arg = "-s %s" % serial
|
|
|
|
def SendCommand(self, command_string, timeout_time=60, retry_count=3):
|
|
"""Send a command via adb.
|
|
|
|
Args:
|
|
command_string: adb command to run
|
|
timeout_time: number of seconds to wait for command to respond before
|
|
retrying
|
|
retry_count: number of times to retry command before raising
|
|
WaitForResponseTimedOutError
|
|
Returns:
|
|
string output of command
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError if device does not respond to command within time
|
|
"""
|
|
adb_cmd = "adb %s %s" % (self._target_arg, command_string)
|
|
logger.SilentLog("about to run %s" % adb_cmd)
|
|
return run_command.RunCommand(adb_cmd, timeout_time=timeout_time,
|
|
retry_count=retry_count)
|
|
|
|
def SendShellCommand(self, cmd, timeout_time=20, retry_count=3):
|
|
"""Send a adb shell command.
|
|
|
|
Args:
|
|
cmd: adb shell command to run
|
|
timeout_time: number of seconds to wait for command to respond before
|
|
retrying
|
|
retry_count: number of times to retry command before raising
|
|
WaitForResponseTimedOutError
|
|
|
|
Returns:
|
|
string output of command
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError: if device does not respond to command
|
|
"""
|
|
return self.SendCommand("shell %s" % cmd, timeout_time=timeout_time,
|
|
retry_count=retry_count)
|
|
|
|
def BugReport(self, path):
|
|
"""Dumps adb bugreport to the file specified by the path.
|
|
|
|
Args:
|
|
path: Path of the file where adb bugreport is dumped to.
|
|
"""
|
|
bug_output = self.SendShellCommand("bugreport", timeout_time=60)
|
|
bugreport_file = open(path, "w")
|
|
bugreport_file.write(bug_output)
|
|
bugreport_file.close()
|
|
|
|
def Push(self, src, dest):
|
|
"""Pushes the file src onto the device at dest.
|
|
|
|
Args:
|
|
src: file path of host file to push
|
|
dest: destination absolute file path on device
|
|
"""
|
|
self.SendCommand("push %s %s" % (src, dest), timeout_time=60)
|
|
|
|
def Pull(self, src, dest):
|
|
"""Pulls the file src on the device onto dest on the host.
|
|
|
|
Args:
|
|
src: absolute file path of file on device to pull
|
|
dest: destination file path on host
|
|
|
|
Returns:
|
|
True if success and False otherwise.
|
|
"""
|
|
# Create the base dir if it doesn't exist already
|
|
if not os.path.exists(os.path.dirname(dest)):
|
|
os.makedirs(os.path.dirname(dest))
|
|
|
|
if self.DoesFileExist(src):
|
|
self.SendCommand("pull %s %s" % (src, dest), timeout_time=60)
|
|
return True
|
|
else:
|
|
logger.Log("ADB Pull Failed: Source file %s does not exist." % src)
|
|
return False
|
|
|
|
def Install(self, apk_path, extra_flags):
|
|
"""Installs apk on device.
|
|
|
|
Args:
|
|
apk_path: file path to apk file on host
|
|
extra_flags: Additional flags to use with adb install
|
|
|
|
Returns:
|
|
output of install command
|
|
"""
|
|
return self.SendCommand("install -r %s %s" % (extra_flags, apk_path))
|
|
|
|
def DoesFileExist(self, src):
|
|
"""Checks if the given path exists on device target.
|
|
|
|
Args:
|
|
src: file path to be checked.
|
|
|
|
Returns:
|
|
True if file exists
|
|
"""
|
|
|
|
output = self.SendShellCommand("ls %s" % src)
|
|
error = "No such file or directory"
|
|
|
|
if error in output:
|
|
return False
|
|
return True
|
|
|
|
def EnableAdbRoot(self):
|
|
"""Enable adb root on device."""
|
|
output = self.SendCommand("root")
|
|
if "adbd is already running as root" in output:
|
|
return True
|
|
elif "restarting adbd as root" in output:
|
|
# device will disappear from adb, wait for it to come back
|
|
time.sleep(2)
|
|
self.SendCommand("wait-for-device")
|
|
return True
|
|
else:
|
|
logger.Log("Unrecognized output from adb root: %s" % output)
|
|
return False
|
|
|
|
def StartInstrumentationForPackage(
|
|
self, package_name, runner_name, timeout_time=60*10,
|
|
no_window_animation=False, instrumentation_args={}, user=None,
|
|
no_hidden_api_checks=False):
|
|
"""Run instrumentation test for given package and runner.
|
|
|
|
Equivalent to StartInstrumentation, except instrumentation path is
|
|
separated into its package and runner components.
|
|
"""
|
|
instrumentation_path = "%s/%s" % (package_name, runner_name)
|
|
return self.StartInstrumentation(instrumentation_path, timeout_time=timeout_time,
|
|
no_window_animation=no_window_animation,
|
|
instrumentation_args=instrumentation_args,
|
|
user=user,
|
|
no_hidden_api_checks=no_hidden_api_checks)
|
|
|
|
def StartInstrumentation(
|
|
self, instrumentation_path, timeout_time=60*10, no_window_animation=False,
|
|
profile=False, instrumentation_args={}, user=None,
|
|
no_hidden_api_checks=False):
|
|
|
|
"""Runs an instrumentation class on the target.
|
|
|
|
Returns a dictionary containing the key value pairs from the
|
|
instrumentations result bundle and a list of TestResults. Also handles the
|
|
interpreting of error output from the device and raises the necessary
|
|
exceptions.
|
|
|
|
Args:
|
|
instrumentation_path: string. It should be the fully classified package
|
|
name, and instrumentation test runner, separated by "/"
|
|
e.g. com.android.globaltimelaunch/.GlobalTimeLaunch
|
|
timeout_time: Timeout value for the am command.
|
|
no_window_animation: boolean, Whether you want window animations enabled
|
|
or disabled
|
|
profile: If True, profiling will be turned on for the instrumentation.
|
|
instrumentation_args: Dictionary of key value bundle arguments to pass to
|
|
instrumentation.
|
|
user: The user id to start the instrumentation with.
|
|
|
|
Returns:
|
|
(test_results, inst_finished_bundle)
|
|
|
|
test_results: a list of TestResults
|
|
inst_finished_bundle (dict): Key/value pairs contained in the bundle that
|
|
is passed into ActivityManager.finishInstrumentation(). Included in this
|
|
bundle is the return code of the Instrumentation process, any error
|
|
codes reported by the activity manager, and any results explicitly added
|
|
by the instrumentation code.
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError: if timeout occurred while waiting for
|
|
response to adb instrument command
|
|
DeviceUnresponsiveError: if device system process is not responding
|
|
InstrumentationError: if instrumentation failed to run
|
|
"""
|
|
|
|
command_string = self._BuildInstrumentationCommandPath(
|
|
instrumentation_path, no_window_animation=no_window_animation,
|
|
profile=profile, raw_mode=True,
|
|
instrumentation_args=instrumentation_args,
|
|
user=user, no_hidden_api_checks=no_hidden_api_checks)
|
|
logger.Log(command_string)
|
|
(test_results, inst_finished_bundle) = (
|
|
am_instrument_parser.ParseAmInstrumentOutput(
|
|
self.SendShellCommand(command_string, timeout_time=timeout_time,
|
|
retry_count=2)))
|
|
|
|
if "code" not in inst_finished_bundle:
|
|
raise errors.InstrumentationError("no test results... device setup "
|
|
"correctly?")
|
|
|
|
if inst_finished_bundle["code"] == "0":
|
|
short_msg_result = "no error message"
|
|
if "shortMsg" in inst_finished_bundle:
|
|
short_msg_result = inst_finished_bundle["shortMsg"]
|
|
logger.Log("Error! Test run failed: %s" % short_msg_result)
|
|
raise errors.InstrumentationError(short_msg_result)
|
|
|
|
if "INSTRUMENTATION_ABORTED" in inst_finished_bundle:
|
|
logger.Log("INSTRUMENTATION ABORTED!")
|
|
raise errors.DeviceUnresponsiveError
|
|
|
|
return (test_results, inst_finished_bundle)
|
|
|
|
def StartInstrumentationNoResults(
|
|
self, package_name, runner_name, no_window_animation=False,
|
|
raw_mode=False, instrumentation_args={}, user=None,
|
|
no_hidden_api_checks=False):
|
|
"""Runs instrumentation and dumps output to stdout.
|
|
|
|
Equivalent to StartInstrumentation, but will dump instrumentation
|
|
'normal' output to stdout, instead of parsing return results. Command will
|
|
never timeout.
|
|
"""
|
|
adb_command_string = self.PreviewInstrumentationCommand(
|
|
package_name, runner_name, no_window_animation=no_window_animation,
|
|
raw_mode=raw_mode, instrumentation_args=instrumentation_args,
|
|
user=user, no_hidden_api_checks=no_hidden_api_checks)
|
|
logger.Log(adb_command_string)
|
|
run_command.RunCommand(adb_command_string, return_output=False)
|
|
|
|
def PreviewInstrumentationCommand(
|
|
self, package_name, runner_name, no_window_animation=False,
|
|
raw_mode=False, instrumentation_args={}, user=None,
|
|
no_hidden_api_checks=False):
|
|
"""Returns a string of adb command that will be executed."""
|
|
inst_command_string = self._BuildInstrumentationCommand(
|
|
package_name, runner_name, no_window_animation=no_window_animation,
|
|
raw_mode=raw_mode, instrumentation_args=instrumentation_args,
|
|
user=user, no_hidden_api_checks=no_hidden_api_checks)
|
|
return self.PreviewShellCommand(inst_command_string)
|
|
|
|
def PreviewShellCommand(self, cmd):
|
|
return "adb %s shell %s" % (self._target_arg, cmd)
|
|
|
|
def _BuildInstrumentationCommand(
|
|
self, package, runner_name, no_window_animation=False, profile=False,
|
|
raw_mode=True, instrumentation_args={}, user=None,
|
|
no_hidden_api_checks=False):
|
|
instrumentation_path = "%s/%s" % (package, runner_name)
|
|
|
|
return self._BuildInstrumentationCommandPath(
|
|
instrumentation_path, no_window_animation=no_window_animation,
|
|
profile=profile, raw_mode=raw_mode,
|
|
instrumentation_args=instrumentation_args, user=user,
|
|
no_hidden_api_checks=no_hidden_api_checks)
|
|
|
|
def _BuildInstrumentationCommandPath(
|
|
self, instrumentation_path, no_window_animation=False, profile=False,
|
|
raw_mode=True, instrumentation_args={}, user=None,
|
|
no_hidden_api_checks=False):
|
|
command_string = "am instrument"
|
|
if no_hidden_api_checks:
|
|
command_string += " --no-hidden-api-checks"
|
|
if user:
|
|
command_string += " --user %s" % user
|
|
if no_window_animation:
|
|
command_string += " --no_window_animation"
|
|
if profile:
|
|
self._CreateTraceDir()
|
|
command_string += (
|
|
" -p %s/%s.dmtrace" %
|
|
(self.DEVICE_TRACE_DIR, instrumentation_path.split(".")[-1]))
|
|
|
|
for key, value in instrumentation_args.items():
|
|
command_string += " -e %s '%s'" % (key, value)
|
|
if raw_mode:
|
|
command_string += " -r"
|
|
command_string += " -w '%s'" % instrumentation_path
|
|
return command_string
|
|
|
|
def _CreateTraceDir(self):
|
|
ls_response = self.SendShellCommand("ls /data/trace")
|
|
if ls_response.strip("#").strip(string.whitespace) != "":
|
|
self.SendShellCommand("create /data/trace", "mkdir /data/trace")
|
|
self.SendShellCommand("make /data/trace world writeable",
|
|
"chmod 777 /data/trace")
|
|
|
|
def WaitForDevicePm(self, wait_time=120):
|
|
"""Waits for targeted device's package manager to be up.
|
|
|
|
Args:
|
|
wait_time: time in seconds to wait
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError if wait_time elapses and pm still does not
|
|
respond.
|
|
"""
|
|
logger.Log("Waiting for device package manager...")
|
|
self.SendCommand("wait-for-device")
|
|
# Now the device is there, but may not be running.
|
|
# Query the package manager with a basic command
|
|
try:
|
|
self._WaitForShellCommandContents("pm path android", "package:",
|
|
wait_time)
|
|
except errors.WaitForResponseTimedOutError:
|
|
raise errors.WaitForResponseTimedOutError(
|
|
"Package manager did not respond after %s seconds" % wait_time)
|
|
|
|
def IsInstrumentationInstalled(self, package_name, runner_name):
|
|
"""Checks if instrumentation is present on device."""
|
|
instrumentation_path = "%s/%s" % (package_name, runner_name)
|
|
command = "pm list instrumentation | grep %s" % instrumentation_path
|
|
try:
|
|
output = self.SendShellCommand(command)
|
|
return output.startswith("instrumentation:")
|
|
except errors.AbortError:
|
|
# command can return error code on failure
|
|
return False
|
|
|
|
def WaitForProcess(self, name, wait_time=120):
|
|
"""Wait until a process is running on the device.
|
|
|
|
Args:
|
|
name: the process name as it appears in `ps`
|
|
wait_time: time in seconds to wait
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError if wait_time elapses and the process is
|
|
still not running
|
|
"""
|
|
logger.Log("Waiting for process %s" % name)
|
|
self.SendCommand("wait-for-device")
|
|
self._WaitForShellCommandContents("ps", name, wait_time)
|
|
|
|
def WaitForProcessEnd(self, name, wait_time=120):
|
|
"""Wait until a process is no longer running on the device.
|
|
|
|
Args:
|
|
name: the process name as it appears in `ps`
|
|
wait_time: time in seconds to wait
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError if wait_time elapses and the process is
|
|
still running
|
|
"""
|
|
logger.Log("Waiting for process %s to end" % name)
|
|
self._WaitForShellCommandContents("ps", name, wait_time, invert=True)
|
|
|
|
def _WaitForShellCommandContents(self, command, expected, wait_time,
|
|
raise_abort=True, invert=False):
|
|
"""Wait until the response to a command contains a given output.
|
|
|
|
Assumes that a only successful execution of "adb shell <command>" contains
|
|
the substring expected. Assumes that a device is present.
|
|
|
|
Args:
|
|
command: adb shell command to execute
|
|
expected: the string that should appear to consider the
|
|
command successful.
|
|
wait_time: time in seconds to wait
|
|
raise_abort: if False, retry when executing the command raises an
|
|
AbortError, rather than failing.
|
|
invert: if True, wait until the command output no longer contains the
|
|
expected contents.
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError: If wait_time elapses and the command has not
|
|
returned an output containing expected yet.
|
|
"""
|
|
# Query the device with the command
|
|
success = False
|
|
attempts = 0
|
|
wait_period = 5
|
|
while not success and (attempts*wait_period) < wait_time:
|
|
# assume the command will always contain expected in the success case
|
|
try:
|
|
output = self.SendShellCommand(command, retry_count=1)
|
|
if ((not invert and expected in output)
|
|
or (invert and expected not in output)):
|
|
success = True
|
|
except errors.AbortError, e:
|
|
if raise_abort:
|
|
raise
|
|
# ignore otherwise
|
|
|
|
if not success:
|
|
time.sleep(wait_period)
|
|
attempts += 1
|
|
|
|
if not success:
|
|
raise errors.WaitForResponseTimedOutError()
|
|
|
|
def WaitForBootComplete(self, wait_time=120):
|
|
"""Waits for targeted device's bootcomplete flag to be set.
|
|
|
|
Args:
|
|
wait_time: time in seconds to wait
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError if wait_time elapses and pm still does not
|
|
respond.
|
|
"""
|
|
logger.Log("Waiting for boot complete...")
|
|
self.SendCommand("wait-for-device")
|
|
# Now the device is there, but may not be running.
|
|
# Query the package manager with a basic command
|
|
boot_complete = False
|
|
attempts = 0
|
|
wait_period = 5
|
|
while not boot_complete and (attempts*wait_period) < wait_time:
|
|
output = self.SendShellCommand("getprop dev.bootcomplete", retry_count=1)
|
|
output = output.strip()
|
|
if output == "1":
|
|
boot_complete = True
|
|
else:
|
|
time.sleep(wait_period)
|
|
attempts += 1
|
|
if not boot_complete:
|
|
raise errors.WaitForResponseTimedOutError(
|
|
"dev.bootcomplete flag was not set after %s seconds" % wait_time)
|
|
|
|
def Sync(self, retry_count=3, runtime_restart=False):
|
|
"""Perform a adb sync.
|
|
|
|
Blocks until device package manager is responding.
|
|
|
|
Args:
|
|
retry_count: number of times to retry sync before failing
|
|
runtime_restart: stop runtime during sync and restart afterwards, useful
|
|
for syncing system libraries (core, framework etc)
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError if package manager does not respond
|
|
AbortError if unrecoverable error occurred
|
|
"""
|
|
output = ""
|
|
error = None
|
|
if runtime_restart:
|
|
self.SendShellCommand("setprop ro.test_harness 1", retry_count=retry_count)
|
|
# manual rest bootcomplete flag
|
|
self.SendShellCommand("setprop dev.bootcomplete 0",
|
|
retry_count=retry_count)
|
|
self.SendShellCommand("stop", retry_count=retry_count)
|
|
|
|
try:
|
|
output = self.SendCommand("sync", retry_count=retry_count)
|
|
except errors.AbortError, e:
|
|
error = e
|
|
output = e.msg
|
|
if "Read-only file system" in output:
|
|
logger.SilentLog(output)
|
|
logger.Log("Remounting read-only filesystem")
|
|
self.SendCommand("remount")
|
|
output = self.SendCommand("sync", retry_count=retry_count)
|
|
elif "No space left on device" in output:
|
|
logger.SilentLog(output)
|
|
logger.Log("Restarting device runtime")
|
|
self.SendShellCommand("stop", retry_count=retry_count)
|
|
output = self.SendCommand("sync", retry_count=retry_count)
|
|
self.SendShellCommand("start", retry_count=retry_count)
|
|
elif error is not None:
|
|
# exception occurred that cannot be recovered from
|
|
raise error
|
|
logger.SilentLog(output)
|
|
if runtime_restart:
|
|
# start runtime and wait till boot complete flag is set
|
|
self.SendShellCommand("start", retry_count=retry_count)
|
|
self.WaitForBootComplete()
|
|
# press the MENU key, this will disable key guard if runtime is started
|
|
# with ro.monkey set to 1
|
|
self.SendShellCommand("input keyevent 82", retry_count=retry_count)
|
|
else:
|
|
self.WaitForDevicePm()
|
|
return output
|
|
|
|
def GetSerialNumber(self):
|
|
"""Returns the serial number of the targeted device."""
|
|
return self.SendCommand("get-serialno").strip()
|
|
|
|
def RuntimeReset(self, disable_keyguard=False, retry_count=3, preview_only=False):
|
|
"""
|
|
Resets the Android runtime (does *not* reboot the kernel).
|
|
|
|
Blocks until the reset is complete and the package manager
|
|
is available.
|
|
|
|
Args:
|
|
disable_keyguard: if True, presses the MENU key to disable
|
|
key guard, after reset is finished
|
|
retry_count: number of times to retry reset before failing
|
|
|
|
Raises:
|
|
WaitForResponseTimedOutError if package manager does not respond
|
|
AbortError if unrecoverable error occurred
|
|
"""
|
|
|
|
logger.Log("adb shell stop")
|
|
logger.Log("adb shell start")
|
|
|
|
if not preview_only:
|
|
self.SendShellCommand("stop", retry_count=retry_count)
|
|
self.SendShellCommand("start", retry_count=retry_count)
|
|
|
|
self.WaitForDevicePm()
|
|
|
|
if disable_keyguard:
|
|
logger.Log("input keyevent 82 ## disable keyguard")
|
|
if not preview_only:
|
|
self.SendShellCommand("input keyevent 82", retry_count=retry_count)
|