Airtest UI 自动化测试工具 Airtest 学习笔记之设备管理

steven · October 16, 2018 · Last by steven replied at June 13, 2019 · 2819 hits

通过本篇你讲了解到Airtest是如何跟安卓设备交互的,以及多设备时的多机交互使用。

在之前从Touch接口分析Airtest的图像识别中,在图像识别获取到目标位置以后,发起点击的操作是通过以下这句:

G.DEVICE.touch(pos, **kwargs)
touch接口.png

看一下有那么多个类里有touch接口,device、minitouch、adb、android、win、linux、ios

另外再翻一下airtest.core.api这个文件里的其他接口


"""
Device Operations
"""


@logwrap
def shell(cmd):
"""
Start remote shell in the target device and execute the command

:param cmd: command to be run on device, e.g. "
ls /data/local/tmp"
:return: the output of the shell cmd
:platforms: Android
"""

return G.DEVICE.shell(cmd)

@logwrap
def start_app(package, activity=None):
"""
Start the target application on device

:param package: name of the package to be started, e.g. "
com.netease.my"
:param activity: the activity to start, default is None which means the main activity
:return: None
:platforms: Android, iOS
"""

G.DEVICE.start_app(package, activity)

可见,这些设备操作的接口都是通过这个G.DEVICE,所以这里就是我们要找的Airtest与各类被测设备交互的实现部分了。

先来看一下这个G.DEVICE是什么


class G(object):
"""Represent the globals variables"""
BASEDIR = []
LOGGER = AirtestLogger(None)
LOGGING = get_logger("airtest.core.api")
SCREEN = None
DEVICE = None
DEVICE_LIST = []
RECENT_CAPTURE = None
RECENT_CAPTURE_PATH = None
CUSTOM_DEVICES = {}

@classmethod
def add_device(cls, dev):
"""
Add device instance in G and set as current device.

Examples:
G.add_device(Android())

Args:
dev: device to init

Returns:
None

"""

cls.DEVICE = dev
cls.DEVICE_LIST.append(dev)

看这个add_device的注释,传入的dev是初始化之后的设备对象,例如安卓,ios等,然后存放在G.DEVICE和添加到G.DEVICE_LIST列表里。既然是初始化,那么想必就是要在脚本的最前面的执行吧,所以Airtest新建脚本时自动生成的那句auto_setup应该就跟设备初始化有关系了,一起去看看。


def auto_setup(basedir=None, devices=None, logdir=None, project_root=None):
"""
Auto setup running env and try connect android device if not device connected.
"""

if devices:
for dev in devices:
connect_device(dev)
elif not G.DEVICE_LIST:
try:
connect_device("Android:///")
except IndexError:
pass
if basedir:
if os.path.isfile(basedir):
basedir = os.path.dirname(basedir)
if basedir not in G.BASEDIR:
G.BASEDIR.append(basedir)
if logdir:
set_logdir(logdir)
if project_root:
ST.PROJECT_ROOT = project_root

def connect_device(uri):
"""
Initialize device with uri, and set as current device.

:param uri: an URI where to connect to device, e.g. `android://adbhost:adbport/serialno?param=value&param2=value2`
:return: device instance
:Example:
* ``android:///`` # local adb device using default params
* ``android://adbhost:adbport/1234566?cap_method=javacap&touch_method=adb`` # remote device using custom params
* ``windows:///`` # local Windows application
* ``ios:///`` # iOS device
"""

d = urlparse(uri)
platform = d.scheme
host = d.netloc
uuid = d.path.lstrip("/")
params = dict(parse_qsl(d.query))
if host:
params["host"] = host.split(":")
dev = init_device(platform, uuid, **params)
return dev

def init_device(platform="Android", uuid=None, **kwargs):
"""
Initialize device if not yet, and set as current device.

:param platform: Android, IOS or Windows
:param uuid: uuid for target device, e.g. serialno for Android, handle for Windows, uuid for iOS
:param kwargs: Optional platform specific keyword args, e.g. `cap_method=JAVACAP` for Android
:return: device instance
"""

cls = import_device_cls(platform)
dev = cls(uuid, **kwargs)
for index, instance in enumerate(G.DEVICE_LIST):
if dev.uuid == instance.uuid:
G.LOGGING.warn("Device:%s updated %s -> %s" % (dev.uuid, instance, dev))
G.DEVICE_LIST[index] = dev
break
else:
G.add_device(dev)
return dev

def import_device_cls(platform):
"""lazy import device class"""
platform = platform.lower()
if platform in G.CUSTOM_DEVICES:
cls = G.CUSTOM_DEVICES[platform]
elif platform == "android":
from airtest.core.android.android import Android as cls
elif platform == "windows":
from airtest.core.win.win import Windows as cls
elif platform == "ios":
from airtest.core.ios.ios import IOS as cls
elif platform == "linux":
from airtest.core.linux.linux import Linux as cls
else:
raise RuntimeError("Unknown platform: %s" % platform)
return cls

由上到下的调用关系:auto_setup -> connect_device -> init_device -> add_device

auto_setup接口:依次连接全部设备,处理日志,工程根目录等事物

connect_device接口:根据传入参数uri的解析出其平台和序列号信息,然后初始化设备

init_device接口:调用import_device_cls导入不同的平台,初始化设备对象,如果DEVICE_LIST列表里没有该设备,则添加设备

add_device接口:将新连接上的设备赋值给G.DEVICE,添加到G.DEVICE_LIST

所以在Airtest教程中的“4.3 多机协作脚本”讲到:

在我们的脚本中,支持通过set_current接口来切换当前连接的手机,因此我们一个脚本中,是能够调用多台手机,编写出一些复杂的多机交互脚本的。

在命令行运行脚本时,只需要将手机依次使用--device Android:///添加到命令行中即可,例如:

airtest run untitled.air --device Android:///serialno1 --device Android:///serialno2 --device

在之前的笔记里分析过run_script接口解析命令行参数中的device会生成成一个设备列表,传入到auto_setup里就会遍历列表逐个去连接,所以多设备交互的操作是:

1.初始化连接所有的设备——命令行或者是调用run_script传入多个设备,当然也可以直接调用connect_device、add_device;

2.调用set_current来切换当前操作的设备。

set_current接口很简单了,在G.DEVICE_LIST里找出目标设备,赋值给G.DEVICE,因为对设备的操作都是通过G.DEVICE的,所以只要换掉G.DEVICE就完成了设备的切换。看下源码:


def set_current(idx):
"""
Set current active device.

:param idx: uuid or index of initialized device instance
:raise IndexError: raised when device idx is not found
:return: None
:platforms: Android, iOS, Windows
"""


dev_dict = {dev.uuid: dev for dev in G.DEVICE_LIST}
if idx in dev_dict:
current_dev = dev_dict[idx]
elif isinstance(idx, int) and idx < len(G.DEVICE_LIST):
current_dev = G.DEVICE_LIST[idx]
else:
raise IndexError("device idx not found in: %s or %s" % (
list(dev_dict.keys()), list(range(len(G.DEVICE_LIST)))))
G.DEVICE = current_dev

关于Airtest的设备管理的分析大概就是以上这些了,多设备的交互很简单,不用在具体的操作方法中指定设备,而是只用在中间调用set_current来完成切换设备,例如切换前是A设备,那么所有的操作都会指向A设备,切换后则都指向B设备,这种设计也挺省事的。

接下来再拿android这部分来看一下airtest是怎么跟设备交互的。

从import_device_cls接口里找进去

'elif platform == "android": from airtest.core.android.android import Android as cls'

android平台的设备管理在airtest.core.android.android的Android类里


class Android(Device):
"""Android Device Class"""

def __init__(self, serialno=None, host=None,
cap_method=CAP_METHOD.MINICAP_STREAM,
touch_method=TOUCH_METHOD.MINITOUCH,
ime_method=IME_METHOD.YOSEMITEIME,
ori_method=ORI_METHOD.MINICAP,
):
super(Android, self).__init__()
self.serialno = serialno or self.get_default_device()
self.cap_method = cap_method.upper()
self.touch_method = touch_method.upper()
self.ime_method = ime_method.upper()
self.ori_method = ori_method.upper()
# init adb
self.adb = ADB(self.serialno, server_addr=host)
self.adb.wait_for_device()
self.sdk_version = self.adb.sdk_version
self._display_info = {}
self._current_orientation = None
# init components
self.rotation_watcher = RotationWatcher(self.adb)
self.minicap = Minicap(self.adb, ori_function=self.get_display_info)
self.javacap = Javacap(self.adb)
self.minitouch = Minitouch(self.adb, ori_function=self.get_display_info)
self.yosemite_ime = YosemiteIme(self.adb)
self.recorder = Recorder(self.adb)
self._register_rotation_watcher()

Android是安卓设备类,父类是Device,这是一个基类,只定义了设备通用接口。android设备初始化,初始化adb,初始化minicap、javacap、minitouch、yosemite、recorder等组件。

翻一下Android类的接口,全都是对安卓设备的操作,基本的一些操作是通过adb完成的,比如:启动应用,卸载应用,唤醒...


def start_app(self, package, activity=None):
"""
Start the application and activity

Args:
package: package name
activity: activity name

Returns:
None

"""

return self.adb.start_app(package, activity)

def unlock(self):
"""
Unlock the device

Notes:
Might not work on all devices

Returns:
None

"""

return self.adb.unlock()

还有就是用到了其他组件的操作了,比如截图用到minicap和javacap组件,截图有四种方式:minicap_stream、minicap、javacap、adb_snapshot,初始化传入参数可配置截图的方式,默认是MINICAP_STREAM,截图之后就是写入,转换成cv2的格式,处理横竖屏的转换。


def snapshot(self, filename=None, ensure_orientation=True):
"""
Take the screenshot of the display. The output is send to stdout by default.

Args:
filename: name of the file where to store the screenshot, default is None which si stdout
ensure_orientation: True or False whether to keep the orientation same as display

Returns:
screenshot output

"""

"""default not write into file."""
if self.cap_method == CAP_METHOD.MINICAP_STREAM:
self.rotation_watcher.get_ready()
screen = self.minicap.get_frame_from_stream()
elif self.cap_method == CAP_METHOD.MINICAP:
screen = self.minicap.get_frame()
elif self.cap_method == CAP_METHOD.JAVACAP:
screen = self.javacap.get_frame_from_stream()
else:
screen = self.adb.snapshot()
# output cv2 object
try:
screen = aircv.utils.string_2_img(screen)
except Exception:
# may be black/locked screen or other reason, print exc for debugging
import traceback
traceback.print_exc()
return None

# ensure the orientation is right
if ensure_orientation and self.display_info["orientation"]:
# minicap screenshots are different for various sdk_version
if self.cap_method in (CAP_METHOD.MINICAP, CAP_METHOD.MINICAP_STREAM) and self.sdk_version <= 16:
h, w = screen.shape[:2] # cvshape是高度在前面!!!!
if w < h: # 当前是横屏,但是图片是竖的,则旋转,针对sdk<=16的机器
screen = aircv.rotate(screen, self.display_info["orientation"] * 90, clockwise=False)
# adb 截图总是要根据orientation旋转
elif self.cap_method == CAP_METHOD.ADBCAP:
screen = aircv.rotate(screen, self.display_info["orientation"] * 90, clockwise=False)
if filename:
aircv.imwrite(filename, screen)
return screen

输入字符用到yosemite输入法,在yosemite初始化时会往安卓设备中安装一个叫yosemite的输入法app,并通过adb命令将设备的当前输入法切换成yosemite,yosemite输入法app有个广播接收器,接收到广播后输入字符。

self.yosemite_ime = YosemiteIme(self.adb)


class YosemiteIme(CustomIme):
"""
Yosemite Input Method Class Object
"""


def __init__(self, adb):
super(YosemiteIme, self).__init__(adb, None, YOSEMITE_IME_SERVICE)
self.yosemite = Yosemite(adb)

def start(self):
self.yosemite.get_ready()
super(YosemiteIme, self).start()

def text(self, value):
"""
Input text with Yosemite input method

Args:
value: text to be inputted

Returns:
output form `adb shell` command

"""

if not self.started:
self.start()
# 更多的输入用法请见 https://github.com/macacajs/android-unicode#use-in-adb-shell
value = ensure_unicode(value)
self.adb.shell(u"am broadcast -a ADB_INPUT_TEXT --es msg '{}'".format(value))


def start(self):
"""
Enable input method

Returns:
None

"""

try:
self.default_ime = self.adb.shell("settings get secure default_input_method").strip()
except AdbError:
# settings cmd not found for older phones, e.g. Xiaomi 2A
# /system/bin/sh: settings: not found
self.default_ime = None
self.ime_list = self._get_ime_list()
if self.service_name not in self.ime_list:
if self.apk_path:
self.device.install_app(self.apk_path)
if self.default_ime != self.service_name:
self.adb.shell("ime enable %s" % self.service_name)
self.adb.shell("ime set %s" % self.service_name)
self.started = True

所以输入字符的接口也有两种方式:yosemite输入法和adb命令,默认是yosemite输入


def text(self, text, enter=True):
"""
Input text on the device

Args:
text: text to input
enter: True or False whether to press `Enter` key

Returns:
None

"""

if self.ime_method == IME_METHOD.YOSEMITEIME:
self.yosemite_ime.text(text)
else:
self.adb.shell(["input", "text", text])

# 游戏输入时,输入有效内容后点击Enter确认,如不需要,enter置为False即可。
if enter:
self.adb.shell(["input", "keyevent", "ENTER"])

录屏用到recorder组件,录屏是用yosemite这个app实现的,pythod这边只是发adb命令,简单的看一下start_record这部分吧,


源码位置:airtest/core/android/android.py

def start_recording(self, *args, **kwargs):
"""
Start recording the device display

Args:
*args: optional arguments
**kwargs: optional arguments

Returns:
None

"""

return self.recorder.start_recording(*args, **kwargs)


源码位置:airtest/core/android/recorder.py

@on_method_ready('install_or_upgrade')
def start_recording(self, max_time=1800, bit_rate=None, vertical=None):
"""
Start screen recording

Args:
max_time: maximum rate value, default is 1800
bit_rate: bit rate value, default is None
vertical: vertical parameters, default is None

Raises:
RuntimeError: if any error occurs while setup the recording

Returns:
None if recording did not start, otherwise True

"""

if getattr(self, "recording_proc", None):
raise AirtestError("recording_proc has already started")
pkg_path = self.adb.path_app(YOSEMITE_PACKAGE)
max_time_param = "-Dduration=%d" % max_time if max_time else ""
bit_rate_param = "-Dbitrate=%d" % bit_rate if bit_rate else ""
if vertical is None:
vertical_param = ""
else:
vertical_param = "-Dvertical=true" if vertical else "-Dvertical=false"
p = self.adb.start_shell('CLASSPATH=%s exec app_process %s %s %s /system/bin %s.Recorder --start-record' %
(pkg_path, max_time_param, bit_rate_param, vertical_param, YOSEMITE_PACKAGE))
nbsp = NonBlockingStreamReader(p.stdout)
while True:
line = nbsp.readline(timeout=5)
if line is None:
raise RuntimeError("start recording error")
if six.PY3:
line = line.decode("utf-8")
m = re.match("start result: Record start success! File path:(.*\.mp4)", line.strip())
if m:
output = m.group(1)
self.recording_proc = p
self.recording_file = output
return True

点击、滑动等用到minitouch组件,同样的可选minitouch或者是adb


def touch(self, pos, duration=0.01):
"""
Perform touch event on the device

Args:
pos: coordinates (x, y)
duration: how long to touch the screen

Returns:
None

"""

if self.touch_method == TOUCH_METHOD.MINITOUCH:
pos = self._touch_point_by_orientation(pos)
self.minitouch.touch(pos, duration=duration)
else:
self.adb.touch(pos)

minitouch、minicap有啥不同呢,这是openstf的库,大概是在安卓设备下放了一个client,pythod这边用safesocket发消息给client,由client执行操作,详细的先不在这里分析了。

android设备类大致就是这样了,再往下可以看看adb类,这个就只看看发命令的核心接口吧。

def start_cmd(self, cmds, device=True):
"""
Start a subprocess with adb command(s)

Args:
cmds: command(s) to be run
device: if True, the device serial number must be specified by `-s serialno` argument

Raises:
RuntimeError: if `device` is True and serialno is not specified

Returns:
a subprocess

"""

if device:
if not self.serialno:
raise RuntimeError("please set serialno first")
cmd_options = self.cmd_options + ['-s', self.serialno]
else:
cmd_options = self.cmd_options

cmds = cmd_options + split_cmd(cmds)
LOGGING.debug(" ".join(cmds))

if not PY3:
cmds = [c.encode(get_std_encoding(sys.stdin)) for c in cmds]

proc = subprocess.Popen(
cmds,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
return proc

总结,Airtest的设备管理只是用G.DEVICE指向当前设备,用G.DEVICE_LIST保存全部设备,所有的操作都通过G.DEVICE转发,所以改变G.DEVICE即可切换设备。而安卓设备的交互则是通过adb命令,和一些别的库:yosemete、minitouch、minicap、javacap。

共收到 2 条回复 时间 点赞

你好请问一下 yosemite如何在初始化的时候给去掉 目前卡着我的自动化测试脚本了

2Floor has been deleted
steven #3 · June 13, 2019 作者
黎斌 回复

翻源码啊,你从connect_device 找进去,Android类的初始化可以配置输入法的

class Android(Device):
"""Android Device Class"""

def __init__(self, serialno=None, host=None,
cap_method=CAP_METHOD.MINICAP_STREAM,
touch_method=TOUCH_METHOD.MINITOUCH,
ime_method=IME_METHOD.YOSEMITEIME,
ori_method=ORI_METHOD.MINICAP,
):

ime_method = IME_METHOD.ADBTOUCH

需要 Sign In 后方可回复, 如果你还没有账号请点击这里 Sign Up