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

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

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

    print("當前進程名:",procName)
    time.sleep(1.2)
    #獲取節點計算機地址
    node_host = arg["node"]
    #獲取瀏覽器
    browser = arg["browserName"]
    print(arg["node"])
    print(arg["browserName"])
    print(‘Run task %s (%s)...\n‘ % (name, os.getpid()))
    #獲取用例執行的開始時間
    start = time.time()
    #獲取driver對象
    driver = webdriver.Remote(
        #節點計算機
        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("測試開發")
        element.send_keys(Keys.RETURN)
        time.sleep(3)
        assert "測試開發" in driver.page_source
        #獲取共享鎖
        lock.acquire()
        #用例執行成功，用例加入成功列表
        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())
        #保存異常現場圖片
        driver.save_screenshot("e:\\screenshot" + str(name) + ".png")
        #獲取共享鎖
        lock.acquire()
        #存儲失敗用例信息
        failTestCases.append("TestCase: " + str(name))
        #釋放共享鎖
        lock.release()
        print("測試用例執行失敗")
    except Exception as e:
        print("Exception occur!")
        print(traceback.print_exc())
        driver.save_screenshot("e:\\screenshot" + str(name) + ".png")
        #獲取共享鎖，存儲失敗用例信息，並釋放鎖
        with lock:
            failTestCases.append("TestCase: " + str(name))
        print("測試 用例執行失敗")

    finally:
        #退出驅動，並關閉所有窗口
        driver.quit()
    end = time.time()
    #打印用例執行時間
    print("Tast %s run %.2f seconds." % (name, (end - start)))

#封裝多進程執行函數
def run(nodeSeq):
    """
    :param nodeSeq: 節點機器和瀏覽器字典，列表
    :return: 返回成功、失敗用例列表successTestCases, failTestCases
    """
    #定義共享manager對象
    manager = Manager()
    #創建進程間共享的用例執行成功列表
    successTestCases = manager.list([])
    # 創建進程間共享的用例執行失敗列表
    failTestCases = manager.list([])

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

    print("Waiting for all subprocesses done...")
    #關閉p
    p.close()
    #等待各個子進程執行結束
    p.join()

    return successTestCases, failTestCases

#封裝寫測試結果函數
def resultReport(testCaseNumber, successTestCases, failTestCases):
    """

    :param testCaseNumber: 用例個數
    :param successTestCases: 成功執行列表
    :param failTestCases: 失敗執行列表
    :return: 無
    """
    print("測試報告: \n")
    print("共執行測試用例:" + str(testCaseNumber) + "個\n")
    print("執行成功的測試用例: " + str(len(successTestCases)) + "個")
    if len(successTestCases) > 0:
        for t in successTestCases:
            print(t)
    else:
        print("沒有執行成功的測試用例")
    print("執行失敗的測試用例: " + str(len(failTestCases)) + "個")
    if len(failTestCases) > 0:
        for t in failTestCases:
            print(t)
    else:
        print("沒有執行失敗的測試用例")

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)
    #執行多進程函數
    successTestCases, failTestCases = run(nodeList)
    print("All processes done")
    #寫測試結果
    resultReport(testCaseNumber, successTestCases, failTestCases)
 

多進程執行分布式自動化測試