pytest多線程與多設備並發appium

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   : [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 命令行版,也可以使用桌面版)

  • 在自動化用例運行之前,必須讓 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"])

到此這篇關於pytest多線程與多設備並發appium的文章就介紹到這瞭。希望對大傢的學習有所幫助,也希望大傢多多支持WalkonNet。

推薦閱讀: