Appium UiAutomator2驱动：并行测试

一、并行测试介绍

UiAutomator2驱动支持并行测试，Appium支持两类并行测试：

• 多服务器进程多会话：多个Appium服务进程监听不同的端口，客户端跟不同的服务端进程建立一对一的会话。

• 单服务器进程多会话：只有一个Appium服务进程，客户端跟一个服务进程建立多个会话。好处是对资源占用更少，且能保证对运行会话更好的控制。

注意：如果不考虑并行测试，那么建议在启动Appium服务端的时候使用参数--session-override，它会在打开新的会话之前强制关闭所有挂起的会话，避免在后台静默运行或过期的会话导致无法预期的问题。

1、真机并行测试相关的重要能力参数

• appium:udid：设备唯一id。

• appium:systemPort ：并行测试时，每个会话需要指定一个唯一的端口，否则可能会出现端口冲突的问题，建议每个会话都分配一个单独的端口。

• appium:chromedriverPort：chrome驱动监听端口（测试web试图或chrome浏览器）。

• appium:mjpegServerPort：如果想记录视频，需要给每一个会话设置一个唯一的MJPEG服务端口。

2、模拟器并行测试相关的重要能力参数

• appium:avd：模拟器的唯一名称。

• appium:systemPort ：并行测试时，每个会话需要指定一个唯一的端口，否则可能会出现端口冲突的问题，建议每个会话都分配一个单独的端口。

• appium:chromedriverPort：chrome驱动监听端口（测试web试图或chrome浏览器）。

• appium:mjpegServerPort：如果想记录视频，需要给每一个会话设置一个唯一的MJPEG服务端口。

二、并行测试实践

备注：本文所有测试代码的前提是已经安装Appium和相关环境（例如JDK、Android SDK、AVD模拟器），可以参考Appium环境搭建。本文使用被测设备是模拟器，在环境搭建的时候已经知道如何创建一个模拟器设备，由于本文要演示并行测试，因此需要再创建一个模拟器设备，假定模拟器设备名称为testPhone2，参考环境搭建一文，使用如下avdmanager命令创建：

avdmanager create avd -n "testPhone2" -k "system-images;android-33;google_apis;x86_64"

创建后使用avdmanager list avd可以看到两个模拟器。

[图片上传失败...(image-ddd19c-1691618507697)]

• 场景1： 1 Appium服务端 + 2 Android模拟器 + 2 Python客户端，2个客户端连接到同一个服务端，建立两个并行会话，分别测试2个模拟器

 # -*- coding: utf-8 -*-
 
 import pytest
 
 from appium import webdriver
 from appium.options.android import UiAutomator2Options
 from appium.webdriver.appium_service import AppiumService
 from appium.webdriver.common.appiumby import AppiumBy
 
 # 开启服务端
 APPIUM_HOST = '127.0.0.1'
 APPIUM_PORT = 4723
 @pytest.fixture(scope="session")
 def start_appium_service():
  server = AppiumService()
  server.start(args=['--address', APPIUM_HOST, '-p', str(APPIUM_PORT)], timeout_ms=60000)
  yield server
  server.stop()
 
 # 创建客户端到服务端的会话
 def create_appium_session_by_api(custom_opts = None, appium_host = APPIUM_HOST, appium_port = APPIUM_PORT):
  options = UiAutomator2Options()
  if custom_opts is not None:
  options.load_capabilities(custom_opts)
  return webdriver.Remote(f'http://{appium_host}:{appium_port}', options=options)

 def test_parallel_per_request(start_appium_service):
  custom_opts = {
  "appium:appPackage": "com.google.android.dialer",
  "appium:appActivity": ".extensions.GoogleDialtactsActivity",
  "appium:avd": "testPhone",
  "appium:systemPort": 8201,
  }
 
  custom_opts2 = {
  "appium:appPackage": "com.google.android.dialer",
  "appium:appActivity": ".extensions.GoogleDialtactsActivity",
  "appium:avd": "testPhone2",
  "appium:systemPort": 8202,
  }
 
  driver = create_appium_session_by_api(custom_opts)
  driver2 = create_appium_session_by_api(custom_opts2)
 
  contact_id_full = "com.google.android.dialer:id/tab_contacts"

  contact = driver.find_element(AppiumBy.ID, value=contact_id_full)
  contact2 = driver2.find_element(AppiumBy.ID, value=contact_id_full)
 
  contact.click()
  contact2.click()

  driver.quit()
  driver2.quit()

• 场景2： 2 Appium服务端 + 2 Android模拟器 + 2 Python客户端，2个客户端连接到2个服务端进程，建立两个并行会话，分别测试2个模拟器

    # -*- coding: utf-8 -*-
    
    import pytest
    
    from appium import webdriver
    from appium.options.android import UiAutomator2Options
    from appium.webdriver.appium_service import AppiumService
    from appium.webdriver.common.appiumby import AppiumBy
    
    # 开启服务端
    APPIUM_HOST = '127.0.0.1'
    APPIUM_PORT = 4723
    @pytest.fixture(scope="session")
    def start_appium_service():
     server = AppiumService()
     server.start(args=['--address', APPIUM_HOST, '-p', str(APPIUM_PORT)], timeout_ms=60000)
     yield server
     server.stop()
    
    APPIUM_PORT2 = 4724
    @pytest.fixture(scope="session")
    def start_appium_service2():
     server = AppiumService()
     server.start(args=['--address', APPIUM_HOST, '-p', str(APPIUM_PORT2)], timeout_ms=60000)
     yield server
     server.stop()
    
    # 创建客户端到服务端的会话
    def create_appium_session_by_api(custom_opts = None, appium_host = APPIUM_HOST, appium_port = APPIUM_PORT):
     options = UiAutomator2Options()
     if custom_opts is not None:
     options.load_capabilities(custom_opts)
     return webdriver.Remote(f'http://{appium_host}:{appium_port}', options=options)
    
    # 多服务端进程测试
    def test_parallel_per_process(start_appium_service, start_appium_service2):
     custom_opts = {
     "appium:appPackage": "com.google.android.dialer",
     "appium:appActivity": ".extensions.GoogleDialtactsActivity",
     "appium:avd": "testPhone",
     "appium:systemPort": 8201,
     }
    
     custom_opts2 = {
     "appium:appPackage": "com.google.android.dialer",
     "appium:appActivity": ".extensions.GoogleDialtactsActivity",
     "appium:avd": "testPhone2",
     "appium:systemPort": 8202,
     }
    
     driver = create_appium_session_by_api(custom_opts, appium_port=APPIUM_PORT)
     driver2 = create_appium_session_by_api(custom_opts2, appium_port=APPIUM_PORT2)
    
     contact_id_full = "com.google.android.dialer:id/tab_contacts"
    
     contact = driver.find_element(AppiumBy.ID, value=contact_id_full)
     contact2 = driver2.find_element(AppiumBy.ID, value=contact_id_full)
    
     contact.click()
     contact2.click()
    
     driver.quit()
     driver2.quit()

附录

1. 官网文档参考
2. 本文原始地址