Android自动化测试框架实现_安卓自动化测试框架

背景介绍

        最近打算梳理一下不同产品领域的自动化测试实现方案，如：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代码示例

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)

三、应用软件（uiautomator封装）

        首先，打开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　允许进行将见状，也就是安装的比手机上带的版本低 

Android源码地址

链接：http://aospxref.com/