blob: 59e24a92251ab7cf541596228ad5ef66302e3e58 [file] [log] [blame]
#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# 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 collections
import logging
import math
import os
import re
import shutil
import socket
import time
from datetime import datetime
from antlion import context
from antlion import logger as acts_logger
from antlion import tracelogger, utils
from antlion.controllers import adb, fastboot
from antlion.controllers.adb_lib.error import AdbError
from antlion.controllers.android_lib import errors
from antlion.controllers.android_lib import events as android_events
from antlion.controllers.android_lib import logcat, services
from antlion.controllers.sl4a_lib import sl4a_manager
from antlion.controllers.utils_lib.ssh import connection, settings
from antlion.event import event_bus
from antlion.libs.proc import job
MOBLY_CONTROLLER_CONFIG_NAME = "AndroidDevice"
ACTS_CONTROLLER_REFERENCE_NAME = "android_devices"
ANDROID_DEVICE_PICK_ALL_TOKEN = "*"
# Key name for SL4A extra params in config file
ANDROID_DEVICE_SL4A_CLIENT_PORT_KEY = "sl4a_client_port"
ANDROID_DEVICE_SL4A_FORWARDED_PORT_KEY = "sl4a_forwarded_port"
ANDROID_DEVICE_SL4A_SERVER_PORT_KEY = "sl4a_server_port"
# Key name for adb logcat extra params in config file.
ANDROID_DEVICE_ADB_LOGCAT_PARAM_KEY = "adb_logcat_param"
ANDROID_DEVICE_EMPTY_CONFIG_MSG = "Configuration is empty, abort!"
ANDROID_DEVICE_NOT_LIST_CONFIG_MSG = "Configuration should be a list, abort!"
CRASH_REPORT_PATHS = (
"/data/tombstones/",
"/data/vendor/ramdump/",
"/data/ramdump/",
"/data/vendor/ssrdump",
"/data/vendor/ramdump/bluetooth",
"/data/vendor/log/cbd",
)
CRASH_REPORT_SKIPS = (
"RAMDUMP_RESERVED",
"RAMDUMP_STATUS",
"RAMDUMP_OUTPUT",
"bluetooth",
)
ALWAYS_ON_LOG_PATH = "/data/vendor/radio/logs/always-on"
DEFAULT_QXDM_LOG_PATH = "/data/vendor/radio/diag_logs"
DEFAULT_SDM_LOG_PATH = "/data/vendor/slog/"
DEFAULT_SCREENSHOT_PATH = "/sdcard/Pictures/screencap"
BUG_REPORT_TIMEOUT = 1800
PULL_TIMEOUT = 300
PORT_RETRY_COUNT = 3
ADB_ROOT_RETRY_COUNT = 2
ADB_ROOT_RETRY_INTERVAL = 10
IPERF_TIMEOUT = 60
SL4A_APK_NAME = "com.googlecode.android_scripting"
WAIT_FOR_DEVICE_TIMEOUT = 180
ENCRYPTION_WINDOW = "CryptKeeper"
DEFAULT_DEVICE_PASSWORD = "1111"
RELEASE_ID_REGEXES = [re.compile(r"\w+\.\d+\.\d+"), re.compile(r"N\w+")]
def create(configs):
"""Creates AndroidDevice controller objects.
Args:
configs: A list of dicts, each representing a configuration for an
Android device.
Returns:
A list of AndroidDevice objects.
"""
if not configs:
raise errors.AndroidDeviceConfigError(ANDROID_DEVICE_EMPTY_CONFIG_MSG)
elif configs == ANDROID_DEVICE_PICK_ALL_TOKEN:
ads = get_all_instances()
elif not isinstance(configs, list):
raise errors.AndroidDeviceConfigError(ANDROID_DEVICE_NOT_LIST_CONFIG_MSG)
elif isinstance(configs[0], str):
# Configs is a list of serials.
ads = get_instances(configs)
else:
# Configs is a list of dicts.
ads = get_instances_with_configs(configs)
ads[0].log.info(f'The primary device under test is "{ads[0].serial}".')
for ad in ads:
if not ad.is_connected():
raise errors.AndroidDeviceError(
("Android device %s is specified in config" " but is not attached.")
% ad.serial,
serial=ad.serial,
)
_start_services_on_ads(ads)
for ad in ads:
if ad.droid:
utils.set_location_service(ad, False)
utils.sync_device_time(ad)
return ads
def destroy(ads):
"""Cleans up AndroidDevice objects.
Args:
ads: A list of AndroidDevice objects.
"""
for ad in ads:
try:
ad.clean_up()
except:
ad.log.exception("Failed to clean up properly.")
def get_info(ads):
"""Get information on a list of AndroidDevice objects.
Args:
ads: A list of AndroidDevice objects.
Returns:
A list of dict, each representing info for an AndroidDevice objects.
"""
device_info = []
for ad in ads:
info = {"serial": ad.serial, "model": ad.model}
info.update(ad.build_info)
device_info.append(info)
return device_info
def _start_services_on_ads(ads):
"""Starts long running services on multiple AndroidDevice objects.
If any one AndroidDevice object fails to start services, cleans up all
existing AndroidDevice objects and their services.
Args:
ads: A list of AndroidDevice objects whose services to start.
"""
running_ads = []
for ad in ads:
running_ads.append(ad)
try:
ad.start_services()
except:
ad.log.exception("Failed to start some services, abort!")
destroy(running_ads)
raise
def _parse_device_list(device_list_str, key):
"""Parses a byte string representing a list of devices. The string is
generated by calling either adb or fastboot.
Args:
device_list_str: Output of adb or fastboot.
key: The token that signifies a device in device_list_str.
Returns:
A list of android device serial numbers.
"""
return re.findall(r"(\S+)\t%s" % key, device_list_str)
def list_adb_devices():
"""List all android devices connected to the computer that are detected by
adb.
Returns:
A list of android device serials. Empty if there's none.
"""
out = adb.AdbProxy().devices()
return _parse_device_list(out, "device")
def list_fastboot_devices():
"""List all android devices connected to the computer that are in in
fastboot mode. These are detected by fastboot.
Returns:
A list of android device serials. Empty if there's none.
"""
out = fastboot.FastbootProxy().devices()
return _parse_device_list(out, "fastboot")
def get_instances(serials):
"""Create AndroidDevice instances from a list of serials.
Args:
serials: A list of android device serials.
Returns:
A list of AndroidDevice objects.
"""
results = []
for s in serials:
results.append(AndroidDevice(s))
return results
def get_instances_with_configs(configs):
"""Create AndroidDevice instances from a list of json configs.
Each config should have the required key-value pair "serial".
Args:
configs: A list of dicts each representing the configuration of one
android device.
Returns:
A list of AndroidDevice objects.
"""
results = []
for c in configs:
try:
serial = c.pop("serial")
except KeyError:
raise errors.AndroidDeviceConfigError(
f"Required value 'serial' is missing in AndroidDevice config {c}."
)
client_port = 0
if ANDROID_DEVICE_SL4A_CLIENT_PORT_KEY in c:
try:
client_port = int(c.pop(ANDROID_DEVICE_SL4A_CLIENT_PORT_KEY))
except ValueError:
raise errors.AndroidDeviceConfigError(
"'%s' is not a valid number for config %s"
% (ANDROID_DEVICE_SL4A_CLIENT_PORT_KEY, c)
)
server_port = None
if ANDROID_DEVICE_SL4A_SERVER_PORT_KEY in c:
try:
server_port = int(c.pop(ANDROID_DEVICE_SL4A_SERVER_PORT_KEY))
except ValueError:
raise errors.AndroidDeviceConfigError(
"'%s' is not a valid number for config %s"
% (ANDROID_DEVICE_SL4A_SERVER_PORT_KEY, c)
)
forwarded_port = 0
if ANDROID_DEVICE_SL4A_FORWARDED_PORT_KEY in c:
try:
forwarded_port = int(c.pop(ANDROID_DEVICE_SL4A_FORWARDED_PORT_KEY))
except ValueError:
raise errors.AndroidDeviceConfigError(
"'%s' is not a valid number for config %s"
% (ANDROID_DEVICE_SL4A_FORWARDED_PORT_KEY, c)
)
ssh_config = c.pop("ssh_config", None)
ssh_connection = None
if ssh_config is not None:
ssh_settings = settings.from_config(ssh_config)
ssh_connection = connection.SshConnection(ssh_settings)
ad = AndroidDevice(
serial,
ssh_connection=ssh_connection,
client_port=client_port,
forwarded_port=forwarded_port,
server_port=server_port,
)
ad.load_config(c)
results.append(ad)
return results
def get_all_instances(include_fastboot=False):
"""Create AndroidDevice instances for all attached android devices.
Args:
include_fastboot: Whether to include devices in bootloader mode or not.
Returns:
A list of AndroidDevice objects each representing an android device
attached to the computer.
"""
if include_fastboot:
serial_list = list_adb_devices() + list_fastboot_devices()
return get_instances(serial_list)
return get_instances(list_adb_devices())
def filter_devices(ads, func):
"""Finds the AndroidDevice instances from a list that match certain
conditions.
Args:
ads: A list of AndroidDevice instances.
func: A function that takes an AndroidDevice object and returns True
if the device satisfies the filter condition.
Returns:
A list of AndroidDevice instances that satisfy the filter condition.
"""
results = []
for ad in ads:
if func(ad):
results.append(ad)
return results
def get_device(ads, **kwargs):
"""Finds a unique AndroidDevice instance from a list that has specific
attributes of certain values.
Example:
get_device(android_devices, label="foo", phone_number="1234567890")
get_device(android_devices, model="angler")
Args:
ads: A list of AndroidDevice instances.
kwargs: keyword arguments used to filter AndroidDevice instances.
Returns:
The target AndroidDevice instance.
Raises:
AndroidDeviceError is raised if none or more than one device is
matched.
"""
def _get_device_filter(ad):
for k, v in kwargs.items():
if not hasattr(ad, k):
return False
elif getattr(ad, k) != v:
return False
return True
filtered = filter_devices(ads, _get_device_filter)
if not filtered:
raise ValueError(
f"Could not find a target device that matches condition: {kwargs}."
)
elif len(filtered) == 1:
return filtered[0]
else:
serials = [ad.serial for ad in filtered]
raise ValueError(f"More than one device matched: {serials}")
def take_bug_reports(ads, test_name, begin_time):
"""Takes bug reports on a list of android devices.
If you want to take a bug report, call this function with a list of
android_device objects in on_fail. But reports will be taken on all the
devices in the list concurrently. Bug report takes a relative long
time to take, so use this cautiously.
Args:
ads: A list of AndroidDevice instances.
test_name: Name of the test case that triggered this bug report.
begin_time: Logline format timestamp taken when the test started.
"""
def take_br(test_name, begin_time, ad):
ad.take_bug_report(test_name, begin_time)
args = [(test_name, begin_time, ad) for ad in ads]
utils.concurrent_exec(take_br, args)
class AndroidDevice:
"""Class representing an android device.
Each object of this class represents one Android device in ACTS, including
handles to adb, fastboot, and sl4a clients. In addition to direct adb
commands, this object also uses adb port forwarding to talk to the Android
device.
Attributes:
serial: A string that's the serial number of the Android device.
log_path: A string that is the path where all logs collected on this
android device should be stored.
log: A logger adapted from root logger with added token specific to an
AndroidDevice instance.
adb_logcat_process: A process that collects the adb logcat.
adb: An AdbProxy object used for interacting with the device via adb.
fastboot: A FastbootProxy object used for interacting with the device
via fastboot.
client_port: Preferred client port number on the PC host side for SL4A
forwarded_port: Preferred server port number forwarded from Android
to the host PC via adb for SL4A connections
server_port: Preferred server port used by SL4A on Android device
"""
def __init__(
self,
serial="",
ssh_connection=None,
client_port=0,
forwarded_port=0,
server_port=None,
):
self.serial = serial
# logging.log_path only exists when this is used in an ACTS test run.
log_path_base = getattr(logging, "log_path", "/tmp/logs")
self.log_dir = f"AndroidDevice{serial}"
self.log_path = os.path.join(log_path_base, self.log_dir)
self.client_port = client_port
self.forwarded_port = forwarded_port
self.server_port = server_port
self.log = tracelogger.TraceLogger(
AndroidDeviceLoggerAdapter(logging.getLogger(), {"serial": serial})
)
self._event_dispatchers = {}
self._services = []
self.register_service(services.AdbLogcatService(self))
self.register_service(services.Sl4aService(self))
self.adb_logcat_process = None
self.adb = adb.AdbProxy(serial, ssh_connection=ssh_connection)
self.fastboot = fastboot.FastbootProxy(serial, ssh_connection=ssh_connection)
if not self.is_bootloader:
self.root_adb()
self._ssh_connection = ssh_connection
self.skip_sl4a = False
self.crash_report = None
self.data_accounting = collections.defaultdict(int)
self._sl4a_manager = sl4a_manager.create_sl4a_manager(self.adb)
self.last_logcat_timestamp = None
# Device info cache.
self._user_added_device_info = {}
self._sdk_api_level = None
def clean_up(self):
"""Cleans up the AndroidDevice object and releases any resources it
claimed.
"""
self.stop_services()
for service in self._services:
service.unregister()
self._services.clear()
if self._ssh_connection:
self._ssh_connection.close()
def recreate_services(self, serial):
"""Clean up the AndroidDevice object and re-create adb/sl4a services.
Unregister the existing services and re-create adb and sl4a services,
call this method when the connection break after certain API call
(e.g., enable USB tethering by #startTethering)
Args:
serial: the serial number of the AndroidDevice
"""
# Clean the old services
for service in self._services:
service.unregister()
self._services.clear()
if self._ssh_connection:
self._ssh_connection.close()
self._sl4a_manager.stop_service()
# Wait for old services to stop
time.sleep(5)
# Re-create the new adb and sl4a services
self.register_service(services.AdbLogcatService(self))
self.register_service(services.Sl4aService(self))
self.adb.wait_for_device()
self.terminate_all_sessions()
self.start_services()
def register_service(self, service):
"""Registers the service on the device."""
service.register()
self._services.append(service)
# TODO(angli): This function shall be refactored to accommodate all services
# and not have hard coded switch for SL4A when b/29157104 is done.
def start_services(self, skip_setup_wizard=True):
"""Starts long running services on the android device.
1. Start adb logcat capture.
2. Start SL4A if not skipped.
Args:
skip_setup_wizard: Whether or not to skip the setup wizard.
"""
if skip_setup_wizard:
self.exit_setup_wizard()
event_bus.post(android_events.AndroidStartServicesEvent(self))
def stop_services(self):
"""Stops long running services on the android device.
Stop adb logcat and terminate sl4a sessions if exist.
"""
event_bus.post(
android_events.AndroidStopServicesEvent(self), ignore_errors=True
)
def is_connected(self):
out = self.adb.devices()
devices = _parse_device_list(out, "device")
return self.serial in devices
@property
def build_info(self):
"""Get the build info of this Android device, including build id and
build type.
This is not available if the device is in bootloader mode.
Returns:
A dict with the build info of this Android device, or None if the
device is in bootloader mode.
"""
if self.is_bootloader:
self.log.error("Device is in fastboot mode, could not get build " "info.")
return
build_id = self.adb.getprop("ro.build.id")
incremental_build_id = self.adb.getprop("ro.build.version.incremental")
valid_build_id = False
for regex in RELEASE_ID_REGEXES:
if re.match(regex, build_id):
valid_build_id = True
break
if not valid_build_id:
build_id = incremental_build_id
info = {
"build_id": build_id,
"incremental_build_id": incremental_build_id,
"build_type": self.adb.getprop("ro.build.type"),
}
return info
@property
def device_info(self):
"""Information to be pulled into controller info.
The latest serial, model, and build_info are included. Additional info
can be added via `add_device_info`.
"""
info = {
"serial": self.serial,
"model": self.model,
"build_info": self.build_info,
"user_added_info": self._user_added_device_info,
"flavor": self.flavor,
}
return info
def add_device_info(self, name, info):
"""Add custom device info to the user_added_info section.
Adding the same info name the second time will override existing info.
Args:
name: string, name of this info.
info: serializable, content of the info.
"""
self._user_added_device_info.update({name: info})
def sdk_api_level(self):
if self._sdk_api_level is not None:
return self._sdk_api_level
if self.is_bootloader:
self.log.error("Device is in fastboot mode. Cannot get build info.")
return
self._sdk_api_level = int(self.adb.shell("getprop ro.build.version.sdk"))
return self._sdk_api_level
@property
def is_bootloader(self):
"""True if the device is in bootloader mode."""
return self.serial in list_fastboot_devices()
@property
def is_adb_root(self):
"""True if adb is running as root for this device."""
try:
return "0" == self.adb.shell("id -u")
except AdbError:
# Wait a bit and retry to work around adb flakiness for this cmd.
time.sleep(0.2)
return "0" == self.adb.shell("id -u")
@property
def model(self):
"""The Android code name for the device."""
# If device is in bootloader mode, get mode name from fastboot.
if self.is_bootloader:
out = self.fastboot.getvar("product").strip()
# "out" is never empty because of the "total time" message fastboot
# writes to stderr.
lines = out.split("\n", 1)
if lines:
tokens = lines[0].split(" ")
if len(tokens) > 1:
return tokens[1].lower()
return None
model = self.adb.getprop("ro.build.product").lower()
if model == "sprout":
return model
else:
return self.adb.getprop("ro.product.name").lower()
@property
def flavor(self):
"""Returns the specific flavor of Android build the device is using."""
return self.adb.getprop("ro.build.flavor").lower()
@property
def droid(self):
"""Returns the RPC Service of the first Sl4aSession created."""
if len(self._sl4a_manager.sessions) > 0:
session_id = sorted(self._sl4a_manager.sessions.keys())[0]
return self._sl4a_manager.sessions[session_id].rpc_client
else:
return None
@property
def ed(self):
"""Returns the event dispatcher of the first Sl4aSession created."""
if len(self._sl4a_manager.sessions) > 0:
session_id = sorted(self._sl4a_manager.sessions.keys())[0]
return self._sl4a_manager.sessions[session_id].get_event_dispatcher()
else:
return None
@property
def sl4a_sessions(self):
"""Returns a dictionary of session ids to sessions."""
return list(self._sl4a_manager.sessions)
@property
def is_adb_logcat_on(self):
"""Whether there is an ongoing adb logcat collection."""
if self.adb_logcat_process:
if self.adb_logcat_process.is_running():
return True
else:
# if skip_sl4a is true, there is no sl4a session
# if logcat died due to device reboot and sl4a session has
# not restarted there is no droid.
if self.droid:
self.droid.logI("Logcat died")
self.log.info("Logcat to %s died", self.log_path)
return False
return False
@property
def device_log_path(self):
"""Returns the directory for all Android device logs for the current
test context and serial.
"""
return context.get_current_context().get_full_output_path(self.serial)
def update_sdk_api_level(self):
self._sdk_api_level = None
self.sdk_api_level()
def load_config(self, config):
"""Add attributes to the AndroidDevice object based on json config.
Args:
config: A dictionary representing the configs.
Raises:
AndroidDeviceError is raised if the config is trying to overwrite
an existing attribute.
"""
for k, v in config.items():
# skip_sl4a value can be reset from config file
if hasattr(self, k) and k != "skip_sl4a":
raise errors.AndroidDeviceError(
f"Attempting to set existing attribute {k} on {self.serial}",
serial=self.serial,
)
setattr(self, k, v)
def root_adb(self):
"""Change adb to root mode for this device if allowed.
If executed on a production build, adb will not be switched to root
mode per security restrictions.
"""
if self.is_adb_root:
return
for attempt in range(ADB_ROOT_RETRY_COUNT):
try:
self.log.debug(f"Enabling ADB root mode: attempt {attempt}.")
self.adb.root()
except AdbError:
if attempt == ADB_ROOT_RETRY_COUNT:
raise
time.sleep(ADB_ROOT_RETRY_INTERVAL)
self.adb.wait_for_device()
def get_droid(self, handle_event=True):
"""Create an sl4a connection to the device.
Return the connection handler 'droid'. By default, another connection
on the same session is made for EventDispatcher, and the dispatcher is
returned to the caller as well.
If sl4a server is not started on the device, try to start it.
Args:
handle_event: True if this droid session will need to handle
events.
Returns:
droid: Android object used to communicate with sl4a on the android
device.
ed: An optional EventDispatcher to organize events for this droid.
Examples:
Don't need event handling:
>>> ad = AndroidDevice()
>>> droid = ad.get_droid(False)
Need event handling:
>>> ad = AndroidDevice()
>>> droid, ed = ad.get_droid()
"""
self.log.debug(
"Creating RPC client_port={}, forwarded_port={}, server_port={}".format(
self.client_port, self.forwarded_port, self.server_port
)
)
session = self._sl4a_manager.create_session(
client_port=self.client_port,
forwarded_port=self.forwarded_port,
server_port=self.server_port,
)
droid = session.rpc_client
if handle_event:
ed = session.get_event_dispatcher()
return droid, ed
return droid
def get_package_pid(self, package_name):
"""Gets the pid for a given package. Returns None if not running.
Args:
package_name: The name of the package.
Returns:
The first pid found under a given package name. None if no process
was found running the package.
Raises:
AndroidDeviceError if the output of the phone's process list was
in an unexpected format.
"""
for cmd in ("ps -A", "ps"):
try:
out = self.adb.shell(
f'{cmd} | grep "S {package_name}"', ignore_status=True
)
if package_name not in out:
continue
try:
pid = int(out.split()[1])
self.log.info("apk %s has pid %s.", package_name, pid)
return pid
except (IndexError, ValueError) as e:
# Possible ValueError from string to int cast.
# Possible IndexError from split.
self.log.warning(
'Command "%s" returned output line: ' '"%s".\nError: %s',
cmd,
out,
e,
)
except Exception as e:
self.log.warning(
'Device fails to check if %s running with "%s"\n' "Exception %s",
package_name,
cmd,
e,
)
self.log.debug("apk %s is not running", package_name)
return None
def get_dispatcher(self, droid):
"""Return an EventDispatcher for an sl4a session
Args:
droid: Session to create EventDispatcher for.
Returns:
ed: An EventDispatcher for specified session.
"""
return self._sl4a_manager.sessions[droid.uid].get_event_dispatcher()
def _is_timestamp_in_range(self, target, log_begin_time, log_end_time):
low = acts_logger.logline_timestamp_comparator(log_begin_time, target) <= 0
high = acts_logger.logline_timestamp_comparator(log_end_time, target) >= 0
return low and high
def cat_adb_log(self, tag, begin_time, end_time=None, dest_path="AdbLogExcerpts"):
"""Takes an excerpt of the adb logcat log from a certain time point to
current time.
Args:
tag: An identifier of the time period, usually the name of a test.
begin_time: Epoch time of the beginning of the time period.
end_time: Epoch time of the ending of the time period, default None
dest_path: Destination path of the excerpt file.
"""
log_begin_time = acts_logger.epoch_to_log_line_timestamp(begin_time)
if end_time is None:
log_end_time = acts_logger.get_log_line_timestamp()
else:
log_end_time = acts_logger.epoch_to_log_line_timestamp(end_time)
self.log.debug("Extracting adb log from logcat.")
logcat_path = os.path.join(
self.device_log_path, f"adblog_{self.serial}_debug.txt"
)
if not os.path.exists(logcat_path):
self.log.warning(f"Logcat file {logcat_path} does not exist.")
return
adb_excerpt_dir = os.path.join(self.log_path, dest_path)
os.makedirs(adb_excerpt_dir, exist_ok=True)
out_name = "%s,%s.txt" % (
acts_logger.normalize_log_line_timestamp(log_begin_time),
self.serial,
)
tag_len = utils.MAX_FILENAME_LEN - len(out_name)
out_name = f"{tag[:tag_len]},{out_name}"
adb_excerpt_path = os.path.join(adb_excerpt_dir, out_name)
with open(adb_excerpt_path, "w", encoding="utf-8") as out:
in_file = logcat_path
with open(in_file, "r", encoding="utf-8", errors="replace") as f:
while True:
line = None
try:
line = f.readline()
if not line:
break
except:
continue
line_time = line[: acts_logger.log_line_timestamp_len]
if not acts_logger.is_valid_logline_timestamp(line_time):
continue
if self._is_timestamp_in_range(
line_time, log_begin_time, log_end_time
):
if not line.endswith("\n"):
line += "\n"
out.write(line)
return adb_excerpt_path
def search_logcat(
self, matching_string, begin_time=None, end_time=None, logcat_path=None
):
"""Search logcat message with given string.
Args:
matching_string: matching_string to search.
begin_time: only the lines with time stamps later than begin_time
will be searched.
end_time: only the lines with time stamps earlier than end_time
will be searched.
logcat_path: the path of a specific file in which the search should
be performed. If None the path will be the default device log
path.
Returns:
A list of dictionaries with full log message, time stamp string,
time object and message ID. For example:
[{"log_message": "05-03 17:39:29.898 968 1001 D"
"ActivityManager: Sending BOOT_COMPLETE user #0",
"time_stamp": "2017-05-03 17:39:29.898",
"datetime_obj": datetime object,
"message_id": None}]
[{"log_message": "08-12 14:26:42.611043 2360 2510 D RILJ : "
"[0853]< DEACTIVATE_DATA_CALL [PHONE0]",
"time_stamp": "2020-08-12 14:26:42.611043",
"datetime_obj": datetime object},
"message_id": "0853"}]
"""
if not logcat_path:
logcat_path = os.path.join(
self.device_log_path, f"adblog_{self.serial}_debug.txt"
)
if not os.path.exists(logcat_path):
self.log.warning(f"Logcat file {logcat_path} does not exist.")
return
output = job.run(f"grep '{matching_string}' {logcat_path}", ignore_status=True)
if not output.stdout or output.exit_status != 0:
return []
if begin_time:
if not isinstance(begin_time, datetime):
log_begin_time = acts_logger.epoch_to_log_line_timestamp(begin_time)
begin_time = datetime.strptime(log_begin_time, "%Y-%m-%d %H:%M:%S.%f")
if end_time:
if not isinstance(end_time, datetime):
log_end_time = acts_logger.epoch_to_log_line_timestamp(end_time)
end_time = datetime.strptime(log_end_time, "%Y-%m-%d %H:%M:%S.%f")
result = []
logs = re.findall(r"(\S+\s\S+)(.*)", output.stdout)
for log in logs:
time_stamp = log[0]
time_obj = datetime.strptime(time_stamp, "%Y-%m-%d %H:%M:%S.%f")
if begin_time and time_obj < begin_time:
continue
if end_time and time_obj > end_time:
continue
res = re.findall(r".*\[(\d+)\]", log[1])
try:
message_id = res[0]
except:
message_id = None
result.append(
{
"log_message": "".join(log),
"time_stamp": time_stamp,
"datetime_obj": time_obj,
"message_id": message_id,
}
)
return result
def start_adb_logcat(self):
"""Starts a standing adb logcat collection in separate subprocesses and
save the logcat in a file.
"""
if self.is_adb_logcat_on:
self.log.warning(
"Android device %s already has a running adb logcat thread. "
% self.serial
)
return
# Disable adb log spam filter. Have to stop and clear settings first
# because 'start' doesn't support --clear option before Android N.
self.adb.shell("logpersist.stop --clear", ignore_status=True)
self.adb.shell("logpersist.start", ignore_status=True)
if hasattr(self, "adb_logcat_param"):
extra_params = self.adb_logcat_param
else:
extra_params = "-b all"
self.adb_logcat_process = logcat.create_logcat_keepalive_process(
self.serial, self.log_dir, extra_params
)
self.adb_logcat_process.start()
def stop_adb_logcat(self):
"""Stops the adb logcat collection subprocess."""
if not self.is_adb_logcat_on:
self.log.warning(
f"Android device {self.serial} does not have an ongoing adb logcat "
)
return
# Set the last timestamp to the current timestamp. This may cause
# a race condition that allows the same line to be logged twice,
# but it does not pose a problem for our logging purposes.
self.adb_logcat_process.stop()
self.adb_logcat_process = None
def get_apk_uid(self, apk_name):
"""Get the uid of the given apk.
Args:
apk_name: Name of the package, e.g., com.android.phone.
Returns:
Linux UID for the apk.
"""
output = self.adb.shell(
f"dumpsys package {apk_name} | grep userId=", ignore_status=True
)
result = re.search(r"userId=(\d+)", output)
if result:
return result.group(1)
else:
None
def get_apk_version(self, package_name):
"""Get the version of the given apk.
Args:
package_name: Name of the package, e.g., com.android.phone.
Returns:
Version of the given apk.
"""
try:
output = self.adb.shell(
f"dumpsys package {package_name} | grep versionName"
)
pattern = re.compile(r"versionName=(.+)", re.I)
result = pattern.findall(output)
if result:
return result[0]
except Exception as e:
self.log.warning(
"Fail to get the version of package %s: %s", package_name, e
)
self.log.debug("apk %s is not found", package_name)
return None
def is_apk_installed(self, package_name):
"""Check if the given apk is already installed.
Args:
package_name: Name of the package, e.g., com.android.phone.
Returns:
True if package is installed. False otherwise.
"""
try:
return bool(
self.adb.shell(
f'(pm list packages | grep -w "package:{package_name}") || true'
)
)
except Exception as err:
self.log.error(
"Could not determine if %s is installed. " "Received error:\n%s",
package_name,
err,
)
return False
def is_sl4a_installed(self):
return self.is_apk_installed(SL4A_APK_NAME)
def is_apk_running(self, package_name):
"""Check if the given apk is running.
Args:
package_name: Name of the package, e.g., com.android.phone.
Returns:
True if package is installed. False otherwise.
"""
for cmd in ("ps -A", "ps"):
try:
out = self.adb.shell(
f'{cmd} | grep "S {package_name}"', ignore_status=True
)
if package_name in out:
self.log.info("apk %s is running", package_name)
return True
except Exception as e:
self.log.warning(
"Device fails to check is %s running by %s " "Exception %s",
package_name,
cmd,
e,
)
continue
self.log.debug("apk %s is not running", package_name)
return False
def is_sl4a_running(self):
return self.is_apk_running(SL4A_APK_NAME)
def force_stop_apk(self, package_name):
"""Force stop the given apk.
Args:
package_name: Name of the package, e.g., com.android.phone.
Returns:
True if package is installed. False otherwise.
"""
try:
self.adb.shell(f"am force-stop {package_name}", ignore_status=True)
except Exception as e:
self.log.warning("Fail to stop package %s: %s", package_name, e)
def take_bug_report(self, test_name=None, begin_time=None):
"""Takes a bug report on the device and stores it in a file.
Args:
test_name: Name of the test case that triggered this bug report.
begin_time: Epoch time when the test started. If none is specified,
the current time will be used.
"""
self.adb.wait_for_device(timeout=WAIT_FOR_DEVICE_TIMEOUT)
new_br = True
try:
stdout = self.adb.shell("bugreportz -v")
# This check is necessary for builds before N, where adb shell's ret
# code and stderr are not propagated properly.
if "not found" in stdout:
new_br = False
except AdbError:
new_br = False
br_path = self.device_log_path
os.makedirs(br_path, exist_ok=True)
epoch = begin_time if begin_time else utils.get_current_epoch_time()
time_stamp = acts_logger.normalize_log_line_timestamp(
acts_logger.epoch_to_log_line_timestamp(epoch)
)
out_name = f"AndroidDevice{self.serial}_{time_stamp}"
out_name = f"{out_name}.zip" if new_br else f"{out_name}.txt"
full_out_path = os.path.join(br_path, out_name)
# in case device restarted, wait for adb interface to return
self.wait_for_boot_completion()
if test_name:
self.log.info("Taking bugreport for %s.", test_name)
else:
self.log.info("Taking bugreport.")
if new_br:
out = self.adb.shell("bugreportz", timeout=BUG_REPORT_TIMEOUT)
if not out.startswith("OK"):
raise errors.AndroidDeviceError(
f"Failed to take bugreport on {self.serial}: {out}",
serial=self.serial,
)
br_out_path = out.split(":")[1].strip().split()[0]
self.adb.pull(f"{br_out_path} {full_out_path}")
else:
self.adb.bugreport(f" > {full_out_path}", timeout=BUG_REPORT_TIMEOUT)
if test_name:
self.log.info("Bugreport for %s taken at %s.", test_name, full_out_path)
else:
self.log.info("Bugreport taken at %s.", test_name, full_out_path)
self.adb.wait_for_device(timeout=WAIT_FOR_DEVICE_TIMEOUT)
def get_file_names(
self, directory, begin_time=None, skip_files=[], match_string=None
):
"""Get files names with provided directory."""
cmd = f"find {directory} -type f"
if begin_time:
current_time = utils.get_current_epoch_time()
seconds = int(math.ceil((current_time - begin_time) / 1000.0))
cmd = f"{cmd} -mtime -{seconds}s"
if match_string:
cmd = f"{cmd} -iname {match_string}"
for skip_file in skip_files:
cmd = f"{cmd} ! -iname {skip_file}"
out = self.adb.shell(cmd, ignore_status=True)
if (
not out
or "No such" in out
or "Permission denied" in out
or "Not a directory" in out
):
return []
files = out.split("\n")
self.log.debug("Find files in directory %s: %s", directory, files)
return files
@property
def external_storage_path(self):
"""
The $EXTERNAL_STORAGE path on the device. Most commonly set to '/sdcard'
"""
return self.adb.shell("echo $EXTERNAL_STORAGE")
def file_exists(self, file_path):
"""Returns whether a file exists on a device.
Args:
file_path: The path of the file to check for.
"""
cmd = f"(test -f {file_path} && echo yes) || echo no"
result = self.adb.shell(cmd)
if result == "yes":
return True
elif result == "no":
return False
raise ValueError(
"Couldn't determine if %s exists. "
"Expected yes/no, got %s" % (file_path, result[cmd])
)
def pull_files(self, device_paths, host_path=None):
"""Pull files from devices.
Args:
device_paths: List of paths on the device to pull from.
host_path: Destination path
"""
if isinstance(device_paths, str):
device_paths = [device_paths]
if not host_path:
host_path = self.log_path
for device_path in device_paths:
self.log.info(f"Pull from device: {device_path} -> {host_path}")
self.adb.pull(f"{device_path} {host_path}", timeout=PULL_TIMEOUT)
def check_crash_report(
self, test_name=None, begin_time=None, log_crash_report=False
):
"""check crash report on the device."""
crash_reports = []
for crash_path in CRASH_REPORT_PATHS:
try:
cmd = f"cd {crash_path}"
self.adb.shell(cmd)
except Exception as e:
self.log.debug("received exception %s", e)
continue
crashes = self.get_file_names(
crash_path, skip_files=CRASH_REPORT_SKIPS, begin_time=begin_time
)
if crash_path == "/data/tombstones/" and crashes:
tombstones = crashes[:]
for tombstone in tombstones:
if self.adb.shell(
f'cat {tombstone} | grep "crash_dump failed to dump process"'
):
crashes.remove(tombstone)
if crashes:
crash_reports.extend(crashes)
if crash_reports and log_crash_report:
crash_log_path = os.path.join(
self.device_log_path, f"Crashes_{self.serial}"
)
os.makedirs(crash_log_path, exist_ok=True)
self.pull_files(crash_reports, crash_log_path)
return crash_reports
def get_qxdm_logs(self, test_name="", begin_time=None):
"""Get qxdm logs."""
# Sleep 10 seconds for the buffered log to be written in qxdm log file
time.sleep(10)
log_path = getattr(self, "qxdm_log_path", DEFAULT_QXDM_LOG_PATH)
qxdm_logs = self.get_file_names(
log_path, begin_time=begin_time, match_string="*.qmdl"
)
if qxdm_logs:
qxdm_log_path = os.path.join(self.device_log_path, f"QXDM_{self.serial}")
os.makedirs(qxdm_log_path, exist_ok=True)
self.log.info("Pull QXDM Log %s to %s", qxdm_logs, qxdm_log_path)
self.pull_files(qxdm_logs, qxdm_log_path)
self.adb.pull(
f"/firmware/image/qdsp6m.qdb {qxdm_log_path}",
timeout=PULL_TIMEOUT,
ignore_status=True,
)
# Zip Folder
utils.zip_directory(f"{qxdm_log_path}.zip", qxdm_log_path)
shutil.rmtree(qxdm_log_path)
else:
self.log.error(f"Didn't find QXDM logs in {log_path}.")
if "Verizon" in self.adb.getprop("gsm.sim.operator.alpha"):
omadm_log_path = os.path.join(self.device_log_path, f"OMADM_{self.serial}")
os.makedirs(omadm_log_path, exist_ok=True)
self.log.info("Pull OMADM Log")
self.adb.pull(
f"/data/data/com.android.omadm.service/files/dm/log/ {omadm_log_path}",
timeout=PULL_TIMEOUT,
ignore_status=True,
)
def get_sdm_logs(self, test_name="", begin_time=None):
"""Get sdm logs."""
# Sleep 10 seconds for the buffered log to be written in sdm log file
time.sleep(10)
log_paths = [
ALWAYS_ON_LOG_PATH,
getattr(self, "sdm_log_path", DEFAULT_SDM_LOG_PATH),
]
sdm_logs = []
for path in log_paths:
sdm_logs += self.get_file_names(
path, begin_time=begin_time, match_string="*.sdm*"
)
if sdm_logs:
sdm_log_path = os.path.join(self.device_log_path, f"SDM_{self.serial}")
os.makedirs(sdm_log_path, exist_ok=True)
self.log.info("Pull SDM Log %s to %s", sdm_logs, sdm_log_path)
self.pull_files(sdm_logs, sdm_log_path)
else:
self.log.error(f"Didn't find SDM logs in {log_paths}.")
if "Verizon" in self.adb.getprop("gsm.sim.operator.alpha"):
omadm_log_path = os.path.join(self.device_log_path, f"OMADM_{self.serial}")
os.makedirs(omadm_log_path, exist_ok=True)
self.log.info("Pull OMADM Log")
self.adb.pull(
f"/data/data/com.android.omadm.service/files/dm/log/ {omadm_log_path}",
timeout=PULL_TIMEOUT,
ignore_status=True,
)
def start_new_session(self, max_connections=None, server_port=None):
"""Start a new session in sl4a.
Also caches the droid in a dict with its uid being the key.
Returns:
An Android object used to communicate with sl4a on the android
device.
Raises:
Sl4aException: Something is wrong with sl4a and it returned an
existing uid to a new session.
"""
session = self._sl4a_manager.create_session(
max_connections=max_connections, server_port=server_port
)
self._sl4a_manager.sessions[session.uid] = session
return session.rpc_client
def terminate_all_sessions(self):
"""Terminate all sl4a sessions on the AndroidDevice instance.
Terminate all sessions and clear caches.
"""
self._sl4a_manager.terminate_all_sessions()
def run_iperf_client_nb(
self, server_host, extra_args="", timeout=IPERF_TIMEOUT, log_file_path=None
):
"""Start iperf client on the device asynchronously.
Return status as true if iperf client start successfully.
And data flow information as results.
Args:
server_host: Address of the iperf server.
extra_args: A string representing extra arguments for iperf client,
e.g. "-i 1 -t 30".
log_file_path: The complete file path to log the results.
"""
cmd = f"iperf3 -c {server_host} {extra_args}"
if log_file_path:
cmd += f" --logfile {log_file_path} &"
self.adb.shell_nb(cmd)
def run_iperf_client(self, server_host, extra_args="", timeout=IPERF_TIMEOUT):
"""Start iperf client on the device.
Return status as true if iperf client start successfully.
And data flow information as results.
Args:
server_host: Address of the iperf server.
extra_args: A string representing extra arguments for iperf client,
e.g. "-i 1 -t 30".
Returns:
status: true if iperf client start successfully.
results: results have data flow information
"""
out = self.adb.shell(f"iperf3 -c {server_host} {extra_args}", timeout=timeout)
clean_out = out.split("\n")
if "error" in clean_out[0].lower():
return False, clean_out
return True, clean_out
def run_iperf_server(self, extra_args=""):
"""Start iperf server on the device
Return status as true if iperf server started successfully.
Args:
extra_args: A string representing extra arguments for iperf server.
Returns:
status: true if iperf server started successfully.
results: results have output of command
"""
out = self.adb.shell(f"iperf3 -s {extra_args}")
clean_out = out.split("\n")
if "error" in clean_out[0].lower():
return False, clean_out
return True, clean_out
def wait_for_boot_completion(self, timeout=900.0):
"""Waits for Android framework to broadcast ACTION_BOOT_COMPLETED.
Args:
timeout: Seconds to wait for the device to boot. Default value is
15 minutes.
"""
timeout_start = time.time()
self.log.debug("ADB waiting for device")
self.adb.wait_for_device(timeout=timeout)
self.log.debug("Waiting for sys.boot_completed")
while time.time() < timeout_start + timeout:
try:
completed = self.adb.getprop("sys.boot_completed")
if completed == "1":
self.log.debug("Device has rebooted")
return
except AdbError:
# adb shell calls may fail during certain period of booting
# process, which is normal. Ignoring these errors.
pass
time.sleep(5)
raise errors.AndroidDeviceError(
f"Device {self.serial} booting process timed out.", serial=self.serial
)
def reboot(
self, stop_at_lock_screen=False, timeout=180, wait_after_reboot_complete=1
):
"""Reboots the device.
Terminate all sl4a sessions, reboot the device, wait for device to
complete booting, and restart an sl4a session if restart_sl4a is True.
Args:
stop_at_lock_screen: whether to unlock after reboot. Set to False
if want to bring the device to reboot up to password locking
phase. Sl4a checking need the device unlocked after rebooting.
timeout: time in seconds to wait for the device to complete
rebooting.
wait_after_reboot_complete: time in seconds to wait after the boot
completion.
"""
if self.is_bootloader:
self.fastboot.reboot()
return
self.stop_services()
self.log.info("Rebooting")
self.adb.reboot()
timeout_start = time.time()
# b/111791239: Newer versions of android sometimes return early after
# `adb reboot` is called. This means subsequent calls may make it to
# the device before the reboot goes through, return false positives for
# getprops such as sys.boot_completed.
while time.time() < timeout_start + timeout:
try:
self.adb.get_state()
time.sleep(0.1)
except AdbError:
# get_state will raise an error if the device is not found. We
# want the device to be missing to prove the device has kicked
# off the reboot.
break
self.wait_for_boot_completion(timeout=(timeout - time.time() + timeout_start))
self.log.debug("Wait for a while after boot completion.")
time.sleep(wait_after_reboot_complete)
self.root_adb()
skip_sl4a = self.skip_sl4a
self.skip_sl4a = self.skip_sl4a or stop_at_lock_screen
self.start_services()
self.skip_sl4a = skip_sl4a
def restart_runtime(self):
"""Restarts android runtime.
Terminate all sl4a sessions, restarts runtime, wait for framework
complete restart, and restart an sl4a session if restart_sl4a is True.
"""
self.stop_services()
self.log.info("Restarting android runtime")
self.adb.shell("stop")
# Reset the boot completed flag before we restart the framework
# to correctly detect when the framework has fully come up.
self.adb.shell("setprop sys.boot_completed 0")
self.adb.shell("start")
self.wait_for_boot_completion()
self.root_adb()
self.start_services()
def get_ipv4_address(self, interface="wlan0", timeout=5):
for timer in range(0, timeout):
try:
ip_string = self.adb.shell(f"ifconfig {interface}|grep inet")
break
except adb.AdbError as e:
if timer + 1 == timeout:
self.log.warning(f"Unable to find IP address for {interface}.")
return None
else:
time.sleep(1)
result = re.search("addr:(.*) Bcast", ip_string)
if result != None:
ip_address = result.group(1)
try:
socket.inet_aton(ip_address)
return ip_address
except socket.error:
return None
else:
return None
def get_ipv4_gateway(self, timeout=5):
for timer in range(0, timeout):
try:
gateway_string = self.adb.shell("dumpsys wifi | grep mDhcpResults")
break
except adb.AdbError as e:
if timer + 1 == timeout:
self.log.warning("Unable to find gateway")
return None
else:
time.sleep(1)
result = re.search("Gateway (.*) DNS servers", gateway_string)
if result != None:
ipv4_gateway = result.group(1)
try:
socket.inet_aton(ipv4_gateway)
return ipv4_gateway
except socket.error:
return None
else:
return None
def send_keycode(self, keycode):
self.adb.shell(f"input keyevent KEYCODE_{keycode}")
def get_my_current_focus_window(self):
"""Get the current focus window on screen"""
output = self.adb.shell(
"dumpsys window displays | grep -E mCurrentFocus | grep -v null",
ignore_status=True,
)
if not output or "not found" in output or "Can't find" in output:
result = ""
else:
result = output.split(" ")[-1].strip("}")
self.log.debug("Current focus window is %s", result)
return result
def get_my_current_focus_app(self):
"""Get the current focus application"""
dumpsys_cmd = [
"dumpsys window | grep -E mFocusedApp",
"dumpsys window displays | grep -E mFocusedApp",
]
for cmd in dumpsys_cmd:
output = self.adb.shell(cmd, ignore_status=True)
if (
not output
or "not found" in output
or "Can't find" in output
or ("mFocusedApp=null" in output)
):
result = ""
else:
result = output.split(" ")[-2]
break
self.log.debug("Current focus app is %s", result)
return result
def is_window_ready(self, window_name=None):
current_window = self.get_my_current_focus_window()
if window_name:
return window_name in current_window
return current_window and ENCRYPTION_WINDOW not in current_window
def wait_for_window_ready(
self, window_name=None, check_interval=5, check_duration=60
):
elapsed_time = 0
while elapsed_time < check_duration:
if self.is_window_ready(window_name=window_name):
return True
time.sleep(check_interval)
elapsed_time += check_interval
self.log.info("Current focus window is %s", self.get_my_current_focus_window())
return False
def is_user_setup_complete(self):
return "1" in self.adb.shell("settings get secure user_setup_complete")
def is_screen_awake(self):
"""Check if device screen is in sleep mode"""
return "Awake" in self.adb.shell("dumpsys power | grep mWakefulness=")
def is_screen_emergency_dialer(self):
"""Check if device screen is in emergency dialer mode"""
return "EmergencyDialer" in self.get_my_current_focus_window()
def is_screen_in_call_activity(self):
"""Check if device screen is in in-call activity notification"""
return "InCallActivity" in self.get_my_current_focus_window()
def is_setupwizard_on(self):
"""Check if device screen is in emergency dialer mode"""
return "setupwizard" in self.get_my_current_focus_app()
def is_screen_lock_enabled(self):
"""Check if screen lock is enabled"""
cmd = "dumpsys window policy | grep showing="
out = self.adb.shell(cmd, ignore_status=True)
return "true" in out
def is_waiting_for_unlock_pin(self):
"""Check if device is waiting for unlock pin to boot up"""
current_window = self.get_my_current_focus_window()
current_app = self.get_my_current_focus_app()
if ENCRYPTION_WINDOW in current_window:
self.log.info("Device is in CrpytKeeper window")
return True
if "StatusBar" in current_window and (
(not current_app) or "FallbackHome" in current_app
):
self.log.info("Device is locked")
return True
return False
def ensure_screen_on(self):
"""Ensure device screen is powered on"""
if self.is_screen_lock_enabled():
for _ in range(2):
self.unlock_screen()
time.sleep(1)
if self.is_waiting_for_unlock_pin():
self.unlock_screen(password=DEFAULT_DEVICE_PASSWORD)
time.sleep(1)
if (
not self.is_waiting_for_unlock_pin()
and self.wait_for_window_ready()
):
return True
return False
else:
self.wakeup_screen()
return True
def wakeup_screen(self):
if not self.is_screen_awake():
self.log.info("Screen is not awake, wake it up")
self.send_keycode("WAKEUP")
def go_to_sleep(self):
if self.is_screen_awake():
self.send_keycode("SLEEP")
def send_keycode_number_pad(self, number):
self.send_keycode(f"NUMPAD_{number}")
def unlock_screen(self, password=None):
self.log.info("Unlocking with %s", password or "swipe up")
# Bring device to SLEEP so that unlock process can start fresh
self.send_keycode("SLEEP")
time.sleep(1)
self.send_keycode("WAKEUP")
if ENCRYPTION_WINDOW not in self.get_my_current_focus_app():
self.send_keycode("MENU")
if password:
self.send_keycode("DEL")
for number in password:
self.send_keycode_number_pad(number)
self.send_keycode("ENTER")
self.send_keycode("BACK")
def screenshot(self, name=""):
"""Take a screenshot on the device.
Args:
name: additional information of screenshot on the file name.
"""
if name:
file_name = f"{DEFAULT_SCREENSHOT_PATH}_{name}"
file_name = f"{file_name}_{utils.get_current_epoch_time()}.png"
self.ensure_screen_on()
self.log.info("Log screenshot to %s", file_name)
try:
self.adb.shell(f"screencap -p {file_name}")
except:
self.log.error("Fail to log screenshot to %s", file_name)
def exit_setup_wizard(self):
# Handling Android TV's setupwizard is ignored for now.
if "feature:android.hardware.type.television" in self.adb.shell(
"pm list features"
):
return
if not self.is_user_setup_complete() or self.is_setupwizard_on():
# b/116709539 need this to prevent reboot after skip setup wizard
self.adb.shell(
"am start -a com.android.setupwizard.EXIT", ignore_status=True
)
self.adb.shell(
f"pm disable {self.get_setupwizard_package_name()}",
ignore_status=True,
)
# Wait up to 5 seconds for user_setup_complete to be updated
end_time = time.time() + 5
while time.time() < end_time:
if self.is_user_setup_complete() or not self.is_setupwizard_on():
return
# If fail to exit setup wizard, set local.prop and reboot
if not self.is_user_setup_complete() and self.is_setupwizard_on():
self.adb.shell("echo ro.test_harness=1 > /data/local.prop")
self.adb.shell("chmod 644 /data/local.prop")
self.reboot(stop_at_lock_screen=True)
def get_setupwizard_package_name(self):
"""Finds setupwizard package/.activity
Bypass setupwizard or setupwraith depending on device.
Returns:
packageName/.ActivityName
"""
packages_to_skip = "'setupwizard|setupwraith'"
android_package_name = "com.google.android"
package = self.adb.shell(
"pm list packages -f | grep -E {} | grep {}".format(
packages_to_skip, android_package_name
)
)
wizard_package = package.split("=")[1]
activity = package.split("=")[0].split("/")[-2]
self.log.info(f"{wizard_package}/.{activity}Activity")
return f"{wizard_package}/.{activity}Activity"
def push_system_file(self, src_file_path, dst_file_path, push_timeout=300):
"""Pushes a file onto the read-only file system.
For speed, the device is left in root mode after this call, and leaves
verity disabled. To re-enable verity, call ensure_verity_enabled().
Args:
src_file_path: The path to the system app to install.
dst_file_path: The destination of the file.
push_timeout: How long to wait for the push to finish.
Returns:
Whether or not the install was successful.
"""
self.adb.ensure_root()
try:
self.ensure_verity_disabled()
self.adb.remount()
out = self.adb.push(
f"{src_file_path} {dst_file_path}", timeout=push_timeout
)
if "error" in out:
self.log.error(
"Unable to push system file %s to %s due to %s",
src_file_path,
dst_file_path,
out,
)
return False
return True
except Exception as e:
self.log.error(
"Unable to push system file %s to %s due to %s",
src_file_path,
dst_file_path,
e,
)
return False
def ensure_verity_enabled(self):
"""Ensures that verity is enabled.
If verity is not enabled, this call will reboot the phone. Note that
this only works on debuggable builds.
"""
user = self.adb.get_user_id()
# The below properties will only exist if verity has been enabled.
system_verity = self.adb.getprop("partition.system.verified")
vendor_verity = self.adb.getprop("partition.vendor.verified")
if not system_verity or not vendor_verity:
self.adb.ensure_root()
self.adb.enable_verity()
self.reboot()
self.adb.ensure_user(user)
def ensure_verity_disabled(self):
"""Ensures that verity is disabled.
If verity is enabled, this call will reboot the phone.
"""
user = self.adb.get_user_id()
# The below properties will only exist if verity has been enabled.
system_verity = self.adb.getprop("partition.system.verified")
vendor_verity = self.adb.getprop("partition.vendor.verified")
if system_verity or vendor_verity:
self.adb.ensure_root()
self.adb.disable_verity()
self.reboot()
self.adb.ensure_user(user)
class AndroidDeviceLoggerAdapter(logging.LoggerAdapter):
def process(self, msg, kwargs):
msg = f"[AndroidDevice|{self.extra['serial']}] {msg}"
return (msg, kwargs)