場(chǎng)景：
進(jìn)入搜狗，輸入搜索關(guān)鍵字進(jìn)行搜索
利用多進(jìn)程分布式實(shí)現(xiàn)

成都創(chuàng)新互聯(lián)公司企業(yè)建站,十多年網(wǎng)站建設(shè)經(jīng)驗(yàn)，專注于網(wǎng)站建設(shè)技術(shù)，精于網(wǎng)頁(yè)設(shè)計(jì)，有多年建站和網(wǎng)站代運(yùn)營(yíng)經(jīng)驗(yàn)，設(shè)計(jì)師為客戶打造網(wǎng)絡(luò)企業(yè)風(fēng)格，提供周到的建站售前咨詢和貼心的售后服務(wù)。對(duì)于成都網(wǎng)站設(shè)計(jì)、成都做網(wǎng)站中不同領(lǐng)域進(jìn)行深入了解和探索，創(chuàng)新互聯(lián)在網(wǎng)站建設(shè)中充分了解客戶行業(yè)的需求，以靈動(dòng)的思維在網(wǎng)頁(yè)中充分展現(xiàn)，通過對(duì)客戶行業(yè)精準(zhǔn)市場(chǎng)調(diào)研，為客戶提供的解決方案。

from multiprocessing import Pool
import os, time
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from multiprocessing import Manager, current_process
import traceback

#定義測(cè)試行為函數(shù)：此處為打開搜狗搜索內(nèi)容
def node_task(name, lock, arg, successTestCases, failTestCases):
    """

    :param name: 執(zhí)行進(jìn)程名
    :param lock:進(jìn)程間的共享資源鎖
    :param arg:node節(jié)點(diǎn)計(jì)算機(jī)、瀏覽器字典 ，如：{"node": "http://127.0.0.1:6666/wd/hub", "browserName": "chrome"}
    :param successTestCases:成功執(zhí)行用例列表
    :param failTestCases:失敗用例列表
    :return:返回成功執(zhí)行、失敗執(zhí)行的用例列表
    """
    procName = current_process().name

    print("當(dāng)前進(jìn)程名:",procName)
    time.sleep(1.2)
    #獲取節(jié)點(diǎn)計(jì)算機(jī)地址
    node_host = arg["node"]
    #獲取瀏覽器
    browser = arg["browserName"]
    print(arg["node"])
    print(arg["browserName"])
    print('Run task %s (%s)...\n' % (name, os.getpid()))
    #獲取用例執(zhí)行的開始時(shí)間
    start = time.time()
    #獲取driver對(duì)象
    driver = webdriver.Remote(
        #節(jié)點(diǎn)計(jì)算機(jī)
        command_executor="%s" %arg["node"],
        desired_capabilities={
            #瀏覽器
            "browserName": "%s" %arg["browserName"],
            "video": "True",
            "platform": "WINDOWS"})

    try:
        driver.maximize_window()
        driver.get("http://www.sogou.com")
        assert "搜狗" in driver.title
        element = driver.find_element_by_id("query")
        element.send_keys("測(cè)試開發(fā)")
        element.send_keys(Keys.RETURN)
        time.sleep(3)
        assert "測(cè)試開發(fā)" in driver.page_source
        #獲取共享鎖
        lock.acquire()
        #用例執(zhí)行成功，用例加入成功列表
        successTestCases.append("TestCase: " + str(name))
        #釋放共享鎖
        lock.release()
        print("TestCase: " + str(name) + " done!")

    except AssertionError as e:
        #斷言失敗，打印異常信息
        print("AssertionError occur!" " testCase " + str(name))
        print(traceback.print_exc())
        #保存異?，F(xiàn)場(chǎng)圖片
        driver.save_screenshot("e:\\screenshot" + str(name) + ".png")
        #獲取共享鎖
        lock.acquire()
        #存儲(chǔ)失敗用例信息
        failTestCases.append("TestCase: " + str(name))
        #釋放共享鎖
        lock.release()
        print("測(cè)試用例執(zhí)行失敗")
    except Exception as e:
        print("Exception occur!")
        print(traceback.print_exc())
        driver.save_screenshot("e:\\screenshot" + str(name) + ".png")
        #獲取共享鎖，存儲(chǔ)失敗用例信息，并釋放鎖
        with lock:
            failTestCases.append("TestCase: " + str(name))
        print("測(cè)試 用例執(zhí)行失敗")

    finally:
        #退出驅(qū)動(dòng)，并關(guān)閉所有窗口
        driver.quit()
    end = time.time()
    #打印用例執(zhí)行時(shí)間
    print("Tast %s run %.2f seconds." % (name, (end - start)))

#封裝多進(jìn)程執(zhí)行函數(shù)
def run(nodeSeq):
    """
    :param nodeSeq: 節(jié)點(diǎn)機(jī)器和瀏覽器字典，列表
    :return: 返回成功、失敗用例列表successTestCases, failTestCases
    """
    #定義共享manager對(duì)象
    manager = Manager()
    #創(chuàng)建進(jìn)程間共享的用例執(zhí)行成功列表
    successTestCases = manager.list([])
    # 創(chuàng)建進(jìn)程間共享的用例執(zhí)行失敗列表
    failTestCases = manager.list([])

    #創(chuàng)建一個(gè)共享資源鎖，各進(jìn)程共享
    lock = manager.Lock()
    #打印父進(jìn)程ID
    print("Parent process %s." % os.getpid())
    #創(chuàng)建包含3個(gè)進(jìn)程的進(jìn)程池
    p = Pool(processes=3)
    #獲取用例數(shù)量
    testCaseNumber = len(nodeSeq)
    #循環(huán)索引遍歷用例列表，多進(jìn)程執(zhí)行node_task函數(shù)（搜狗搜索）
    for i in range(testCaseNumber):
        p.apply_async(node_task, args=(i + 1, lock, nodeSeq[i], successTestCases, failTestCases))

    print("Waiting for all subprocesses done...")
    #關(guān)閉p
    p.close()
    #等待各個(gè)子進(jìn)程執(zhí)行結(jié)束
    p.join()

    return successTestCases, failTestCases

#封裝寫測(cè)試結(jié)果函數(shù)
def resultReport(testCaseNumber, successTestCases, failTestCases):
    """

    :param testCaseNumber: 用例個(gè)數(shù)
    :param successTestCases: 成功執(zhí)行列表
    :param failTestCases: 失敗執(zhí)行列表
    :return: 無(wú)
    """
    print("測(cè)試報(bào)告: \n")
    print("共執(zhí)行測(cè)試用例:" + str(testCaseNumber) + "個(gè)\n")
    print("執(zhí)行成功的測(cè)試用例: " + str(len(successTestCases)) + "個(gè)")
    if len(successTestCases) > 0:
        for t in successTestCases:
            print(t)
    else:
        print("沒有執(zhí)行成功的測(cè)試用例")
    print("執(zhí)行失敗的測(cè)試用例: " + str(len(failTestCases)) + "個(gè)")
    if len(failTestCases) > 0:
        for t in failTestCases:
            print(t)
    else:
        print("沒有執(zhí)行失敗的測(cè)試用例")

if __name__ == "__main__":
    nodeList = [
        {"node": "http://127.0.0.1:6666/wd/hub", "browserName": "internet explorer"},
        {"node": "http://127.0.0.1:6666/wd/hub", "browserName": "chrome"},
        {"node": "http://127.0.0.1:6666/wd/hub", "browserName": "firefox"}]

    testCaseNumber = len(nodeList)
    #執(zhí)行多進(jìn)程函數(shù)
    successTestCases, failTestCases = run(nodeList)
    print("All processes done")
    #寫測(cè)試結(jié)果
    resultReport(testCaseNumber, successTestCases, failTestCases)

新聞標(biāo)題：多進(jìn)程執(zhí)行分布式自動(dòng)化測(cè)試
當(dāng)前URL：http://chinadenli.net/article38/gicesp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供企業(yè)網(wǎng)站制作、App設(shè)計(jì)、網(wǎng)站設(shè)計(jì)公司、品牌網(wǎng)站設(shè)計(jì)、全網(wǎng)營(yíng)銷推廣、網(wǎng)站內(nèi)鏈

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)