"""

@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

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

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已结束')

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

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

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

需要解决的问题的是：

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

解决方案：

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

4.4.1 第一个，pytest 的命令行参数。

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

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

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"])