pytest多线程与多设备并发appium

发布时间：2023年12月25日

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

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

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

  • 确定设备个数
  • 每个设备对应一个 appium server 的端口号，并启动 appium
  • 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中的列表。

实现的部分代码为：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

"""

@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的报错。

实现的部分代码：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

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 命令行版，也可以使用桌面版)

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

代码实现如下：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

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 有参数可以传递多个设备启动信息，但它是串行执行的。

需要解决的问题的是：

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

解决方案：

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

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

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

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

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。

run_case 函数。

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

1

2

3

4

5

6

7

8

9

10

11

# 根据设备启动信息，通过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)]

????????????????)

每有一个设备，就启动一个线程，执行 run_cases 方法。

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

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

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

?现在我也找了很多测试的朋友，做了一个分享技术的交流群，共享了很多我们收集的技术文档和视频教程。
如果你不想再体验自学时找不到资源，没人解答问题，坚持几天便放弃的感受
可以加入我们一起交流。而且还有很多在自动化，性能，安全，测试开发等等方面有一定建树的技术大牛
分享他们的经验，还会分享很多直播讲座和技术沙龙
可以免费学习！划重点！开源的！！！
qq群号：485187702【暗号：csdn11】

最后感谢每一个认真阅读我文章的人，看着粉丝一路的上涨和关注，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走！?希望能帮助到你！【100%无套路免费领取】

文章来源:https://blog.csdn.net/qq_45158700/article/details/135200627
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
最新文章