416 lines
16 KiB
Python
416 lines
16 KiB
Python
# This is the interface for adb
|
|
import subprocess
|
|
import logging
|
|
import re
|
|
import time
|
|
try:
|
|
from shlex import quote # Python 3
|
|
except ImportError:
|
|
from pipes import quote # Python 2
|
|
|
|
from ....exceptions import ADBException
|
|
|
|
|
|
class ADB(object):
|
|
"""
|
|
interface of ADB
|
|
send adb commands via this, see:
|
|
http://developer.android.com/tools/help/adb.html
|
|
"""
|
|
|
|
DOWN_AND_UP = 2
|
|
MODEL_PROPERTY = "ro.product.model"
|
|
VERSION_SDK_PROPERTY = 'ro.build.version.sdk'
|
|
VERSION_RELEASE_PROPERTY = 'ro.build.version.release'
|
|
|
|
def __init__(self, device=None):
|
|
"""
|
|
initiate a ADB connection from serial no
|
|
the serial no should be in output of `adb devices`
|
|
:param device: instance of Device
|
|
:return:
|
|
"""
|
|
self.logger = logging.getLogger(self.__class__.__name__)
|
|
if device is None:
|
|
from DroidBot.platforms.android import AndroidDevice as Device
|
|
device = Device()
|
|
self.device = device
|
|
|
|
self.cmd_prefix = ['adb', "-s", device.serial]
|
|
|
|
self._display_info_cache = None
|
|
self._display_info_cache_time = 0
|
|
self._sdk_version_cache = None
|
|
self.last_error = None
|
|
|
|
def _check_adb_connection(self):
|
|
"""检查 adb 连接状态"""
|
|
try:
|
|
result = subprocess.check_output(
|
|
['adb', '-s', self.device.serial, 'get-state'],
|
|
stderr=subprocess.STDOUT,
|
|
timeout=5
|
|
).strip()
|
|
if isinstance(result, bytes):
|
|
result = result.decode()
|
|
return result.startswith('device')
|
|
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
|
|
return False
|
|
|
|
def run_cmd(self, extra_args):
|
|
"""
|
|
run an adb command and return the output
|
|
:return: output of adb command
|
|
@param extra_args: arguments to run in adb
|
|
"""
|
|
if isinstance(extra_args, str) or isinstance(extra_args, str):
|
|
extra_args = extra_args.split()
|
|
if not isinstance(extra_args, list):
|
|
msg = "invalid arguments: %s\nshould be list or str, %s given" % (extra_args, type(extra_args))
|
|
self.logger.warning(msg)
|
|
raise ADBException(msg)
|
|
|
|
# 执行命令前检查连接状态
|
|
if not self._check_adb_connection():
|
|
error_msg = f"ADB device {self.device.serial} is not connected"
|
|
self.logger.error(error_msg)
|
|
raise ADBException(error_msg)
|
|
|
|
args = [] + self.cmd_prefix
|
|
args += extra_args
|
|
|
|
self.logger.debug('command:')
|
|
self.logger.debug(args)
|
|
|
|
try:
|
|
r = subprocess.check_output(args, stderr=subprocess.STDOUT, timeout=30).strip()
|
|
except subprocess.CalledProcessError as e:
|
|
error_msg = f"ADB command failed: {' '.join(args)}\nOutput: {e.output.decode('utf-8') if e.output else ''}"
|
|
self.logger.error(error_msg)
|
|
self.last_error = error_msg
|
|
return ""
|
|
except subprocess.TimeoutExpired as e:
|
|
error_msg = f"ADB command timeout: {' '.join(args)}"
|
|
self.logger.error(error_msg)
|
|
self.last_error = error_msg
|
|
raise ADBException(error_msg)
|
|
|
|
if not isinstance(r, str):
|
|
r = r.decode()
|
|
self.logger.debug('return:')
|
|
self.logger.debug(r)
|
|
return r
|
|
|
|
def shell(self, extra_args):
|
|
"""
|
|
run an `adb shell` command
|
|
@param extra_args:
|
|
@return: output of adb shell command
|
|
"""
|
|
if isinstance(extra_args, str) or isinstance(extra_args, str):
|
|
extra_args = extra_args.split()
|
|
if not isinstance(extra_args, list):
|
|
msg = "invalid arguments: %s\nshould be list or str, %s given" % (extra_args, type(extra_args))
|
|
self.logger.warning(msg)
|
|
raise ADBException(msg)
|
|
|
|
shell_extra_args = ['shell'] + [ quote(arg) for arg in extra_args ]
|
|
return self.run_cmd(shell_extra_args)
|
|
|
|
def shell_grep(self, cmd):
|
|
"""执行真实的 ADB 命令"""
|
|
# 执行命令前检查连接状态
|
|
if not self._check_adb_connection():
|
|
error_msg = f"ADB device {self.device.serial} is not connected"
|
|
self.logger.error(error_msg)
|
|
raise ADBException(error_msg)
|
|
|
|
try:
|
|
# 这里的 shell=True 是为了支持管道符 | grep
|
|
full_cmd = f"adb -s {self.device.serial} shell \"{cmd}\""
|
|
result = subprocess.check_output(full_cmd, shell=True, stderr=subprocess.STDOUT, timeout=30)
|
|
return result.decode('utf-8')
|
|
except subprocess.TimeoutExpired as e:
|
|
error_msg = f"ADB shell_grep command timeout: {full_cmd}"
|
|
self.logger.error(error_msg)
|
|
self.last_error = error_msg
|
|
raise ADBException(error_msg)
|
|
except subprocess.CalledProcessError as e:
|
|
if e.returncode == 1:
|
|
# grep 没有匹配到任何内容，返回空字符串
|
|
return ""
|
|
error_msg = f"ADB shell_grep failed: {cmd}\nOutput: {e.output.decode('utf-8') if e.output else ''}"
|
|
self.logger.error(error_msg)
|
|
self.last_error = error_msg
|
|
raise ADBException(error_msg)
|
|
except Exception as e:
|
|
error_msg = f"ADB shell_grep unexpected error: {e}"
|
|
self.logger.error(error_msg)
|
|
self.last_error = error_msg
|
|
raise ADBException(error_msg)
|
|
|
|
def check_connectivity(self):
|
|
"""
|
|
check if adb is connected
|
|
:return: True for connected
|
|
"""
|
|
r = self.run_cmd("get-state")
|
|
return r.startswith("device")
|
|
|
|
def connect(self):
|
|
"""
|
|
connect adb
|
|
"""
|
|
self.logger.debug("connected")
|
|
|
|
def disconnect(self):
|
|
"""
|
|
disconnect adb
|
|
"""
|
|
print("[CONNECTION] %s is disconnected" % self.__class__.__name__)
|
|
|
|
def get_property(self, property_name):
|
|
"""
|
|
get the value of property
|
|
@param property_name:
|
|
@return:
|
|
"""
|
|
return self.shell(["getprop", property_name])
|
|
|
|
def get_model_number(self):
|
|
"""
|
|
Get device model number. e.g. SM-G935F
|
|
"""
|
|
return self.get_property(ADB.MODEL_PROPERTY)
|
|
|
|
def get_sdk_version(self):
|
|
"""
|
|
Get version of SDK, e.g. 18, 20
|
|
"""
|
|
if self._sdk_version_cache is not None:
|
|
return self._sdk_version_cache
|
|
self._sdk_version_cache = int(self.get_property(ADB.VERSION_SDK_PROPERTY))
|
|
return self._sdk_version_cache
|
|
|
|
def get_release_version(self):
|
|
"""
|
|
Get release version, e.g. 4.3, 6.0
|
|
"""
|
|
return self.get_property(ADB.VERSION_RELEASE_PROPERTY)
|
|
|
|
|
|
# The following methods are originally from androidviewclient project.
|
|
# https://github.com/dtmilano/AndroidViewClient.
|
|
def get_display_info(self, use_cache=True, cache_ttl=2.0):
|
|
"""
|
|
Gets C{mDefaultViewport} and then C{deviceWidth} and C{deviceHeight} values from dumpsys.
|
|
This is a method to obtain display dimensions and density
|
|
@param use_cache: whether to use cached display info
|
|
@param cache_ttl: cache time-to-live in seconds (default 2.0s)
|
|
"""
|
|
current_time = time.time()
|
|
if use_cache and self._display_info_cache is not None:
|
|
if current_time - self._display_info_cache_time < cache_ttl:
|
|
return self._display_info_cache
|
|
|
|
display_info = {}
|
|
logical_display_re = re.compile(".*DisplayViewport{valid=true, .*orientation=(?P<orientation>\d+),"
|
|
" .*deviceWidth=(?P<width>\d+), deviceHeight=(?P<height>\d+).*")
|
|
dumpsys_display_result = self.shell("dumpsys display")
|
|
if dumpsys_display_result is not None:
|
|
for line in dumpsys_display_result.splitlines():
|
|
m = logical_display_re.search(line, 0)
|
|
if m:
|
|
for prop in ['width', 'height', 'orientation']:
|
|
display_info[prop] = int(m.group(prop))
|
|
|
|
if 'width' not in display_info or 'height' not in display_info:
|
|
physical_display_re = re.compile('Physical size: (?P<width>\d+)x(?P<height>\d+)')
|
|
m = physical_display_re.search(self.shell('wm size'))
|
|
if m:
|
|
for prop in ['width', 'height']:
|
|
display_info[prop] = int(m.group(prop))
|
|
|
|
if 'width' not in display_info or 'height' not in display_info:
|
|
# This could also be mSystem or mOverscanScreen
|
|
display_re = re.compile('\s*mUnrestrictedScreen=\((?P<x>\d+),(?P<y>\d+)\) (?P<width>\d+)x(?P<height>\d+)')
|
|
# This is known to work on older versions (i.e. API 10) where mrestrictedScreen is not available
|
|
display_width_height_re = re.compile('\s*DisplayWidth=(?P<width>\d+) *DisplayHeight=(?P<height>\d+)')
|
|
for line in self.shell('dumpsys window').splitlines():
|
|
m = display_re.search(line, 0)
|
|
if not m:
|
|
m = display_width_height_re.search(line, 0)
|
|
if m:
|
|
for prop in ['width', 'height']:
|
|
display_info[prop] = int(m.group(prop))
|
|
|
|
if 'orientation' not in display_info:
|
|
surface_orientation_re = re.compile("SurfaceOrientation:\s+(\d+)")
|
|
output = self.shell("dumpsys input")
|
|
m = surface_orientation_re.search(output)
|
|
if m:
|
|
display_info['orientation'] = int(m.group(1))
|
|
|
|
density = None
|
|
float_re = re.compile(r"[-+]?\d*\.\d+|\d+")
|
|
d = self.get_property('ro.sf.lcd_density')
|
|
if float_re.match(d):
|
|
density = float(d)
|
|
else:
|
|
d = self.get_property('qemu.sf.lcd_density')
|
|
if float_re.match(d):
|
|
density = float(d)
|
|
else:
|
|
physical_density_re = re.compile('Physical density: (?P<density>[\d.]+)', re.MULTILINE)
|
|
m = physical_density_re.search(self.shell('wm density'))
|
|
if m:
|
|
density = float(m.group('density'))
|
|
if density is not None:
|
|
display_info['density'] = density
|
|
|
|
display_info_keys = {'width', 'height', 'orientation', 'density'}
|
|
if not display_info_keys.issuperset(display_info):
|
|
self.logger.warning("getDisplayInfo failed to get: %s" % display_info_keys)
|
|
|
|
self._display_info_cache = display_info
|
|
self._display_info_cache_time = current_time
|
|
return display_info
|
|
|
|
def get_enabled_accessibility_services(self):
|
|
"""
|
|
Get enabled accessibility services
|
|
:return: the enabled service names, each service name is in <package_name>/<service_name> format
|
|
"""
|
|
r = self.shell("settings get secure enabled_accessibility_services")
|
|
r = re.sub(r'(?m)^WARNING:.*\n?', '', r)
|
|
return r.strip().split(":") if r.strip() != '' else []
|
|
|
|
def enable_accessibility_service(self, service_name):
|
|
"""
|
|
Enable an accessibility service
|
|
:param service_name: the service to enable, in <package_name>/<service_name> format
|
|
"""
|
|
service_names = self.get_enabled_accessibility_services()
|
|
if service_name not in service_names:
|
|
service_names.append(service_name)
|
|
self.shell("settings put secure enabled_accessibility_services %s" % ":".join(service_names))
|
|
self.shell("settings put secure accessibility_enabled 1")
|
|
|
|
def enable_accessibility_service_db(self, service_name):
|
|
"""
|
|
Enable an accessibility service
|
|
:param service_name: the service to enable, in <package_name>/<service_name> format
|
|
"""
|
|
subprocess.check_call(
|
|
"adb shell \""
|
|
"sqlite3 -batch /data/data/com.android.providers.settings/databases/settings.db \\\""
|
|
"DELETE FROM secure WHERE name='enabled_accessibility_services' OR name='accessibility_enabled' "
|
|
"OR name='touch_exploration_granted_accessibility_services' OR name='touch_exploration_enabled';"
|
|
"INSERT INTO secure (name, value) VALUES "
|
|
"('enabled_accessibility_services','" + service_name + "'), "
|
|
"('accessibility_enabled','1'), "
|
|
"('touch_exploration_granted_accessibility_services','" + service_name + "'), "
|
|
"('touch_exploration_enabled','1')\\\";\"", shell=True)
|
|
self.shell("stop")
|
|
time.sleep(1)
|
|
self.shell("start")
|
|
|
|
def get_installed_apps(self):
|
|
"""
|
|
Get the package names and apk paths of installed apps on the device
|
|
:return: a dict, each key is a package name of an app and each value is the file path to the apk
|
|
"""
|
|
app_lines = self.shell("pm list packages -f").splitlines()
|
|
app_line_re = re.compile("package:(?P<apk_path>.+)=(?P<package>[^=]+)")
|
|
package_to_path = {}
|
|
for app_line in app_lines:
|
|
m = app_line_re.match(app_line)
|
|
if m:
|
|
package_to_path[m.group('package')] = m.group('apk_path')
|
|
return package_to_path
|
|
|
|
|
|
|
|
def __transform_point_by_orientation(self, xy, orientation_orig, orientation_dest, display_info=None):
|
|
(x, y) = xy
|
|
if orientation_orig != orientation_dest:
|
|
if display_info is None:
|
|
display_info = self.get_display_info()
|
|
if orientation_dest == 1:
|
|
_x = x
|
|
x = display_info['width'] - y
|
|
y = _x
|
|
elif orientation_dest == 3:
|
|
_x = x
|
|
x = y
|
|
y = display_info['height'] - _x
|
|
return x, y
|
|
|
|
def get_orientation(self):
|
|
display_info = self.get_display_info()
|
|
if 'orientation' in display_info:
|
|
return display_info['orientation']
|
|
else:
|
|
return -1
|
|
|
|
def unlock(self):
|
|
"""
|
|
Unlock the screen of the device
|
|
"""
|
|
self.shell("input keyevent MENU")
|
|
self.shell("input keyevent BACK")
|
|
|
|
def press(self, key_code):
|
|
"""
|
|
Press a key
|
|
"""
|
|
self.shell("input keyevent %s" % key_code)
|
|
|
|
def touch(self, x, y, orientation=-1, event_type=DOWN_AND_UP):
|
|
display_info = self.get_display_info()
|
|
current_orientation = display_info.get('orientation', -1)
|
|
if orientation == -1:
|
|
orientation = current_orientation
|
|
self.shell("input tap %d %d" %
|
|
self.__transform_point_by_orientation((x, y), orientation, current_orientation, display_info))
|
|
|
|
def long_touch(self, x, y, duration=2000, orientation=-1):
|
|
"""
|
|
Long touches at (x, y)
|
|
"""
|
|
self.drag((x, y), (x, y), duration, orientation)
|
|
|
|
def drag(self, start_xy, end_xy, duration, orientation=-1):
|
|
"""
|
|
Sends drag event n PX (actually it's using C{input swipe} command.
|
|
@param start_xy: starting point in pixel
|
|
@param end_xy: ending point in pixel
|
|
@param duration: duration of the event in ms
|
|
@param orientation: the orientation (-1: undefined)
|
|
"""
|
|
(x0, y0) = start_xy
|
|
(x1, y1) = end_xy
|
|
display_info = self.get_display_info()
|
|
current_orientation = display_info.get('orientation', -1)
|
|
if orientation == -1:
|
|
orientation = current_orientation
|
|
(x0, y0) = self.__transform_point_by_orientation((x0, y0), orientation, current_orientation, display_info)
|
|
(x1, y1) = self.__transform_point_by_orientation((x1, y1), orientation, current_orientation, display_info)
|
|
|
|
version = self.get_sdk_version()
|
|
if version <= 15:
|
|
self.logger.error("drag: API <= 15 not supported (version=%d)" % version)
|
|
elif version <= 17:
|
|
self.shell("input swipe %d %d %d %d" % (x0, y0, x1, y1))
|
|
else:
|
|
self.shell("input touchscreen swipe %d %d %d %d %d" % (x0, y0, x1, y1, duration))
|
|
|
|
def type(self, text):
|
|
if isinstance(text, str):
|
|
escaped = text.replace("%s", "\\%s")
|
|
encoded = escaped.replace(" ", "%s")
|
|
else:
|
|
encoded = str(text)
|
|
# TODO find out which characters can be dangerous, and handle non-English characters
|
|
self.shell("input text %s" % encoded)
|