Apppium（八）PO代码示例

1、BasePage，所有Page类的父类，主要封装find元素查找方法，配合显示等待，不用每个元素查找都使用一次显示等待，driver初始化

#coding=utf-8

import os
from datetime import datetime

import allure
from appium.webdriver import WebElement
from appium.webdriver.common.touch_action import TouchAction
from appium.webdriver.webdriver import WebDriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait

from config.log_config import logger
from config.global_config import SCREENSHOT_DIR

log=logger()
class BasePage:
    black_list = [
        (By.ID, 'image_cancel'),
    ]
    max = 0

    def __init__(self,driver:WebDriver):
        self.driver=driver

    #封装查找单个方法，使用显示等待，考虑进入app可能会有升级等弹窗，使用blacklist来判断    
    def find_element(self, locator) -> WebElement:
        #todo: 处理弹框 异常处理 动态浮动的元素的处理
        try:
            return WebDriverWait(self.driver,10,0.2).until(expected_conditions.visibility_of_element_located(locator))
            # return self.driver.find_element(*locator)

        except Exception as e:
            # raise e
            #黑名单处理
            if BasePage.max > 3:
                raise e
            BasePage.max = BasePage.max + 1
            for black in self.black_list:
                elements = self.driver.find_elements(*black)
                if len(elements) >= 1:
                    elements[0].click()

            return self.find(locator)

     #封装查找元素方法，返回的是list   
    def find_elements(self, locator) -> WebElement:
        #todo: 处理弹框 异常处理 动态浮动的元素的处理
        try:
            return WebDriverWait(self.driver,10,0.2).until(expected_conditions.visibility_of_any_elements_located(locator))
            # return self.driver.find_element(*locator)

        except Exception as e:
            # raise e
            #黑名单处理
            if BasePage.max > 3:
                raise e
            BasePage.max = BasePage.max + 1
            for black in self.black_list:
                elements = self.driver.find_elements(*black)
                if len(elements) >= 1:
                    elements[0].click()

            return self.find(locator)


    @classmethod
    def id_locator(cls,value):
        return (By.ID,value)

    @classmethod
    def text_locator(cls,value):
        return (By.NAME,value)

    @classmethod
    def xpath_locator(cls,value):
        return (By.XPATH,"//*[@text=%s]"%value)

    @classmethod
    def toast_locator(cls):
        return (By.XPATH,"//*[@class='android.widget.Toast']")

    def lon_press(self,locator,times=None):
        TouchAction(self.driver).long_press(locator,times).perform()

    def get_webview(self,webname):
        webviews=self.driver.contexts
        for web in webviews:
            if webname in web:
                return webname
    def swich_webview(self):
        self.driver.switch_to.context(self.get_webview("xxxx"))
        #切换后可以使用css方式定位
        self.driver.find_element_by_css_selector()


    #截屏
    def screen_shot(self,describ):
        if not os.path.exists(SCREENSHOT_DIR):
            os.mkdir(SCREENSHOT_DIR)
        file_name=SCREENSHOT_DIR+"/{}_{}.png".format((datetime.now()).strftime('%Y-%m-%d %H:%M:%S'),describ)
        self.driver.save_screenshot(file_name)
        with open(file_name,'rb') as f:
            file=f.read()
        allure.attach(file,describ,allure.attachment_type.PNG)
        log.info("截图文件保存在:{}".format(file_name))

2、DriverConfig，driver初始化方法，主要用户appium server启动，driver初始化

#coding=utf-8

import os
import subprocess
from datetime import datetime
from time import sleep

import yaml
from appium import webdriver

from config.log_config import logger
from config.global_config import project_path

log=logger()
class DriverConfig:
    def __init__(self,device_info):
        self.device_info=device_info
        self.system_port=device_info['serverPort']+2000
        cmd="appium -p {0} -bp {1} -U {2} --log-timestamp --local-timezone".format(self.device_info['serverPort'],self.device_info["serverPort"]+2000,
                                                  self.device_info["deviceName"])
        log.info(cmd)
        subprocess.Popen(cmd,shell=True,stdout=open("./logs/{0}_{1}_appium.log".format(datetime.now().strftime("%Y-%m-%d_%H:%M:%S"),device_info["deviceName"]),'a'),stderr=subprocess.STDOUT)
        sleep(5) #防止appium服务没有完全启动就执行测试报错

    def get_driver(self,automationName='appium'):
        # 配置yaml文件路径
        DESIRED_CAPS_PATH=os.path.join(project_path,"yaml","desired_caps.yaml")
        with open(DESIRED_CAPS_PATH,encoding="utf-8") as file:
            data=yaml.load(file,Loader=yaml.FullLoader)

        log.info("读取配置文件成功")
        log.info(data)
        try:
            caps = {}
            caps["platformName"] = data["platformName"]
            caps["deviceName"] = self.device_info["deviceName"]
            caps["appPackage"] = data["appPackage"]
            caps['platformVersion'] = self.device_info["platformVersion"]
            caps["appActivity"] = data["appActivity"]
            # caps['unicodeKeyboard'] = data["unicodeKeyboard"] # 是否支持unicode的键盘。如果需要输入中文，要设置为“true”
            # caps['resetKeyboard'] = data["resetKeyboard"] # 是否在测试结束后将键盘重轩为系统默认的输入法。
            caps['newCommandTimeout'] = data["newCommandTimeout"]# Appium服务器待appium客户端发送新消息的时间。默认为60秒
            caps['systemPort'] = self.system_port  # 重要，不定义会出现socket hang up错误！
            # caps["app"] = data["app"]
            caps["noReset"] = data["noReset"] #不重新安装app
            if automationName!='appium':
                caps['automationName']=automationName
            # 权限弹窗自动处理
            caps["autoGrantPermissions"] = data["autoGrantPermissions"] #处理权限弹窗 True默认授权
            driver=webdriver.Remote("http://localhost:{0}/wd/hub".format(str(self.device_info['serverPort'])), caps)
            # 隐式等待
            # driver.implicitly_wait(IMPLICITLY_WAIT_TIME)
            log.info("启动App成功")
            return driver
        except Exception as e:
            raise e

3.conftest.py,pytest的方法，用于获取driver实例

#coding=utf-8


import pytest

from config.log_config import logger
from config.driver_config import DriverConfig

log=logger()
base_driver=None

def pytest_addoption(parser):
    parser.addoption("--cmdopt", action="store", default="device", help="None")

@pytest.fixture
def cmdopt(request):
    return request.config.getoption("--cmdopt")

@pytest.fixture
def common_driver(cmdopt):
    global base_driver
    if base_driver==None:
        base_driver=DriverConfig(eval(cmdopt))
    driver=base_driver.get_driver()
    yield driver
    # driver.close_app()
    # driver.quit()

4.run.py 测试执行

#coding=utf-8

import multiprocessing
import os
import shutil
from datetime import datetime

import pytest
from multiprocessing import Pool
from config import global_config
from config.log_config import logger
from util.global_manager import GlobalManager

log=logger()

device_infos = [
        {"platformVersion": "7.1.2", "serverPort": 4726, "deviceName": "5483e9c3"},
        # {"platformVersion": "9", "serverPort": 4727, "deviceName": "emulator-5554"}
    ]
def run_pytest(device_info):
    #多设备每个设备生成单独的result
    pytest.main([f"--cmdopt={device_info}","--alluredir","./report/allure_result_{0}".format(device_info['deviceName'])])

def pytest_start():

    with Pool(len(device_infos)) as pool:
        pool.map(run_pytest, device_infos)
        pool.close()
        pool.join()

def close_appium_server():
    for i in device_infos:
        os.system("lsof -n -i:{0}".format(format(i.get('serverPort'))+" | grep LISTEN | awk '{print $2}' | xargs kill"))
def generate_report():
    log.info("生成报告……")
        # if not os.path.exists(global_config.REPORT_RESULT_PATH):
        #     os.mkdir(global_config.REPORT_RESULT_PATH)
    #多设备针对每个设备生成单独的report
    for device in device_infos:
        os.system(f"allure generate {global_config.REPORT_RESULT_PATH}_{device['deviceName']} -o {global_config.REPORT_END_PATH}_{device['deviceName']} --clean")
            # 复制history文件夹，在本地生成趋势图
        REPORT_RESULT_FILES=global_config.REPORT_RESULT_PATH+"_"+device['deviceName']
        files = os.listdir(REPORT_RESULT_FILES)
        result_history_dir = os.path.join(REPORT_RESULT_FILES, "history")
        # 如果不存在则先创建文件夹
        if not os.path.exists(result_history_dir):
            os.mkdir(result_history_dir)
        for file in files:
            shutil.copy(os.path.join(REPORT_RESULT_FILES, file), result_history_dir)



if __name__ == '__main__':
    global_m=GlobalManager()
    global_m.delet_report()
    pytest_start()
    close_appium_server()
    generate_report()