app 自动化测试 - 多设备并发 -appium+pytest+ 多线程

本贴最后更新于 1287 天前,其中的信息可能已经时移世改

1、appium+python实现单设备的app自动化测试

  1. 启动appium server,占用端口4723
  2. 电脑与一个设备连接,通过adb devices获取已连接的设备
  3. 在python代码当中,编写启动参数,通过pytest编写测试用例,来进行自动化测试。



2、若要多设备并发,同时执行自动化测试,那么需要:

  1. 确定设备个数
  2. 每个设备对应一个appium server的端口号,并启动appium
  3. pytest要获取到每个设备的启动参数,然后执行自动化测试。

image.png



3、实现策略

第一步:从设备池当中,获取当前连接的设备。若设备池为空,则无设备连接。

第二步:若设备池不为空,启动一个线程,用来启动appium server.与设备个数对应。
       起始server端口为4723,每多一个设备,端口号默认+4

第三步:若设备池不为空,则启用多个线程,来执行app自动化测试。



4、具体实现步骤

4.1 通过adb命令,获取当前已连接的设备数、设备名称、设备的安卓版本号。

定义一个ManageDevices类。

1. 重启adb服务。
2. 通过adb devices命令获取当前平台中,已连接的设备个数,和设备uuid.
3. 通过adb -P 5037 -s 设备uuid shell getprop ro.build.version.release获取每一个设备的版本号。
4. 将所有已连接设备的设备名称、设备版本号存储在一个列表当中。
5. 通过调用get_devices_info函数,即可获得4中的列表。

实现的部分代码为:

"""
@Title   : app多设备并发-appium+pytest
@Author  : 柠檬班-小简
@Email   : lemonban_simple@qq.com
"""

class ManageDevices:
    """
       1、重启adb服务。
       2、通过adb devices命令获取当前平台中,已连接的设备个数,和设备uuid.
       3、通过adb -P 5037 -s 设备uuid shell getprop ro.build.version.release获取每一个设备的版本号。
       4、将所有已连接设备的设备名称、设备版本号存储在一个列表当中。
       5、通过调用get_devices_info函数,即可获得4中的列表。
    """

    def __init__(self):
        self.__devices_info = []
        # 重启adb服务
        self.__run_command_and_get_stout("adb kill-server")
        self.__run_command_and_get_stout("adb start-server")

    def get_devices_info(self):
        """
        获取已连接设备的uuid,和版本号。
        :return: 所有已连接设备的uuid,和版本号。
        """
        self.__get_devices_uuid()
        print(self.__devices_info)
        self.__get_device_platform_vesion()
        return self.__devices_info

4.2 定义一个设备配置池。

设备启动参数管理池。
每一个设备:对应一个启动参数,以及appium服务的端口号。

1. desired_caps_config/desired_caps.yaml文件中存储了启动参数模板。
2. 从1中的模板读取出启动参数。
3. 从设备列表当中,获取每个设备的设备uuid、版本号,与2中的启动参数合并。
4. 每一个设备,指定一个appium服务端口号。从4723开始,每多一个设备,默认递增4
5. 每一个设备,指定一个本地与设备tcp通信的端口号。从8200开始,每多一个设备,默认递增4.
   在启动参数当中,通过systemPort指定。
   因为appium服务会指定一个本地端口号,将数据转发到安卓设备上。
   默认都是使用8200端口,当有多个appium服务时就会出现端口冲突。会导致运行过程中出现socket hang up的报错。

实现的部分代码:

def devices_pool(port=4723,system_port=8200):
    """
    设备启动参数管理池。含启动参数和对应的端口号
    :param port: appium服务的端口号。每一个设备对应一个。
    :param system_port: appium服务指定的本地端口,用来转发数据给安卓设备。每一个设备对应一个。
    :return: 所有已连接设备的启动参数和appium端口号。
    """
    desired_template = __get_yaml_data()
    devs_pool = []
    # 获取当前连接的所有设备信息
    m = ManageDevices()
    all_devices_info = m.get_devices_info()
    # 补充每一个设备的启动信息,以及配置对应的appium server端口号
    if all_devices_info:
        for dev_info in all_devices_info:
            dev_info.update(desired_template)
            dev_info["systemPort"] = system_port
            new_dict = {
                "caps": dev_info,
                "port": port
            }
            devs_pool.append(new_dict)
            port += 4
            system_port += 4
    return devs_pool

特别注意事项:

2个及2个以设备并发时,会遇到设备socket hang up的报错。

原因是什么呢:

在appium server的日志当中,有这样一行adb命令:

adb -P 5037 -s 08e7c5997d2a forward tcp\:8200 tcp\:6790

什么意思呢?

将本地8200端口的数据,转发到安卓设备的6790端口
所以,本地启动多个appium server,都是用的8200端口,就会出现冲突。

解决方案:

应该设置为,每一个appium server用不同的本地端口号,去转发数据给不同的设备。
启动参数当中:添加 systemPort=端口号 来设置。
这样,每个设备都使用不同的本地端口,那么可解决此问题。

4.3 appium server启停管理 。

(ps 此处可以使用appium命令行版,也可以使用桌面版)

  1. 在自动化用例运行之前,必须让appium server启动起来。
  2. 在自动化用例执行完成之后,要kill掉appium服务。这样才不会影响下一次运行。

代码实现如下:

import subprocess
import os

from Common.handle_path import appium_logs_dir

class ManageAppiumServer:
    """
    appium desktop通过命令行启动appium服务。
    不同平台上安装的appium,默认的appium服务路径不一样。
    初始化时,设置appium服务启动路径
    再根据给定的端口号启动appium
    """

    def __init__(self,appium_server_apth):
        self.server_apth = appium_server_apth

    # 启动appium server服务
    def start_appium_server(self,port=4723):
        appium_log_path = os.path.join(appium_logs_dir,"appium_server_{0}.log".format(port))
        command = "node {0} -p {1} -g {2} " \
                  "--session-override " \
                  "--local-timezone " \
                  "--log-timestamp & ".format(self.server_apth, port, appium_log_path)
        subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,shell=True).communicate()

    # 关闭appium服务
    @classmethod
    def stop_appium(cls,pc,post_num=4723):
        '''关闭appium服务'''
        if pc.upper() == 'WIN':
            p = os.popen(f'netstat  -aon|findstr {post_num}')
            p0 = p.read().strip()
            if p0 != '' and 'LISTENING' in p0:
                p1 = int(p0.split('LISTENING')[1].strip()[0:4])  # 获取进程号
                os.popen(f'taskkill /F /PID {p1}')  # 结束进程
                print('appium server已结束')
        elif pc.upper() == 'MAC':
            p = os.popen(f'lsof -i tcp:{post_num}')
            p0 = p.read()
            if p0.strip() != '':
                p1 = int(p0.split('\n')[1].split()[1])  # 获取进程号
                os.popen(f'kill {p1}')  # 结束进程
                print('appium server已结束')

4.4 pytest当中根据不同的启动参数来执行自动化测试用例

在使用pytest执行用例时,是通过pytest.main()会自动收集所有的用例,并自动执行生成结果。

image.png

这种情况下,appium会话的启动信息是在代码当中给定的。

image.png

image.png

以上模式当中,只会读取一个设备的启动信息,并启动与设备的会话。

虽然fixture有参数可以传递多个设备启动信息,但它是串行执行的。

需要解决的问题的是:

  1. 可以传递多个设备的启动参数,但不是通过fixture的参数。
  2. 每传递一个设备启动参数进来,执行一次pytest.main()

解决方案:

  1. 通过pytest的命令行参数。即在pytest.main()的参数当中,将设备的启动信息传进来。
  2. 使用python的多线程来实现。每接收到一个设备启动参数,就启动一个线程来执行pytest.main
4.4.1 第一个,pytest的命令行参数。

首先需要在conftest.py添加命令行选项,命令行传入参数”--cmdopt“。

用例如果需要用到从命令行传入的参数,就调用cmdopt函数。

image.png

def pytest_addoption(parser):
    parser.addoption(
        "--cmdopt", action="store", default="{platformName:'Android',platformVersion:'5.1.1'}",
        help="my devices info"
    )


@pytest.fixture(scope="session")
def cmdopt(request):
    return request.config.getoption("--cmdopt")


@pytest.fixture
def start_app(cmdopt):
    device = eval(cmdopt)
    print("开始与设备 {} 进行会话,并执行测试用例 !!".format(device["caps"]["deviceName"]))
    driver = start_appium_session(device)
    yield driver
    driver.close_app()
    driver.quit()
4.4.2 使用多线程实现: 每接收到一个设备启动参数,就启动一个线程来执行pytest.main

定义一个main.py。

  1. run_case函数。

此方法主要是:接收设备启动参数,通过pytest.main去收集并执行用例。

# 根据设备启动信息,通过pytest.main来收集并执行用例。
def run_cases(device):
  """
  参数:device为设备启动参数。在pytest.main当中,传递给--cmdopt选项。
  """
    print(["-s", "-v", "--cmdopt={}".format(device)])
    reports_path = os.path.join(reports_dir,"test_result_{}_{}.html".format(device["caps"]["deviceName"], device["port"]))
    pytest.main(["-s", "-v",
                 "--cmdopt={}".format(device),
                 "--html={}".format(reports_path)]
                )
  1. 每有一个设备,就启动一个线程,执行run_cases方法。
# 第一步:从设备池当中,获取当前连接的设备。若设备池为空,则无设备连接。
devices = devices_pool()

# 第二步:若设备池不为空,启动appium server.与设备个数对应。起始server端口为4723,每多一个设备,端口号默认+4
if devices and platform_name and appium_server_path:
    # 创建线程池
    T = ThreadPoolExecutor()
    # 实例化appium服务管理类。
    mas = ManageAppiumServer(appium_server_path)
    for device in devices:
        # kill 端口,以免占用
        mas.stop_appium(platform_name,device["port"])
        # 启动appium server
        task = T.submit(mas.start_appium_server,device["port"])
        time.sleep(1)

    # 第三步:若设备池不为空,在appium server启动的情况下,执行app自动化测试。
    time.sleep(15)
    obj_list = []
    for device in devices:
        index = devices.index(device)
        task = T.submit(run_cases,device)
        obj_list.append(task)
        time.sleep(1)

    # 等待自动化任务执行完成
    for future in as_completed(obj_list):
        data = future.result()
        print(f"sub_thread: {data}")

    # kill 掉appium server服务,释放端口。
    for device in devices:
        ManageAppiumServer.stop_appium(platform_name, device["port"])

2 操作
xiaojian 在 2020-09-08 14:01:39 更新了该帖
xiaojian 在 2020-09-08 13:59:40 更新了该帖
1 回帖
请输入回帖内容 ...
  • Soner

    有这个的单独的教程课吗