场景一、
有100条测试用例：两台手机设备，一台操作系统，实现每台手机设备同时执行100条测试用例
思路：
1、获取多个设备
2、启动多个appium server
appiun-server1 -> 设备1 appiun-server2 -> 设备2 。。。。。。
3、有多少台设备就启动多少个进程
4、每一台设备生成各生成一份alluredir报告
1、获取设备，封装设备信息
image.png
image.png
image.png
import os
from common._common import get_caps
class Devices():
def get_app_devices(self):
lists = os.popen('adb devices').read()
devices = lists.strip().split('\n')
devices_list = []
for i in range(1, len(devices)):
device = (devices[i].split('\t')[0])
devices_list.append(device)
return devices_list
def get_devices_info(self):
# 获取多设备信息
devices_info = []
cmd1 = 'adb -s {} shell getprop ro.product.model'
cmd2 = 'adb -s {} shell getprop ro.build.version.release'
devices_uuids = self.get_app_devices()
desired_caps = get_caps()
if devices_uuids:
for devices_uuid in devices_uuids:
device_model = os.popen(cmd1.format(devices_uuid)).read()
deviceos_version = os.popen(cmd2.format(devices_uuid)).read()
info = {
'deviceName':devices_uuid,
'device_model':device_model.strip('\n'),
'platformVersion':deviceos_version.strip('\n')
}
info.update(desired_caps)
devices_info.append(info)
return devices_info
conftest.py
import pytest, os
from selenium import webdriver
from appium import webdriver
import logging
from py._xmlgen import html
import allure
cur_path = os.path.dirname(os.path.realpath(__file__))
driver = None
def pytest_configure(config):
marker_list = ["smoke"] # 标签名集合
for markers in marker_list:
config.addinivalue_line(
"markers", markers
)
def pytest_addoption(parser):
parser.addoption("--cmdopt", action="store", default="device", help="None")
@pytest.fixture(scope="session")
def cmdopt(request):
return request.config.getoption("--cmdopt")
@pytest.fixture()
def app_page(cmdopt):
logging.info('----------------测试开始-----------------')
desired_caps = eval(cmdopt)
print("=======",desired_caps)
global driver
driver = webdriver.Remote('http://127.0.0.1:{0}/wd/hub'.format(desired_caps['appium_port']), desired_caps)
yield driver
logging.info('----------------测试结束-----------------')
driver.quit()
#通过conftest来实现报告的描述
@pytest.mark.optionalhook
def pytest_html_results_table_header(cells):
cells.insert(1, html.th('Description')) #html报告中插入一列，列头名为Description
@pytest.mark.optionalhook
def pytest_html_results_table_row(report, cells):
try:
cells.insert(1, html.td(report.description))
except:
pass
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
'''
获取每个用例状态的钩子函数
:param item:
:param call:
:return:
'''
# 获取钩子方法的调用结果
outcome = yield
rep = outcome.get_result()
# 仅获取用例call且执行结果是失败的情况, 不包含 setup/teardown
if rep.when == "call" and rep.failed:
# 添加allure报告截图
if hasattr(driver, "get_screenshot_as_png"):
with allure.step('添加失败截图..'):
allure.attach(driver.get_screenshot_as_png(), "失败截图", allure.attachment_type.PNG)
main.py
import pytest,os
from common import startappium,checkdevice
from concurrent.futures import ProcessPoolExecutor
def runPytest(device,report_list):
pytest.main([
'-m','smoke', #筛选带有smoke标记的所有测试用例
"--cmdopt={}".format(device),
'--clean-alluredir',
'--alluredir=' + report_list['result_path'],
'test_suites/test_caseone/',
])
def runnerPool(getdevice,report_list):
with ProcessPoolExecutor(len(getdevice)) as pool:
pool.map(runPytest, getdevice,report_list)
if __name__ == '__main__':
cur_path = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))))
r_path = cur_path + '/web/autotest/ui/'
r_report = cur_path + '/web/autotest/ui/'
#获取测试设备
getdevice = checkdevice.Devices().get_devices_info()
port = 4723
report_list = []
for index, value in enumerate(getdevice):
result_path = r_path + 'result_{0}/'.format(str(index + 1))
result_report = r_report + 'result_report_{0}/'.format(str(str(index + 1)))
if not os.path.exists(result_path): os.makedirs(result_path)
if not os.path.exists(result_report): os.makedirs(result_report)
report_list.append({'result_path': result_path, 'r_report': result_report})
getdevice[index]['appium_port'] = port
# 启动appium server
startappium.appium_start('127.0.0.1', port)
port += 2
import time
time.sleep(1)
runnerPool(getdevice,report_list)
#关闭appium server
po = 4723
for i in getdevice:
startappium.appium_close(po)
po += 2
for i in report_list:
os.system("allure generate --clean " + i['result_path'] + " --report-dir " + i[
'r_report']) # 转换为html
执行后成功生成两份测试报告：
image.png
image.png
场景二、
有100条测试用例：两台手机设备，一台操作系统，实现每台手机各自跑50条测试用例
(1)、方案一：粗暴的方法，将测试用例分成2组，每台设备跑一组用例，这样的方案会引起负载不均衡问题：
import pytest,os
from common import startappium,checkdevice
from concurrent.futures import ProcessPoolExecutor
def runPytest(device,case):
pytest.main([
'-m','smoke', #筛选带有smoke标记的所有测试用例
"--cmdopt={}".format(device),
'--alluredir=' + result_path,
case,
])
os.system("allure generate --clean "+result_path+" --report-dir "+result_report) #转换为html
def runnerPool(getdevice,cases):
with ProcessPoolExecutor(len(getdevice)) as pool:
pool.map(runPytest, getdevice,case)
if __name__ == '__main__':
cur_path = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))))
result_path = cur_path + '/web/autotest/ui/result/'
result_report = cur_path + '/web/autotest/ui/result_report/'
if not os.path.exists(result_path): os.makedirs(result_path)
if not os.path.exists(result_report): os.makedirs(result_report)
# 获取测试设备
getdevice = checkdevice.Devices().get_devices_info()
port = 4723
report_list = []
for index, value in enumerate(getdevice):
getdevice[index]['appium_port'] = port
# 启动appium server
startappium.appium_start('127.0.0.1', port)
port += 2
case = ['test_suites/test_casetwo','test_suites/test_caseone']
runnerPool(getdevice,case)
# 关闭appium server
port2 = 4723
for i in getdevice:
startappium.appium_close(port2)
port2 += 2
缺点就是：负载不均衡，也就是说，两台设备同时开始跑，设备a跑完了，停止休息了，设备b还有很多用例没跑完，而设备a没法继续分担设备b未跑完的用例。
方案二：解决负载均衡问题，使用： pytest-xdist https://pypi.org/project/pytest-xdist/
可参考这篇文章：https://blog.csdn.net/weixin_28947385/article/details/113040199
