app自動化測試----多設備並行運行

1. 總結

 

 

app自動化測試---多台設備並行運行monkey（subprocss 子進程方式 && multiprocessing 多進程方式）

app自動化測試----使用Python代碼啟動和關閉 一個/多個設備Appium

app自動化測試---- pytest 多設備連接,並行執行測試用例（pytest 通過設置變量的方式傳參）

 

2.樣例代碼展示

main.py  （項目運行文件）

"""
  項目運行文件，並添加測試報告
"""
import pytest
import os,time
import multiprocessing
from utils.adb_handler import get_connect_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 = time.strftime('%Y-%m-%d_%H-%M-%S') + '_' + str(device[1])

    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()

 

conftest.py  （pytest配置文件）

 

from appium import webdriver
import os, time,pytest
from utils.adb_handler import start_appium, stop_appium
from utils.file_handler import FileHandler

chromedriver= os.path.join(os.path.dirname(os.path.abspath(__file__)),'drivers/chrome/75.0.3770.140/chromedriver.exe')

# scope='session' 標記的方法執行域為---->所有測試用例運行之前/之后 運行的方法
@pytest.fixture(scope='session',autouse=True)
def driver():
    # 啟動appium服務
    port = os.environ['port']
    start_appium(port)

    desired_caps = {
        'platformName': 'Android',                    # 測試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"  # 應用的活動頁名稱(appium會啟動app的加載頁)
    }
    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)


# 該方法是用來獲取測試用例執行的結果（passed / FAILED）
@pytest.hookimpl(tryfirst=True,hookwrapper=True)
def pytest_runtest_makereport(item, call):
    outcome = yield
    rep = outcome.get_result()               # 獲取用例的執行結果
    # print('用例的執行結果rep---->',rep)
    setattr(item, "rep_" + rep.when, rep)    # 將執行結果保存到 item 屬性中 , req.when 執行時


# scope='function' 標記的方法執行域為---->每個測試用例運行之前/之后 運行的方法
@pytest.fixture(scope='function',autouse=True)
def case_run(driver:webdriver,request):   # request 為 pytest_runtest_makereport 方法獲取到的執行結果(固定參數和用法)
    yield
    if request.node.rep_call.failed:
        file_handler = FileHandler()
        screenshots = file_handler.save_file_dir('img')
        casename: str = request.node.nodeid
        # print("執行測試用例的名字：", casename)
        # 測試用例的名字很長 testcases/test_ddt/test_ddt_login.py::TestDdtLogin::test_login[....]
        # 對字符串進行截取，截取之后顯示為  test_ddt_login-TestDdtLogin
        casename = casename[casename.rfind('/')+1:casename.rfind('::')].replace('.py', '').replace('::', '-')
        filename = time.strftime('%H_%M_%S') + '-' + casename +".png"
        screenshot_file = os.path.join(screenshots, filename)
        # 保存截圖
        driver.save_screenshot(screenshot_file)

 

 

 

utils/adb_handler.py   （adb命令處理文件類）

"""
  adb命令處理文件類
"""
import subprocess
import os,sys,time
from utils.file_handler import FileHandler
from config import appoum_port

file_handler = FileHandler()
logs_dir = file_handler.save_file_dir('log')

# 獲取所有的連接設備
def get_connect_devices():
    devices = []
    port = appoum_port
    proc = subprocess.Popen('adb devices',stdout=subprocess.PIPE,shell=True)
    for line in proc.stdout.readlines():
        # 將字節類型轉換為字符串
        line_str = line.decode(encoding='utf8')
        if '\tdevice' in line_str:
            # 字符串分割 提取 deviceid值
            device_id = line_str.strip().split('\tdevice')[0]
            devices.append((device_id,port))
            port += 2
    print('devices連接設備----》', devices)
    return devices


# 使用命令行的方式啟動appium
def start_appium(port):
    """
    啟動appium 服務
    :param port: 服務的端口號
    :return:
    """
    stop_appium(port)
    cmd = f"appium -p {port}"
    logsdir = file_handler.save_file_dir('log')
    log_name = time.strftime('%Y_%m_%d') + '-appium_log-' +  str(port) +".log"
    appium_logs_dirName = os.path.join(logsdir,log_name)
    with open(appium_logs_dirName, mode='a', encoding="utf8") as file:
        subprocess.Popen(cmd, shell=True, stdout=file,stderr=subprocess.PIPE)


# 使用命令行的方式關閉appium
def stop_appium(port):
    mac_cmd = f"lsof -i tcp:{port}"
    win_cmd = f"netstat -ano | findstr {port}"
    # 判斷操作系統
    os_platform = sys.platform
    print('操作系統：',os_platform)
    # #windows 系統
    if os_platform == "win32":
        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()
                    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')
            if "node" in line:
                stdoutline = line.split(" ")
                # print(stdoutline)
                pid = stdoutline[4]
                os.system(f"kill {pid}")