最近打算梳理一下不同产品领域的自动化测试实现方案，如：Android终端、Web、服务端、智能硬件等，就先从Android终端产品开始梳理吧。本文主要介绍UI自动化测试的实现，因为这类测试解决方案比较通用，Android系统层、内核层的自动化测试解决方案可能要根据公司的具体业务来定了。
目前已经有比较不错的一些开源工具，如：Appnium、Python-UiAutomator，功能基本差不多，实现原理也是一样的，这里不做过多描述，有兴趣的同学可以查阅相关资料学习。
既然已经有开源的工具了，为什么还要自研？
1、根据公司业务需要，满足定制化需求
2、结合公司自动化框架，打造更加归一化、集成化的平台
3、降低维护成本、提升易用性
我们先讲一下PC怎么跟Android终端通信，进而实现对它的控制，方式有两种：有线方式和无线方式。
有线方式：PC跟终端通过USB线连接，这种方式其实是通过adb作为中间媒介来实现通信的，如下图所示：
上图提到了Adb Client、Adb Server、Adb Daemon三个关键要素，下面简单讲一下三者的作用：
Adb Client：这部分就是需要我们自研开发的socket客户端
Adb Server：其实就是通过adb start-server命令在PC上启动的一个socket服务
Adb Daemon：是在终端上后台运行的一个守护进程，开机自启动，而且即使被杀掉，系统也会重新启动该进程，主要作用是跟Adb Server进行连接通信
注意：Adb Client与Adb Server通信遵循adb协议，具体协议内容可以百度查一下，相关资料很多
通过以上内容大家应该清楚了PC怎么连接终端并且跟终端进行通信，那么重点来了，想要自研UI自动化测试工具就必须开发一个应用软件实现对uiautomator的封装，并在终端上作为Server运行起来，那么Adb Client跟我们开发的终端应用软件是如何进行通信的呢？如下图所示：
本质上就是通过adb forward进行端口转发，关于这部分的内容不在这里做重点介绍，大家可以去查询adb协议。
无线方式：PC跟终端都连接到了同一个网络，通过网络进行通信，如下图所示：
可以看出，相比有线方式，无线方式更加简单直接，PC作为客户端、应用软件作为Server端，二者通过建立socket通信来实现数据交互，从而实现PC对终端的控制。
总结：不管哪种通信方式，要实现自研工具，有两块要进行开发（PC侧的客户端和终端侧的应用软件）。
Adb Client代码示例：
# -*- coding:utf-8 -*-
import os
import subprocess
import socket
import whichcraft
from collections import namedtuple
from components.device.android.adb.errors import AdbError
_OKAY = "OKAY"
_FAIL = "FAIL"
_DENT = "DENT" # Directory Entity
_DONE = "DONE"
ForwardItem = namedtuple("ForwardItem", ["serial", "local", "remote"])
def where_adb():
adb_path = whichcraft.which('adb')
if adb_path is None:
raise EnvironmentError("Can't find adb,please install adb first.")
return adb_path
class _AdbStreamConnect(object):
"""
连接adb服务
即通过adb start-server在PC侧ANDROID_ADB_SERVER_PORT端口起的服务
具体可以发送哪些命令需要查看adb协议
"""
def __init__(self, host=None, port=None):
self.__host = host
self.__port = port
self.__conn = None
self._connect()
def _create_socket(self):
adb_host = self.__host or os.environ.get('ANDROID_ADB_SERVER_HOST', '127.0.0.1')
adb_port = self.__port or int(os.environ.get('ANDROID_ADB_SERVER_PORT', 7305))
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5)
try:
s.connect((adb_host, adb_port))
return s
except:
s.close()
raise
@property
def conn(self):
return self.__conn
def _connect(self):
try:
self.__conn = self._create_socket()
except Exception as e:
subprocess.Popen('adb kill-server')
subprocess.Popen('adb start-server')
self.__conn = self._create_socket()
def close(self):
self.conn.close()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
def send(self, cmd):
# cmd: str表示期望cmd是字符串类型
if not isinstance(cmd, str):
cmd = str(cmd)
self.conn.send("{:04x}{}".format(len(cmd), cmd).encode("utf-8"))
def read(self, n):
# -> str表示接口返回值为字符串
return self.conn.recv(n).decode()
def read_string(self):
size = int(self.read(4), 16)
return self.read(size)
def read_until_close(self):
content = ""
while True:
chunk = self.read(4096)
if not chunk:
break
content += chunk
return content
def check_okay(self):
# 前四位是状态码
data = self.read(4)
if data == _FAIL:
raise AdbError(self.read_string())
elif data == _OKAY:
return
raise AdbError("Unknown data: %s" % data)
class AdbClient(object):
def __init__(self, host=None, port=None):
self.__host = host
self.__port = port
def server_version(self):
"""
@summary:获取adb版本号
@return :版本号1.0.41中的41
"""
with self._connect() as c:
c.send("host:version")
c.check_okay()
return int(c.read_string(), 16)
def _connect(self):
return _AdbStreamConnect(self.__host, self.__port)
def forward(self, serial, local, remote, norebind=False):
"""
@summary:给adb服务端发送host-serial:<sn>:forward:tcp:<pc_port>;tcp:<phone_port>进行端口转发
@param serial:手机sn号,sn为None时按默认连接一部手机处理
@param local:PC侧socket客户端端口
@param remote:手机侧socket服务端端口
@param norebind:fail if already forwarded when set to true
@attention :PC跟手机通过USB方式通信
"""
with self._connect() as c:
cmds = ["host", "forward"]
if serial:
cmds = ["host-serial", serial, "forward"]
if norebind:
cmds.append("norebind")
cmds.append("tcp:%s;tcp:%s" % (local, remote))
print(cmds)
c.send(":".join(cmds))
c.check_okay()
def forward_list(self, serial=None):
"""
@summary:查看端口转发是否成功
@param serial:手机sn号
@attention :PC跟手机通过USB方式通信
"""
with self._connect() as c:
list_cmd = "host:list-forward"
if serial:
list_cmd = "host-serial:{}:list-forward".format(serial)
c.send(list_cmd)
c.check_okay()
content = c.read_string()
for line in content.splitlines():
parts = line.split()
if len(parts) != 3:
continue
if serial and parts[0] != serial:
continue
yield ForwardItem(*parts)
def shell(self, serial, command):
"""
@summary:执行shell命令
@param serial:手机sn号
@param command:要执行的命令
@attention :只能执行adb shell命令
"""
with self._connect() as c:
c.send("host:transport:" + serial)
c.check_okay()
c.send("shell:" + command)
c.check_okay()
return c.read_until_close()
def device_list(self):
"""
@summary:获取手机列表
@attention :
"""
device_list = []
with self._connect() as c:
c.send("host:devices")
c.check_okay()
output = c.read_string()
for line in output.splitlines():
parts = line.strip().split("\t")
if len(parts) != 2:
continue
if parts[1] == 'device':
device_list.append(parts[0])
return device_list
def must_one_device(self, serial):
device_list = self.device_list()
if len(device_list) == 0:
raise RuntimeError(
"Can't find any android device/emulator"
)
elif serial is None and len(device_list) > 1:
raise RuntimeError(
"more than one device/emulator, please specify the serial number"
)uiautomator.py代码实现：
# -*- coding: utf-8 -*-
# @Time : 2023/2/5 20:24
# @Author : 十年
# @Site : https://gitee.com/chshao/aiplotest
# @CSDN : https://blog.csdn.net/m0_37576542?type=blog
# @File : uiautomator.py
# @Description : 对这个模块文件的总体描述
import re
import time
import json
import socket
from typing import Union
from aitest.buildin.AiDecorator import singleton
from aitest.components.android.adb import AdbClient
TAG_UI = "UiAutomator"
TAG_WIFI = "WifiManager"
TAG_STUB = "Stub"
PORT10086 = 10086
PORT12306 = 12306
ip_pattern = re.compile('\d+.\d+.\d+.\d+')
@singleton
class UiClient(object):
gSocket = None
def connect(self, addr: Union[None, str, tuple]):
"""
@summary:PC连接手机
@param addr:手机sn号或者(ip,port)
@attention :
"""
if isinstance(addr, tuple):
self._connect_wifi(addr[0], addr[1])
self._connect_usb(addr)
def _connect_usb(self, serial):
adb_client = AdbClient()
adb_client.must_one_device(serial)
self._check_and_forward(adb_client, serial)
def _check_and_forward(self, adb_client: AdbClient, serial=None):
forward_list = adb_client.forward_list(serial)
for forwardItem in forward_list:
if forwardItem.serial == serial:
port = forwardItem.local.split(':')[-1]
self._connect_wifi('127.0.0.1', port)
# 通过adb在pc上创建一个server监听10086端口，并将10086端口收到的数据转发给手机server的12306端口
adb_client.forward(serial, PORT10086, PORT12306)
# 创建一个socket客户端连接10086端口
self._connect_wifi('127.0.0.1', PORT10086)
def _connect_wifi(self, host, port):
self.gSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.gSocket.connect((host, port))
self.gSocket.settimeout(2)
def send_cmd(self, cmd):
cmd_bytes = json.dumps(cmd).encode('utf-8')
cmd_len = len(cmd_bytes).to_bytes(4, "little")
self.gSocket.send(cmd_len + cmd_bytes)
def recv_ack(self, timeout=10):
endTime = time.time() + timeout
while time.time() < endTime:
# 取前4位包头，为数据内容长度
dataLenBytes = self.gSocket.recv(4)
if dataLenBytes != b"":
dataLen = int.from_bytes(dataLenBytes, "little")
# 取出数据内容并返回
return self.gSocket.recvData(dataLen)
return b""
def close(self):
if self.gSocket:
self.gSocket.close()
class Device(object):
def __init__(self, addr=None):
"""
@param addr: None/手机sn/IP
"""
self.ui = UiClient()
if addr is not None:
if re.match(ip_pattern, addr):
addr = (addr, PORT12306)
self.ui.connect(addr)
def command(self, clsName, method, args=None):
cmd = {"class": clsName,
"method": method,
"args": args,
"requestId": "request_" + str(round(time.time(), 3))}
return cmd
def executeCmd(self, clsName, method, args=None):
self.ui.send_cmd(self.command(clsName, method, args))
ack = self.ui.recv_ack()
if ack:
return json.loads(ack.decode('utf-8')).get("result")
return ""
def __checkResult(self, ack, expect):
if ack == expect:
return True
return False
def click(self, x, y):
args = [{"k": "int", "v": x}, {"k": "int", "v": y}]
ack = self.executeCmd(TAG_UI, "click", args)
return self.__checkResult(ack, 'true')
def clickById(self, id):
args = [{"k": "string", "v": id}]
ack = self.executeCmd(TAG_UI, "clickById", args)
return self.__checkResult(ack, 'true')
def clickByText(self, text):
args = [{"k": "string", "v": text}]
ack = self.executeCmd(TAG_UI, "clickByText", args)
return self.__checkResult(ack, 'true')
def clickByImage(self, image):
pass
def isScreenOn(self):
ack = self.executeCmd(TAG_UI, "isScreenOn", [])
if ack == "":
return "unknown"
return self.__checkResult(ack, 'true')
def screenOn(self):
self.executeCmd(TAG_UI, "wakeUp", [])
return self.isScreenOn()
def screenOff(self):
self.executeCmd(TAG_UI, "sleep", [])
return not self.isScreenOn()
def goBack(self):
ack = self.executeCmd(TAG_UI, "pressBack", [])
return self.__checkResult(ack, 'true')
def goHome(self):
ack = self.executeCmd(TAG_UI, "pressHome", [])
return self.__checkResult(ack, 'true')
def getTextById(self, id):
args = [{"k": "string", "v": id}]
return self.executeCmd(TAG_UI, "getTextById", args)
首先，打开Androdi Studio新建一个Android工程，然后在app文件夹下的build.gradle文件添加依赖库，如下图：
androidTestImplementation 'androidx.test:runner:1.3.0'
androidTestImplementation 'androidx.test:rules:1.3.0'
androidTestImplementation 'androidx.test.uiautomator:uiautomator:2.2.0'
androidTestImplementation 'androidx.core:core:1.3.0'
androidTestImplementation 'androidx.annotation:annotation:1.1.0'AndroidManifest.xml中添加如下权限：
<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.DISABLE_KEYGUARD" />
<uses-permission android:name="android.permission.READ_PHONE_STATE" />
<uses-permission android:name="android.permission.WAKE_LOCK" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<uses-permission android:name="android.permission.GET_ACCOUNTS" />
<uses-permission android:name="android.permission.MANAGE_ACCOUNTS" />
<uses-permission android:name="android.permission.CHANGE_WIFI_STATE" />
<uses-permission android:name="android.permission.ACCESS_WIFI_STATE" />
<uses-permission android:name="android.permission.FOREGROUND_SERVICE" />
<uses-permission android:name="android.permission.SYSTEM_ALERT_WINDOW" />
<uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE"/>
<uses-permission android:name="android.permission.READ_EXTERNAL_STORAGE" />
<uses-permission android:name="android.permission.MANAGE_EXTERNAL_STORAGE"
tools:ignore="ScopedStorage" />开发完成后，需要打包编译两个APP：app-debug.apk、app-debug-androidTest.apk
app-debug.apk为待测apk，实际上没什么作用，但是需要安装，只要保证能正常运行就行
app-debug-androidTest.apk为测试apk，uiautomator的封装实现就在这个apk里面
启动测试服务：adb shell am instrument -w -r -e debug false -e class com.aiplot.wetest.Stub com.aiplot.wetest.test/androidx.test.runner.AndroidJUnitRunner
adb install命令参数如下
-t 允许测试包
-l 锁定该应用程序
-s 把应用程序安装到sd卡上
-g 为应用程序授予所有运行时的权限
-r 替换已存在的应用程序，也就是说强制安装
-d 允许进行将见状，也就是安装的比手机上带的版本低