pytest 多设备连接

pytest 通过设置变量的方式传参

并行运行 pytest 的测试用例

import os
import pytest
import multiprocessing
devices = [('127.0.0.1:62001',4723),('127.0.0.1:62003',4725)]
@pytest.fixture(scope='session',autouse=True)
def session():
    # 读取环境变量
    print(f'启动服务器{os.environ["udid"]}, {os.environ["port"]}')
    yield
    print(f'关闭服务器{os.environ["udid"]}, {os.environ["port"]}')
def test_01():
    print('执行测试用例')
def run(value):
    # 设置环境变量 ('127.0.0.1:62001',4723)
    os.environ["udid"] = str(value[0]) # 设置变量的时候要求字符串
    os.environ['port'] = str(value[1])
    pytest.main(['test_pytest.py', '-s', '-v'])
if __name__ == '__main__':
    process = []
    for val in devices:
        # 针对每个进程设置环境变量
        p = multiprocessing.Process(target=run,args=(val,))
        p.start()
        process.append(p)
    for proc in process:
        proc.join()

项目代码

配置conftest.py

from appium import webdriver
import pytest
import os,sys,subprocess
from appium.webdriver.webdriver import  WebDriver
chromedriver= os.path.join(os.path.dirname(os.path.abspath(__file__)),'drivers/chromedriver.exe')
def stop_appium(port):
    """
    停止appium
    :param port: 启动的端口号
    :return:
    """
    mac_cmd = f"lsof -i tcp:{port}"
    win_cmd = f"netstat -ano | findstr {port}"
    # 判断操作系统
    os_platform = sys.platform
    if os_platform == "win32":  #windows 系统
        win_p = subprocess.Popen(win_cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
        for line in win_p.stdout.readlines():
            if line:
                line = line.decode('utf8')
                if "LISTENING" in line:
                    win_pid = line.split("LISTENING")[1].strip()
                    print(f"taskkill -f -pid {win_pid}")
                    os.system(f"taskkill -f -pid {win_pid}")
    else:                        # unix系统
        p = subprocess.Popen(mac_cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
        for line in p.stdout.readlines():
            line = line.decode('utf8')
            # print("line",line)
            if "node" in line:
                stdoutline = line.split(" ")
                print(stdoutline)
                pid = stdoutline[4]
                os.system(f"kill {pid}")
def start_appium(port):
    """
    启动appium 服务
    :param port: 服务的端口号
    :return:
    """
    stop_appium(port)
    cmd = f"appium -p {port}"
    logsdir = os.path.join(os.path.dirname(os.path.abspath(__file__)),"logs")
    if not os.path.exists(logsdir):
        os.mkdir(logsdir)
    subprocess.Popen(cmd,shell=True, stdout=open('./logs/'+str(port)+".log",mode='a',encoding="utf8"),
                     stderr=subprocess.PIPE)
@pytest.fixture(scope='session',autouse=True)
def driver():
    # 启动appium服务
    port = os.environ['port']
    start_appium(port)
    desired_caps = {
        'platformName': 'Android',  # 测试Android系统
        # 'platformVersion': '7.1.2',  # Android版本 可以在手机的设置中关于手机查看
        'udid': os.environ['udid'],  # adb devices 命令查看  设置为自己的设备
        'automationName': 'UiAutomator2',  # 自动化引擎
        'noReset': False,  # 不要重置app的状态
        'fullReset': False,  # 不要清理app的缓存数据
        'chromedriverExecutable': chromedriver,  # chromedriver 对应的绝对路径
        'appPackage': "org.cnodejs.android.md",  # 应用的包名
        'appActivity': ".ui.activity.LaunchActivity"  # 应用的活动页名称
    }
    driver = webdriver.Remote('http://127.0.0.1:4723/wd/hub', desired_capabilities=desired_caps)
    driver.implicitly_wait(5)  # 全局的隐式等待时间
    yield driver  # 将driver 传递出来
    # 所有的用例执行之后
    driver.quit()
    stop_appium(port)
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
    # execute all other hooks to obtain the report object
    outcome = yield
    # 获取用例的执行结果
    rep = outcome.get_result()
    # 将执行结果保存到 item 属性中  req.when 执行时
    setattr(item, "rep_" + rep.when, rep)
@pytest.fixture(scope='function',autouse=True)
def case_run(driver:webdriver,request):
    """
    每个测试用例执行完成之后，如果执行失败截图，截图的名称为测试用例名称+时间格式
    :param request:
    :return:
    """
    yield
    if request.node.rep_call.failed:
        import os,time
        screenshots = os.path.join(os.path.dirname(os.path.abspath(__file__)),'screeshots')
        if not os.path.exists(screenshots):
            os.mkdir(screenshots)
        casename:str = request.node.nodeid
        print("执行测试用例的名字：",casename)
        # 测试用例的名字
        # casename = casename.replace('.py::','_')
        filename = time.strftime('%Y_%m_%d_%H_%M_%S')+".png"
        screenshot_file = os.path.join(screenshots,filename)
        # 保存截图
        driver.save_screenshot(screenshot_file)

在main.py 文件中设置多进程

import pytest
import os,time
import multiprocessing,subprocess
def get_connect_devices():
    """
    获取已经成功连接的设备
    :return: list
    """
    devices = []
    port = 4723
    proc = subprocess.Popen('adb devices', stdout=subprocess.PIPE,shell=True)
    for line in proc.stdout.readlines():
        # print(type(line),line)
        # 将字节类型转换为字符串
        linestr = line.decode(encoding='utf8')
        if '\tdevice' in linestr:
            # 字符串分割 提取 deviceid值
            device_id = linestr.strip().split('\tdevice')[0]
            devices.append((device_id,port))
            port += 2 # 端口递增2
    return devices
def run(device):
    # 进程启动之后设置变量
    os.environ['udid'] = str(device[0])
    os.environ['port'] = str(device[1])
    report_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'reports')
    if not os.path.exists(report_dir):
        os.mkdir(report_dir)
    report = str(device[1])+time.strftime('%Y_%m_%d_%H_%M_%S')
    reportfile = os.path.join(report_dir, report + '.html')
    pytest.main(['testcases', '-s', '-v', f'--html={reportfile}'])
if __name__ == '__main__':
    devices = get_connect_devices()
    process = []
    for device in devices:
        # 创建进程
        p = multiprocessing.Process(target=run,args=(device,))
        p.start()
        process.append(p)
    for proc in process:
        proc.join()