基于Selenium Grid 分布式管理的appium测试设备系统(二)

from time import sleep
from ddt import ddt, file_data
from appium import webdriver
from common.parame import Parmer
from config import apk_path, login_element
from common.pertools import caijicpu, getnencun
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from  config import  BASH_DIR,perdata
from  common.pytoolsFoYaml import *
import  os
element = read(login_element)
from common.isintsallapp import isinstallapp


@ddt
class testCase(Parmer):
    def setUp(self) -> None:
        self.datafile=os.path.join(BASH_DIR,perdata)
        is_first_install = isinstallapp(devicesName=self.parme['deviceName'], apk_path=apk_path,
                                        apppackage=self.parme['appPackage'])
        desired_caps = self.parme['desired_caps']
        self.driver = webdriver.Remote('http://0.0.0.0:4444/wd/hub', desired_caps)
        sleep(20)
        try:
        # 启动同意用户协议
            agree = element['agreeelement']
            self.driver.find_element_by_id(agree).click()
            if is_first_install:
                # 首次安装需要加载文件
                sleep(10)

        except:
            self.driver.quit()
    def tearDown(self) -> None:
        self.driver.quit()
        sleep(10)

    @file_data("../data/casedata/logincase.yaml")
    def testlogin(self, user, user_password, msg, casename):
        sleep(30)
        self.driver.find_element_by_id(element['opearelement']).click()
        sleep(10)
        WebDriverWait(self.driver, 10, 0.5).until(
            EC.presence_of_element_located((By.ID, element['logintextclik']))).click()
        cpu = caijicpu(packagename=self.parme['appPackage'], devices=self.parme['deviceName'])
        memory = getnencun(packagename=self.parme['appPackage'], devices=self.parme['deviceName'])
        dataper={}
        dataper[casename+"_start"]={'cpu':float(cpu),'mem':float(memory)}
        save(file=self.datafile,data=dataper)
        loc = ("id", element['usernameedit'])
        username = WebDriverWait(self.driver, 10, 0.5).until(EC.presence_of_element_located(loc))
        username.clear()
        username.send_keys(user)
        password = WebDriverWait(self.driver, 10, 0.5).until(
            EC.presence_of_element_located((By.ID, element['usernameedit'])))
        password.clear()
        password.send_keys(user_password)
        sleep(2)
        login = self.driver.find_element_by_id(element['loginbtn'])
        login.click()
        sleep(10)
        try:
            self.driver.find_element_by_id(element['assertelement'])
            self.assertTrue(False, msg=msg)

        except:
            self.assertTrue(True, msg=msg)
        
        cpuend = caijicpu(packagename=self.parme['appPackage'], devices=self.parme['deviceName'])
        memoryend = getnencun(packagename=self.parme['appPackage'], devices=self.parme['deviceName'])
        dataper = {}
        dataper[casename + "_end"] = {'cpu': float(cpuend), 'mem': float(memoryend)}
        save(file=self.datafile, data=dataper)

指定设备，
选择执行的case
执行用例出报告

def run(dev):
    desired_caps = {
        'platformName': 'ANDROID',
        'platformVersion': '10.0',
        'appPackage': apkname,
        'appActivity': get_apk_lautc(apk_path),
        "deviceName":dev,
        "udid":dev,
        "ReSet":True
    }

    test_suit = unittest.TestSuite()
    test_suit.addTest(Parmer.parametrize(testcase_klass=testCase,
                                         parame={'desired_caps':desired_caps,
                                                 'deviceName':dev,
                                                                          'appPackage': apkname,
                                                 'appActivity': get_apk_lautc(apk_path) }))  # 扩展的其他的测试用例均这样添加

    openone = open(dev+"name.html", 'w+')
    '''利用unittest的testsuite来组织测试用例'''
    besautiful = BSTestRunner(title="报告",
                              description="测试报告", stream=openone,
                              trynum=runprojecttrynum, is_show=ISSHOWHISTORY,
                              filepath=BASH_DIR)
    besautiful.run(test_suit)

1.连接设备
2.创建线程
3.增加到线程列表
4.分别启动线程

代码如下

def run_threed():
    get_device = get_devices()
    thread_list = []
    for dev in get_device:
        t = threading.Thread(target=run, args=(dev,))
        thread_list.append(t)
    for t in thread_list:
        t.start()

    for t in thread_list:
        t.join(60)