app 自动化测试 - 多设备并发 -appium+pytest+ 多线程
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中的列表。
实现的部分代码为:
-
""" -
@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 命令行版,也可以使用桌面版)
- 在自动化用例运行之前,必须让 appium server 启动起来。
- 在自动化用例执行完成之后,要 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 有参数可以传递多个设备启动信息,但它是串行执行的。
需要解决的问题的是:
- 可以传递多个设备的启动参数,但不是通过 fixture 的参数。
- 每传递一个设备启动参数进来,执行一次 pytest.main()
解决方案:
- 通过 pytest 的命令行参数。即在 pytest.main()的参数当中,将设备的启动信息传进来。
- 使用 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。
- 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)] -
)
- 每有一个设备,就启动一个线程,执行 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"])
总结:
感谢每一个认真阅读我文章的人!!!
作为一位过来人也是希望大家少走一些弯路,如果你不想再体验一次学习时找不到资料,没人解答问题,坚持几天便放弃的感受的话,在这里我给大家分享一些自动化测试的学习资源,希望能给你前进的路上带来帮助。

软件测试面试文档
我们学习必然是为了找到高薪的工作,下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料,并且有字节大佬给出了权威的解答,刷完这一套面试资料相信大家都能找到满意的工作。


视频文档获取方式:
这份文档和视频资料,对于想从事【软件测试】的朋友来说应该是最全面最完整的备战仓库,这个仓库也陪伴我走过了最艰难的路程,希望也能帮助到你!以上均可以分享,点下方小卡片即可自行领取。
更多推荐


所有评论(0)