Nico，从零开始干掉Appium，移动端自动化测试框架实现

开头先让我碎碎念一波~去年差不多时间发布了一篇《 UiAutomator Nico，一个基于纯 adb 命令实现的安卓自动化测试框》（https://testerhome.com/topics/37042）， 由于种种原因 (详见此篇帖子) 当时选择了用纯 adb 命令来实现安卓自动化，这个版本我且称为 Nico1.0 版本。

但在实际使用一段时候后确实遇到了各种问题，首当其冲的就是 dump 元素慢，即使强制不等待元素页面加载完成，整体速度更是一言难尽。其次是获取元素树不够完整，除了稳定似乎已经没有别的优点了。特别是在跑多条测试用例的时候，时间成本是肉眼可见的直线上升。用 airtest 大概是 3 秒左右，而如果用 adb dump 的方式则会来到 30 秒。

这时候一个问题摆在我面前，是要选择切回到 Airtest 还是继续使用？切回原来的测试方案这肯定不是我的行事风格，既然选择了一条路走到黑，那不管怎么样都得将眼前的难题攻克。

之前读过另一位大佬的系列文章 《【Day 0】关于我是如何干掉 Appium 和 RobotFramework 这件事的——开篇》（https://vancheung.github.io/posts/Day-0-%E5%85%B3%E4%BA%8E%E6%88%91%E6%98%AF%E5%A6%82%E4%BD%95%E5%B9%B2%E6%8E%89Appium%E5%92%8CRobotFramework%E8%BF%99%E4%BB%B6%E4%BA%8B%E7%9A%84-%E5%BC%80%E7%AF%87/）（非常推荐大家去看完，我相信对想要深耕自动化的同学会有很大帮助），大佬的文章给了我很帮助，在此隔空对这位大佬表示感谢！有了大佬玉珠在前，我便开始了我自己的对从零开始干掉 Appium 的攻略历程，这个版本我称它为 Nico2.0。

接下来的文章将会从整体思路，Android 部分, iOS 部分开始，对技术原理，关键代码讲解等方面进行深度剖析，争取做到小白也能轻松理解。

知识预备

在介绍整个项目之前，我先介绍一下目前所有移动端自动化操作实现的基本原理（对自动化已有基本概念的同学可以跳过这段）。无论是安卓还是 iOS，当我们写脚本执行任意操作，譬如查找元素，操作元素的的时候，其实真正在工作的是系统自带的自动化测试框架 UIAutomator 和 XCUITest。它们是安卓和 iOS 提供给开发者对自己的 App 应用进行 UI 自测的一个测试套件。即开发者可以在写单元测试自测的时候，调用它们来做一些简单的 UI 界面的验证，而它们底层其实都调用了各自的 Accessibility API 。

那么什么 Accessibility API？Accessibility API 是一组用于帮助开发人员创建可以被所有用户访问和使用的应用程序的接口和工具。它最初是为帮助有视力、听力或其他障碍的用户访问和使用计算机和移动设备而设计的。当我们打开手机中的辅助功能-> 旁白之后，我们点击屏幕的任意位置，它都会出现一个白框并且发出语音告诉当前点击的是什么按钮。

为实现这个功能 Accessibility 则需要能够遍历整个用户界面，获取其所有控件及其属性 (包括坐标，id，文本等)，而这正是自动化测试所需要的能力。因此安卓和 iOS 使用这个接口提供了 UIAutomator 和 XCUITest。回到 UIAutomator 和 XCUITest 本身，作为自动化测试套件对于开发者来说是非常友好的，因为它们通常伴随着开发工具一同集成在 IDE 中，编写脚本的语言也跟编写 app 的语言保持一致。但坏处是它无法脱离项目本身独立运行，同时也要求测试人员要能够掌握对应的开发语言，这明显是不现实的。

为了解决这一难题，早期的大神们便想出了一个迂回策略，借助一个中间商！也就是在移动设备上安装一个独立的应用作为服务端，通过这个应用实现调用系统自带的自动化测试套件的能力，同时持续监听指定端口的接收到的请求，然后再转而在设备上进行指定操作。

这里拿 appium 的工作流程图举例：

以下是一段摘自 Chat GTP 的解析：

各组件的功能与交互方式
Appium Server：作为 HTTP 服务器，监听客户端请求，解析命令，并与移动设备上的 Bootstrap 进程进行通信。
Bootstrap.jar：在 Android 设备上运行的 Java 守护进程，负责接收 Appium Server 的命令，转发给 UIAutomator，并返回执行结果。
UIAutomator：Android 系统自带的测试框架，用于执行实际的自动化测试命令。
ADB：Android Debug Bridge，用于在 PC 与 Android 设备之间建立连接，并允许用户发送各种命令到设备上。
客户端：编写测试脚本的开发者使用的工具或库，通过 HTTP 请求与 Appium Server 交互。

至此，我们可以知道整个移动端自动化框架大致的运行原理，之后我的 Nico 也是基于这个逻辑进行开发。

安卓方面

PC 端与移动端的通讯

基于之前的探索，我放弃了使用 adb 命令来进行 dump 的操作，还是转而使用和 appium，airtest 一样使用安卓自带的 UIAutomator2 来进行底层的元素 dump 和操作交互。

但我之前也有提到过，最初不想使用这种这套模式的根本原因就是，它存在一些弊端 (相对于我们的项目来说)：
1.通讯过程比较复杂
2.在初始化 driver 时，移动端的启动监听服务的 app 会在前台闪现一次 (设备重启的时候，容易和我们自启动 app 冲突，导致应用 crash)

首先是第一点，在分析完整个 appium 的运行逻辑之后，我认为整体流程其实可以简单理解为 PC 与 Mobile 之间进行通讯。那我们需要做的就是在移动端启动一个 server，并持续监听一个指定端口例如：8080，然后我们可以借助端口转发工具 adb forward tcp 来进行端口发转，这时候 PC 端只需要向 8080 端口发送请求就能够间接的访问到移动端。这里去掉的 appium server 的处理工作，使用端口转发，将 PC 端的请求直接转发给移动端 。逻辑如下图所示：

然后是代码，这里安卓服务点端的代码，它启动了一个服务器线程，该线程在指定端口上监听客户端连接，通过解析传入的消息处理客户端请求，并支持根据特定消息内容（如 "close"）关闭服务器连接，同时主线程持续监控服务器线程的运行状态。

1. // 初始化服务器init();Thread serverThread = new Thread(() -> {

2. try {

3. // 获取传递给Instrumentation的参数

4. Bundle arguments = InstrumentationRegistry.getArguments();

5. String portString = arguments.getString("port");

7. // 使用提供的端口，如果没有提供则默认为9000

8. int port = portString != null ? Integer.parseInt(portString) : 9000;

9. InetAddress serverAddress = InetAddress.getByName("localhost");

11. // 创建绑定到指定端口和地址的ServerSocket

12. serverSocket = new ServerSocket(port, 0, serverAddress);

13. Log.i(TAG, "服务器正在监听: " + serverAddress.getHostAddress() + ":" + portString);

15. while (true) {

16. Log.i(TAG, "开始接收请求: ");

18. // 接受一个传入的连接

19. Socket socket = serverSocket.accept();

20. String msg;

21. OutputStream outputStream;

22. Map<String, Object> result;

24. // 格式化传入的socket消息

25. result = format_socket_msg(socket);

26. msg = (String) result.get("msg");

27. outputStream = (OutputStream) result.get("outputStream");

29. // 确保消息和输出流不为空

30. assert msg != null;

31. assert outputStream != null;

33. // 检查消息是否包含"close"以关闭连接

34. if (msg.contains("close")) {

35. socket.close(); // 关闭客户端socket

36. serverSocket.close(); // 关闭服务器socket

37. return;

38. }

40. // 处理客户端请求

41. handleClientRequest(socket, msg, outputStream);

42. }

43. } catch (IOException e) {

44. e.printStackTrace();

45. }});// 启动服务器线程serverThread.start();// 主线程循环检查服务器线程是否仍在运行while (serverThread.isAlive()) {

46. try {

47. // 每秒检查一次

48. Thread.sleep(1000);

49. } catch (InterruptedException e) {

50. e.printStackTrace();

51. }}

然后是 PC 端代码，因为整体框架是 python 写的，所以这里也是由 python 实现，简单来说就是向指定端口发送 socket 的请求：

1. def send_tcp_request(port: int, message: str):

2. # logger.debug(f"send_tcp_request: {port} {message}") try:

3. client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

4. client_socket.connect(("localhost", port))

5. client_socket.sendall(message.encode())

6. client_socket.sendall('\n'.encode())

8. # 接收服务器响应 chunks = []

9. while True:

10. chunk = client_socket.recv(1024) # 一次最多接收 1024 字节数据 if not chunk:

11. break

12. chunks.append(chunk)

13. client_socket.close()

14. response = b''.join(chunks)

15. if "get_jpg_pic" in message or message == "stop_recording" or "get_png_pic" in message:

16. return response

17. else:

18. response = response.decode()

19. return response

21. except ConnectionRefusedError as b:

22. logger.error(f"{str(b)} by {port}")

23. return f"{str(b)} by {port}"

24. except ConnectionResetError as b:

25. logger.error(f"{str(b)} by {port}")

26. return f"{str(b)} by {port}"

端口转发只需要执行adb forward tcp:xxxx tcp:xxxx，这样就完成了一个完成的通讯过程。

接下来我们来说第二点问题。现有的大多数例如 Appium、Airtest 对服务端的实现基本都是依托一个常规的 apk 包，在启动测试的时候调用adb shell am start -n com.example.package/.MainActivity，这导致在启动 driver 的时候不可避免的会有一个启动页面然后才迅速进入后台的动作，那么如何做到无感知地启动我们的测试 driver 呢？经过一番调研，我发现可以利用安卓测试包的特性来完成。

安卓的测试包（Test APK）是用于移动应用程序测试的安装包文件，其全称是 Android Package Kit for Testing，简称 APK for Testing。APK 是 Android 操作系统上的应用程序安装包格式，而测试包则是在开发过程中用于进行各种测试（如功能测试、性能测试、兼容性测试等）的特定 APK 文件。

那么设计思路就是，我写一个什么内容都没有的空包.apk，然后再为其写一个测试包，将所有调用 uiautomator 的逻辑以及服务监听的逻辑都放在里面。

接着使用adb shell am instrument -r -w -e port 指定端口 -e 空壳包包名 空壳测试包包名来启动监听服务，当我看到如下的图的输出时，表示 server 启动成功

然后，让我发起 sokect 请求来查看是否能够成功读取数据到 dump 数据，这里可以看到，sockect 发送成功，并且成功获取到了返回值。

到此为止，就算是实现了整个通讯过程。

服务端的初始化和保活

与 Appium 不同的是，Appium 通过会话来管理和与特定的设备交互，每次启动一个新的测试时，Appium 会创建一个新的会话，并返回一个会话 ID。所有后续的命令都通过这个会话 ID 与 Appium 服务器通信，以确保命令被发送到正确的设备和应用。

而 Nico 不存在额外的客户端，因此管理与特定设备的交互都是通过指定端口号来实现的。且在 Appium 在初始化的过程中，没有直接提供获取当前设备已启动的会话 id 的动作。这就意味着，如下图我们在不同的测试脚本中分别初始化 driver，会启动两个不同的会话，移动端的 server 也会被启动两次。

当然我们也可以编写额外的逻辑来实现实例的共享，避免不断产生新的会话。例如 pytest，我们通常会在 setup 或者 conftest 中先初始化当前设备的实例，然后在不同的测试脚本中去调用

但在 Nico 中，我希望对 driver 的管理直接在底层处理掉，简化整个框架的使用。因此我的初始化逻辑如下图

然后我们执行用例看看，第一次完全初始化运行：
从启动服务，到执行查找总共花了 3s 接近 4s

第二次运行：
由于不需要再次启动，只花了 265ms

这样不仅能够极大的减少初始化的时间，同时用户几乎不需要考虑实例共享的问题。每次直接实例化全新的类，剩下的交由代码底层去判断。

再一个是保活的机制，在我的印象里，appium 是没有做保活机制的。即如果中途 UIAutomator Server crash 或者 appium 服务 crash，整个会话则会中断，且不会自动恢复。需要自己重启启动一个新的会话。

然后 Airtest 则是独立开了一个线程来持续观察移动端服务存活情况，随时掉线随时重启。

这样做的好处就是可以保证移动端 UIAutomator Server 的健壮性，但坏处就是如果 UIAutomator Server 重启的时间点刚好与当前 app 的某些操作重叠。比如 app 刚好也需要打开某个 Activity，或者处于 Activity 切换的过程中，UIAutomator Server 也正好在进行重启，就会发生两个 app 互相抢占前台，或者另一方启动失败。

我为了避免这种情况，因此也没有做相关的保活措施。如同前文所说，我会在每次调用请求接口之前，先发送一个简单的状态查询。如果状态查询失败，再重启 UIAutomator Server，且因为我的 UIAutomator Server 是全程后台执行，整个重启过程也不会让前台 app 有任何感知。

元素查找与操作

查找方面，我借鉴了 Airtest 的语法规则，添加了两种查找逻辑
1.nico(text="蓝牙").wait_for_appearance(timeout)， 在指定超时时间内不断重试，超时则报错，提示无法在指定时间内以指定查询条件找到该元素
2.nico(text="蓝牙”).exit()，在指定超时时间内不断重试，超时只返回 rue, False 表示元素存在的状态，不会报错

其他的一些都是基于这两类进行的延申，详情可查看 github 里的 Readme，这里就不赘述。值得一提的是，除了基本的父子节点的获取之外，我还添加了兄弟节点的直接获取，例如nico(text="蓝牙").last_sibling 和nico(text="蓝牙").next_sibling

然后对于元素查找的底层逻辑，我没有选择使用调用安卓 uiautomtor 自带的 API，而是直接将当前整个元素页面 dump 下来获得一个 xml 结构的树。然后使用 lxml 库将其初始化为为一个lxml.tree的对象，利用lxml.tree对象自带的查方式进行元素查找。这里的root参数即为我们 dump 下来的元素树对象，query是一个查询条件的键值对，例如：{text:"蓝牙"}。

1. ###底层查 def __find_element_by_query_by_xml(root, query) -> Union[list, None]:

2. xpath_expression = ".//*"

3. conditions = []

4. is_re = False

5. for attribute, value in query.items():

6. if attribute == "compressed":

7. pass

8. elif attribute == "xpath":

9. xpath_expression = convert_xpath(value)

10. matching_elements = root.xpath(xpath_expression)

11. else:

12. if attribute.find("_matches") > 0:

13. is_re = True

14. attribute = attribute.replace("_matches", "")

15. condition = f"re:match(@{attribute},'{value}')"

16. elif attribute.find("_contains") > 0:

17. attribute = attribute.replace("_contains", "")

18. condition = f"contains(@{attribute},'{value}')"

19. else:

20. condition = f"@{attribute}='{value}'"

21. conditions.append(condition)

22. if conditions:

23. xpath_expression += "[" + " and ".join(conditions) + "]"

24. if is_re:

25. ns = {"re": "http://exslt.org/regular-expressions"}

26. matching_elements = root.xpath(xpath_expression, namespaces=ns)

27. else:

28. matching_elements = root.xpath(xpath_expression)

29. if len(matching_elements) == 0:

30. return None

31. else:

32. return matching_elements

这意味着移动端只负责 dump 元素，所有查询逻辑全部放到 pc 端来处理。那么这样做的有什么好处呢？听我娓娓道来。

首先，也是最重要的一点，这个 dump 元素的速并不算慢，而且dumpWindowHierarchy 这个方法还有一个参数是 compressed，用于控制是否需压缩元素。代价是失去一些不重要的元素节点，但却能够进一步提升元素 dump 的速度 (如果有特殊需要，也可以将参数设为 false 以获取更完整的元素树)。

1. private void handleDumpRequest(OutputStream outputStream,boolean compressed) throws IOException {

2. File response = dumpWindowHierarchy(compressed, "ui_tree.xml");

3. try (FileInputStream fis = new FileInputStream(response)) {

4. byte[] buffer = new byte[1024];

5. int length;

6. while ((length = fis.read(buffer)) != -1) {

7. outputStream.write(buffer, 0, length);

8. }

9. outputStream.flush(); // 确保所有的数据都被写入到OutputStream

10. } catch (IOException e) {

11. e.printStackTrace();

12. }

13. }

在 dump 不算慢的基础上，我们能够获取整个元素结构和其所有属性，这样非常便于我们去获取当前元素的兄弟节点，父子节点。要知道，有时候当一个元素不存在唯一标识符的时候，相对位置的元素查找方式是一个非常好用的（这里吐槽一下，airtest 中没有直接通过当前元素节点获取上下节点的方法，提了 PR , 过了快两年了也没人审核通过）。同时，一次 dump 即可获取左右兄弟或者父子各种节点，无需发送多次请求。

然后为了进一步减少请求次数，我设计了缓存机制，每一次 dump 元素都将当前的元素树写入临时环境变量进行缓存。

并且通过一个全局的 tag 来记录当前 UI 页面是否变化。在执行 UI 查找动作前先获取当前 tag 的状态，为True则进行新一轮的 dump，反之，则直接使用缓存的元素数据。

1. def _get_root_node(self, configuration: dict):

2. """

3. get the root node of the element tree

4. @param configuration: The configuration of the platform

5. @param udid: The device ID

6. @param port: The port number of automation server

7. @param force_reload: Whether to force reload the element tree, default is False

8. @return: The root node of the element tree

9. """

10. ui_change_status = RunningCache(self.udid).get_ui_change_status()

11. # logger.debug(f"ui tree change is {ui_change_status}")

12. if not ui_change_status:

13. # logger.debug(f"{self.udid}'s UI no change. There is no need to dump again!!") return RunningCache(self.udid).get_current_cache_ui_tree()

14. else:

15. # logger.debug(f"{self.udid}'s UI is change. dump again!!") return self._dump_ui_xml(configuration)

这样在进行一个静态页面的多元素检查时，可以减少请求次数，提升测试速度。例如以下场景需要同时检查多个文本是否存在，实际上只需要在发起一次 dump 请求，后续的几次查找都是基于之前的缓存内容。

如果此时执行了一些动作，例如元素点击，滑动等，都会在执行之后主动清除缓存元素树。

1. def click(self, x=None, y=None, x_offset=None, y_offset=None):

2. if x is None and y is None:

3. x = self.center_coordinate()[0]

4. y = self.center_coordinate()[1]

5. if x_offset is not None:

6. x = x + x_offset

7. if y_offset is not None:

8. y = y + y_offset

9. command = f'adb -s {self.udid} shell input tap {x} {y}'

10. os.system(command)

11. RunningCache(self.udid).clear_current_cache_ui_tree()

12. logger.debug(f"click {x} {y}")

同时，在移动端，我额外添加了一个监听者线程，用于检查页面 UI 是否发生变动，这样针对一些动态场景一样能保证及时的刷新

1. private final UiAutomation.AccessibilityEventFilter checkWindowUpdate = event -> {

2. if (event.getEventType() == AccessibilityEvent.TYPE_WINDOW_CONTENT_CHANGED) {

3. lastWindowChangeEvent = event;

4. return true;

5. }

6. return false;};private final AtomicBoolean uiChanged = new AtomicBoolean(false);private void startWatchingUiChanges() {

7. Thread watcherThread = new Thread(() -> {

8. while (true) {

9. try {

10. InstrumentationRegistry.getInstrumentation().getUiAutomation().executeAndWaitForEvent(

11. () -> {},

12. checkWindowUpdate,

13. 5000

14. );

15. uiChanged.set(true);

16. } catch (TimeoutException e) {

17. continue;

18. }

19. }

20. });

21. watcherThread.start();}

对于元素的操作，我还是选择在获取到元素之后，通过该元素的坐标直接调用 adb shell 命令来直接进行 click 等

1. def click(self, x=None, y=None, x_offset=None, y_offset=None):

2. if x is None and y is None:

3. x = self.center_coordinate()[0]

4. y = self.center_coordinate()[1]

5. if x_offset is not None:

6. x = x + x_offset

7. if y_offset is not None:

8. y = y + y_offset

9. command = f'adb -s {self.udid} shell input tap {x}{y}'

10. os.system(command)

11. RunningCache(self.udid).clear_current_cache_ui_tree()

12. logger.debug(f"click {x}{y}")

为啥不调 uiautomator 的 api 呢，一个原因是我想减少与 uiautomator 服务的请求次数，第二就是我偷懒了:）想少写点安卓的代码。。。再加上目前 adb 命令基本可以覆盖大部分场景了，如果有额外的需求再往 apk 里加吧，毕竟底层也是自己写的。

结束语

以上就是 Nico 安卓部分全部内容了，由于篇幅有点长了。。剩下的内容我将分成另外的篇幅进行撰写。接下来我会带来 iOS 部分以及一个额外的 inspector(检索 iOS 和安卓页面元素的工具）的分享，同样会包括技术原理和关键代码讲解。