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

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

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

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

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()会自动收集所有的用例，并自动执行生成结果。

 

这种情况下，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. 1run_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. 2每有一个设备，就启动一个线程，执行 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"])

最后： 下方这份完整的软件测试视频教程已经整理上传完成，需要的朋友们可以自行领取 【100%免费无套路】 

 全套资料获取方式：点击下方小卡片自行领取即可