app 自動化測驗 - 多設備并發 -appium+pytest+ 多執行緒

1、appium+python 實作單設備的 app 自動化測驗

1. 啟動 appium server，占用埠 4723
2. 電腦與一個設備連接，通過 adb devices 獲取已連接的設備
3. 在 python 代碼當中，撰寫啟動引數，通過 pytest 撰寫測驗用例，來進行自動化測驗，

2、若要多設備并發，同時執行自動化測驗，那么需要：

1. 確定設備個數
2. 每個設備對應一個 appium server 的埠號，并啟動 appium
3. 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   : [email protected]
"""

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

1. 在自動化用例運行之前，必須讓 appium server 啟動起來，
2. 在自動化用例執行完成之后，要 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 有引數可以傳遞多個設備啟動資訊，但它是串行執行的，

需要解決的問題的是：

1. 可以傳遞多個設備的啟動引數，但不是通過 fixture 的引數，
2. 每傳遞一個設備啟動引數進來，執行一次 pytest.main()

解決方案：

1. 通過 pytest 的命令列引數，即在 pytest.main()的引數當中，將設備的啟動資訊傳進來，
2. 使用 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，

1. 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)]
                )

2. 每有一個設備，就啟動一個執行緒，執行 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"])

標籤：其他

上一篇：5G模組FM150-NA順利通過FCC/IC/PTCRB認證

下一篇：Git 作業區、快取區以及倉庫內容比較方法