Appium 併發測試基於unitest

時間 2019-11-17

標籤 appium 併發測試基於 unitest 简体版

原文原文鏈接

前言：html

在迴歸測試階段，UI測試，兼容測試是測試的必要步驟。UI自動化的自己是比較冗餘的測試，可是換個角度思考，UI自動化同時鏈接多臺設備，那麼在迴歸測試時，在同一個腳本下產生的測試結果是很是有價值的。java

不一樣設備在併發下的測試結果能夠爲咱們提供：node

1. 兼容性測試（不一樣的手機品牌，Android版本，分辨率等）python

2. 性能測試（經過安裝Emmagee，監控不一樣手機在同腳本下，性能的變化）web

3. 界面對比（經過圖像識別opencv，截圖對比等查看在相同頁面的變化）shell

思路：windows

1. 啓動多路appium服務session

2. 啓動並鏈接多路手機端多線程

3. 運行並生成測試報告併發

問題：

1. python的unittest框架和java不一樣，不支持參數傳入，能夠經過重寫unittest.TestCase的init添加參數

2. appium 經過命令行啓動，須要安裝非desktop的版本，且最好安裝1.9版本appium，1.0我啓動不了

框架代碼截取：

1. 重寫unittest的初始化函數

class ParametrizedCase(unittest.TestCase):
        def __init__(self, methodName='runTest', param=None):
        super(ParametrizedCase, self).__init__(methodName)
        global devices
        devices = param

        @classmethod
        def setUpClass(cls):
            cls.driver = connect_device(devices)

        @classmethod
        def tearDownClass(cls):
            cls.driver.close_app()
            cls.driver.quit()

2. 封裝啓動appium的服務方法：

基於 appium 的啓動命令

appium  -p  -bp  -U

封裝多線程啓動

class AppiumServer:
    def __init__(self, kwargs=None):
        self.kwargs = kwargs

    def start_server(self):
        """start the appium server
        """
        for i in range(0, len(self.kwargs)):
            cmd = "appium --session-override -p %s -bp %s -U %s" % (
                self.kwargs[i]["port"], self.kwargs[i]["bport"], self.kwargs[i]["devices"])
            print(cmd)
            if platform.system() == "Windows":  # windows下啓動server
                t1 = RunServer(cmd)
                p = Process(target=t1.start())
                p.start()
                while True:
                    print("--------start_win_server-------------")
                    if self.win_is_runnnig("http://127.0.0.1:" + self.kwargs[i]["port"] + "/wd/hub" + "/status"):
                        print("-------win_server_ 成功--------------")
                        break
            else:
                appium = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=1,
                                          close_fds=True)
                while True:
                    appium_line = appium.stdout.readline().strip().decode()
                    time.sleep(1)
                    print("---------start_server----------")
                    if 'listener started' in appium_line or 'Error: listen' in appium_line:
                        print("----server_ 成功---")
                        break

    def win_is_runnnig(self, url):
        """Determine whether server is running
        :return:True or False
        """
        response = None
        time.sleep(1)
        try:
            response = urllib.request.urlopen(url, timeout=5)

            if str(response.getcode()).startswith("2"):
                return True
            else:
                return False
        except URLError:
            return False
        except socket.timeout:
            return False
        finally:
            if response:
                response.close()

    def stop_server(self, devices):
        sysstr = platform.system()

        if sysstr == 'Windows':
            os.popen("taskkill /f /im node.exe")
        else:
            for device in devices:
                # mac
                cmd = "lsof -i :{0}".format(device["port"])
                plist = os.popen(cmd).readlines()
                plisttmp = plist[1].split("    ")
                plists = plisttmp[1].split(" ")
                # print plists[0]
                os.popen("kill -9 {0}".format(plists[0]))

class RunServer(threading.Thread):
    def __init__(self, cmd):
        threading.Thread.__init__(self)
        self.cmd = cmd

    def run(self):
        os.system(self.cmd)

3. 封裝鏈接Android設備的方法：

def connect_device(devices):
    desired_caps = {}
    desired_caps['platformVersion'] = devices["platformVersion"]
    desired_caps['platformName'] = devices["platformName"]
    desired_caps["automationName"] = devices['automationName']
    desired_caps['deviceName'] = devices["deviceName"]
    desired_caps["appPackage"] = devices["appPackage"]
    desired_caps["appActivity"] = devices["appActivity"]
    desired_caps["noReset"] = True
    desired_caps['noSign'] = True
    desired_caps["unicodeKeyboard"] = True
    desired_caps["resetKeyboard"] = True
    desired_caps["systemPort"] = devices["systemPort"]

    # desired_caps['app'] = devices["app"]
    remote = "http://127.0.0.1:" + str(devices["port"]) + "/wd/hub"
    # remote = "http://127.0.0.1:" + "4723" + "/wd/hub"
    driver = webdriver.Remote(remote, desired_caps)
    return driver

4. 多線程啓動服務和多線程鏈接多終端，生成日誌報告

def runnerPool(getDevices):
    devices_Pool = []

    for i in range(0, len(getDevices)):
        _initApp = {}
        _initApp["deviceName"] = getDevices[i]["devices"]
        _initApp["platformVersion"] = getPhoneInfo(devices=_initApp["deviceName"])["release"]
        _initApp["platformName"] = "Android"
        _initApp["port"] = getDevices[i]["port"]
        _initApp["automationName"] = "UiAutomator2"
        _initApp["systemPort"] = getDevices[i]["systemPort"]
        _initApp["appPackage"] = 'cn.vsx.vc'
        _initApp["appActivity"] = '.activity.RegistActivity'
        devices_Pool.append(_initApp)
    print(f'devices pool are {devices_Pool}')
    with ProcessPoolExecutor(len(devices_Pool)) as pool:
        pool.map(runnerCaseApp, devices_Pool)


def runnerCaseApp(devices):
    suite = unittest.TestSuite()
    suite.addTest(ParametrizedCase.parametrize(group_call, param=devices))  # 加入測試類
    now = time.strftime('%Y-%m-%d %H_%M_%S')
    result = BeautifulReport(suite)
    result.report(filename=now + f"_{devices['deviceName'].split(':', 1)[0]}.html", log_path='E:\\TestReports\\',
                  description=f"{devices['deviceName'].split(':', 1)[0]}")


if __name__ == '__main__':
    devices = attached_devices()
    if len(devices) > 0:
        l_devices = []
        for dev in devices:
            app = {}
            app["devices"] = dev
            app["port"] = str(random.randint(4700, 4900))
            app["bport"] = str(random.randint(4700, 4900))
            app["systemPort"] = random.randint(4700, 4900)
            l_devices.append(app)
        print(f'list of server:{l_devices}')
        appium_server = AppiumServer(l_devices)
        appium_server.start_server()
        runnerPool(l_devices)
        appium_server.stop_server(l_devices)
    else:
        print("沒有可用的安卓設備")