本文主要是介绍UI自动化测试工具Airtest学习笔记之设备管理，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
> 通过本篇你讲了解到Airtest是如何跟安卓设备交互的，以及多设备时的多机交互使用。
在之前从Touch接口分析Airtest的图像识别中，在图像识别获取到目标位置以后，发起点击的操作是通过以下这句：
`G.DEVICE.touch(pos, **kwargs)`
看一下有那么多个类里有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 = NoneDEVICE = NoneDEVICE_LIST = []RECENT_CAPTURE = NoneRECENT_CAPTURE_PATH = NoneCUSTOM_DEVICES = {}@classmethoddef add_device(cls, dev):"""Add device instance in G and set as current device.Examples:G.add_device(Android())Args:dev: device to initReturns:None"""cls.DEVICE = devcls.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:passif 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_rootdef 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¶m2=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.schemehost = d.netlocuuid = d.path.lstrip("/")params = dict(parse_qsl(d.query))if host:params["host"] = host.split(":")dev = init_device(platform, uuid, **params)return devdef 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] = devbreakelse:G.add_device(dev)return devdef 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 clselif platform == "windows":from airtest.core.win.win import Windows as clselif platform == "ios":from airtest.core.ios.ios import IOS as clselif platform == "linux":from airtest.core.linux.linux import Linux as clselse: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 adbself.adb = ADB(self.serialno, server_addr=host)self.adb.wait_for_device()self.sdk_version = self.adb.sdk_versionself._display_info = {}self._current_orientation = None# init componentsself.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 activityArgs:package: package nameactivity: activity nameReturns:None"""return self.adb.start_app(package, activity)def unlock(self):"""Unlock the deviceNotes:Might not work on all devicesReturns: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 stdoutensure_orientation: True or False whether to keep the orientation same as displayReturns: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 objecttry:screen = aircv.utils.string_2_img(screen)except Exception:# may be black/locked screen or other reason, print exc for debuggingimport tracebacktraceback.print_exc()return None# ensure the orientation is rightif ensure_orientation and self.display_info["orientation"]:# minicap screenshots are different for various sdk_versionif 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 methodArgs:value: text to be inputtedReturns:output form `adb shell` command"""if not self.started:self.start()# 更多的输入用法请见 https://github.com/macacajs/android-unicode#use-in-adb-shellvalue = ensure_unicode(value)self.adb.shell(u"am broadcast -a ADB_INPUT_TEXT --es msg '{}'".format(value))```
```
def start(self):"""Enable input methodReturns: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 foundself.default_ime = Noneself.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 deviceArgs:text: text to inputenter: True or False whether to press `Enter` keyReturns: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 displayArgs:*args: optional arguments**kwargs: optional argumentsReturns: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 recordingArgs:max_time: maximum rate value, default is 1800bit_rate: bit rate value, default is Nonevertical: vertical parameters, default is NoneRaises:RuntimeError: if any error occurs while setup the recordingReturns: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 = pself.recording_file = outputreturn True```
点击、滑动等用到minitouch组件，同样的可选minitouch或者是adb
```
def touch(self, pos, duration=0.01):"""Perform touch event on the deviceArgs:pos: coordinates (x, y)duration: how long to touch the screenReturns: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 rundevice: if True, the device serial number must be specified by `-s serialno` argumentRaises:RuntimeError: if `device` is True and serialno is not specifiedReturns: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_optionscmds = 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。
这篇关于UI自动化测试工具Airtest学习笔记之设备管理的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
