[Python知识库]APP 测试框架二次开发：Python Appium & BeatifulReport

APP 测试框架二次开发：Python, Appium & BeatifulReport

一、需求：

需要对APP功能进行测试，重新封装 Python，Appium & BeatifulReport

二、结构：

• 项目
  • 配置文件：conf
  • 核心模块：core
  • 业务逻辑：logic
  • 脚本：scripts
  • 软件包：apk
  • 截图：img
  • 报告：report
  • 入口：run_script.py
    

三、入口：run_script.py

相关问题：查看 App自动化： 安装/卸载问题 & App自动化2： 安装/卸载问题

# coding=utf-8
import argparse
import os
import threading
import unittest

import requests
from BeautifulReport import BeautifulReport
from core.AdbShell import AdbShell
from core.utils import change_report_format


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("-apk", "--apk-version", default="", help="apk version")
    args = parser.parse_args()
    return args


if __name__ == '__main__':
    apk_dir = os.getenv('APK_DIR')
    args = parse_args()
    apk_version = args.apk_version
    if apk_version:
        res = requests.post(apk_version).content
        apk_version = apk_version.split('/')[-1]
        print("apk_version", apk_version)
        os.environ['APK_VERSION'] = apk_version
        print(apk_version)
        apk = os.path.join(apk_dir, apk_version)
        with open(apk, 'wb') as f:
            f.write(res)
    else:
        apk = os.path.join(apk_dir, os.environ['APK_VERSION'])

    adb_shell = AdbShell()
    device = adb_shell.get_device_name()
    adb_shell.set_device_name(device)
    # 安装App
    package_name = adb_shell.get_target_package()
    if package_name:
        adb_shell.adb_uninstall_package(package_name)
    threads = []
    install = threading.Thread(target=adb_shell.adb_install_package, args=(device, apk))
    protect = threading.Thread(target=adb_shell.adb_tap_until_install_success, args=(2, ))
    threads.append(install)
    threads.append(protect)
    for t in threads:
        t.setDaemon(True)
        t.start()
    t.join()
    os.environ['U_APP_PACKAGE_NAME'] = adb_shell.get_target_package()

    test_suite = unittest.defaultTestLoader.discover('./scripts', pattern='test*.py')
    result = BeautifulReport(test_suite)
    result.report(filename='report', description='UI自动化测试报告', report_dir='report')
    report_file = os.path.join(os.getenv('U_APPMBT_ROOT'), 'report', 'report.html')
    change_report_format(report_file)

四、配置文件：conf

1.config.py：全局参数设置目录、手机和软件包等

# encoding = utf-8
import os
from datetime import datetime

# Project Start Time
os.environ['U_PRJ_START_TIME'] = datetime.now().strftime(' %Y-%m-%d %H:%M:%S')

# Appium Info
# platform name
os.environ['U_APPIUM_PLATFORM_NAME'] = 'Android'
# platform version
os.environ['U_APPIUM_PLATFORM_VERSION'] = ''
# device name
os.environ['U_APPIUM_DEVICE_NAME'] = 'xxx'
# App server platform
os.environ['U_APP_SERVER_PLATFORM'] = 'Appium'
# app package
os.environ['U_APP_PACKAGE_NAME'] = 'com.xxx.xxx'
os.environ['U_APP_MAIN_ACTIVITY'] = 'com.xxx.xxx.activity.MainActivity'
# apk version
os.environ['APK_VERSION'] = 'xxx.apk'
# server ip
os.environ['U_APPIUM_SERVER_IP'] = 'localhost'
# server port
os.environ['U_APPIUM_SERVER_PORT'] = '4723'
# testLogs
os.environ['U_APPMBT_ROOT'] = str(os.getcwd())
# apk directory
os.environ['APK_DIR'] = os.path.join(os.getenv('U_APPMBT_ROOT'), 'apk')
os.environ['U_CURRENT_LOG'] = ''

2. element_json: 存储APP中各元素

{
        "RC":
        {
        "authorization_request_id": "com.android.packageinstaller:id/permission_allow_button",
        "confirm_id": "com.xxx.rocket:id/confirm",
        "notice_id": "android:id/checkbox",
        "ignore_id": "com.xxx.rocket:id/tv_ignore",

        "mine_id": "com.xxx.rocket:id/frame_container_mine",
        "switch_channel_id": "com.xxx.rocket:id/switch_open_double",
        "language_setting_id": "com.xxx.rocket:id/tv_go2_language_setting",
        "home_page_id": "com.xxx.rocket:id/frame_go2_app_intro",
        "user_avatar_id": "com.xxx.rocket:id/iv_user_avatar",
        "taptap_toolbar_id": "com.xxx.rocket:id/toolbar",
        "login_confirm_position_percent": {
                "x": "50",
                "y": "85"
        },
        "login_tips_id": "com.xxx.rocket:id/iv_login_tips",
        "email_name_id":"com.xxx.rocket:id/email_name",
        "witch_mode_id":"com.xxx.rocket:id/switch_mode",
        "email_login_id":"com.xxx.rocket:id/login_register_btn",
        "item_edittext_id":"com.xxx.rocket:id/item_edittext",
        "setting_id": "com.xxx.rocket:id/tv_setting",
        "logout_id": "com.xxx.rocket:id/tv_log_out",
        "logout_confirm_id": "android:id/button1",
        "support_id": "com.xxx.rocket:id/tv_feedback_parent",
        "support_title_id": "com.xxx.rocket:id/tv_title",
        "support_xpath": "//*[@resource-id='app']//android.widget.EditText[1]",
        "support_back_id": "com.xxx.rocket:id/iv_back",

        "discover_id": "com.xxx.rocket:id/frame_container_discover",
        "search_icon_id": "com.xxx.rocket:id/discoveryIndexSearchIcon",
        "search_input_id": "com.xxx.rocket:id/discoverySearchEditTextInput",
        "search_browse_id": "com.xxx.rocket:id/tvGameAction",
        "search_history_clear_id": "com.xxx.rocket:id/discoverySearchIvHistoryClear",
        "cancel_id": "android:id/button2",
        "accept_id": "android:id/button1",
   ......
        }
}

3. report_config.py 报告优化

# encoding = utf-8
report_config ={
    "https://cdn.bootcss.com/bootstrap/3.3.5/css/bootstrap.min.css":
        "https://cdn.bootcdn.net/ajax/libs/twitter-bootstrap/3.3.5/css/bootstrap.min.css",

    "https://cdn.bootcss.com/font-awesome/4.4.0/css/font-awesome.min.css":
        "https://cdn.bootcdn.net/ajax/libs/fontawesome-iconpicker/3.2.0/css/fontawesome-iconpicker.min.css",

    "https://cdn.bootcss.com/animate.css/3.5.2/animate.min.css":
        "https://cdn.bootcdn.net/ajax/libs/animate.css/4.1.1/animate.min.css",

    "https://cdn.bootcss.com/chosen/1.8.2/chosen.css":
        "https://cdn.bootcdn.net/ajax/libs/chosen/1.8.8.rc6/chosen.min.css",

    "https://cdn.bootcss.com/jquery/2.1.4/jquery.min.js":
        "https://cdn.bootcdn.net/ajax/libs/jquery/2.1.4/jquery.min.js",

    "https://cdn.bootcss.com/bootstrap/3.3.5/js/bootstrap.min.js":
        "https://cdn.bootcdn.net/ajax/libs/twitter-bootstrap/3.3.5/js/bootstrap.min.js",

    "https://cdn.bootcss.com/echarts/3.8.5/echarts.min.js":
        "https://cdn.bootcdn.net/ajax/libs/echarts/5.0.2/echarts.common.js",

    "https://cdn.bootcss.com/chosen/1.8.2/chosen.jquery.js":
        "https://cdn.bootcdn.net/ajax/libs/chosen/1.8.8.rc6/chosen.jquery.js"
}

五、核心模块：core：

1. AdbShell.py：ADB 相关

# encoding=utf-8
import os
import time
from conf import config


class AdbShell:

    def __init__(self):
        self.__url = None
        self.__device_name = os.getenv('U_APPIUM_DEVICE_NAME')
        self.__package_name = os.getenv('U_APP_PACKAGE_NAME')

    def adb_install_package(self, device_name=None, apk=None):
        """
        adb安装包文件
        :param apk: apk文件
        :param device_name: device name
        """

        if device_name is None:
            device_name = self.__device_name
        if apk is None:
            apk = os.path.join(os.getenv('APK_DIR'), os.getenv('APK_VERSION'))
        try:
            cmd = 'adb -s {} install -r {}'.format(device_name, apk)
            text = os.popen(cmd)
            content = text.read()
            if 'Success' in content:
                print('Pass: Installed on device {} succeeded. \nVersion: {}'.format(device_name, apk))
        except Exception as e:
            print(str(e))

    def adb_uninstall_package(self, package_name=None):
        """
        adb 卸载包
        :param package_name: 包名
        """
        if package_name is None:
            package_name = self.__package_name
        try:
            cmd = 'adb -s {} uninstall {}'.format(self.__device_name, package_name)
            text = os.popen(cmd)
            content = text.read()
            if 'Success' in content:
                print('Pass: Uninstall {} succeeded.'.format(package_name))
            else:
                print('Fail: Could NOT unintall {}'.format(package_name))
        except Exception as e:
            print(str(e))

    def get_device_name(self):
        """
        获取 device name
        :return: device_name
        """
        device_name = ''
        adb_devices = "adb devices"
        try:
            text = os.popen(adb_devices)
            time.sleep(3)
            content = text.read().strip()
            res = content.splitlines()
            if 'device' not in content:
                print('Error: Could Not get device -> {}'.format(res[-1].split()[1]))
            device_name = res[-1].split()[0]
        except Exception as e:
            if str(e) == 'list index out of range':
                print('Error: Could NOT find device! Please check the phone has been attached to TestBed.')
            else:
                print(str(e))
        return device_name

    def set_device_name(self, device_name=None):
        """
        设置device name
        :param device_name: device name
        :return: 是否成功
        """
        if device_name is None:
            device_name = self.get_device_name()
        if self.__device_name != '' and self.__device_name == device_name:
            print('Device Name:' + self.__device_name)
        elif device_name == '':
            return False
        else:
            self.__device_name = device_name
            os.environ['U_APPIUM_DEVICE_NAME'] = str(device_name)
        return True

    def adb_input_text(self, content):
        try:
            os.popen("adb shell input text {}".format(content))
        except Exception as e:
            print(str(e))

    def adb_tap_until_install_success(self, num):
        for i in range(num):
            self.adb_tap()

    def adb_tap(self):
        time.sleep(5)
        width, height = self.get_device_wm_size()
        width = int(int(width)*0.5)
        height = int(int(height)*0.83)
        try:
            os.popen("adb shell input tap {} {}".format(width, height))
        except Exception as e:
            print(str(e))

    def get_device_wm_size(self):
        """
        获取手机屏幕大小
        :return: size，手机屏幕大小
        """
        try:
            content = os.popen("adb shell wm size")
            size = content.read().strip()
            if "Physical size:" in size:
                width = size.split(":")[1].split("x")[0]
                height = size.split(":")[1].split("x")[1]
                return width, height
        except Exception as e:
            print(str(e))

    def get_target_package(self):
        """
        Get Third-party packages through ADB command.
        """
        try:
            f = os.popen('adb shell pm list package -3')
            for line in f.readlines():
                if 'xindong' in line:
                    package = line.strip().split(':')[1]
                    return package
        except Exception as e:
            print(str(e))

2. JsonConf.py：Json文件处理

# encoding=utf-8
import json
import os
from pathlib import Path


class JsonConf:
    """处理Json文件。"""

    def __init__(self, file_path='server_json', json_conf='conf.json'):
        self.json = None
        self._exec(file_path, json_conf)

    def _exec(self, path, conf):
        """读取json文件。"""
        if Path(path).is_absolute():
            conf_path = Path(conf)
        else:
            dir_path = os.path.dirname((os.path.dirname(__file__)))
            conf_path = Path().joinpath(dir_path, 'conf', path, conf)
        with conf_path.open(encoding='utf-8') as conf:
            self.json = json.loads(conf.read())

    def globals(self, key=None):
        if "Beta" in os.getenv("APK_VERSION"):
            var = 'Beta'
        elif "Alpha" in os.getenv("APK_VERSION"):
            var = 'Beta'
        else:
            var = 'RC'
        if key is None:
            value = self.json
            value = value.get(var, value)
        else:
            try:
                value = self.json[key]
            except KeyError as e:
                raise e
        return value

3. logs.py 日志相关

# encoding = utf-8

import os
import datetime
import xlsxwriter
from conf import config

data = {}

def createLogPath():
    # Create log Direction
    root = os.getenv('U_APPMBT_ROOT')
    # create currDir based on current time.
    currentDir = os.path.join(root, 'logs', datetime.datetime.now().strftime('%Y%m%d_%H%M%S'))
    os.mkdir(currentDir)
    # currentDir soft links to currentDir
    currDir = os.path.join(root, 'logs', 'current')
    os.system('rd -Q ' + '"' + currDir + '"')
    os.system('mklink /D ' + '"' + currDir + '"' + ' ' + '"' + currentDir + '"')
    os.environ['U_CURRENT_LOG'] = str(currentDir)
    print('Current Log Dir: ', os.getenv('U_CURRENT_LOG'))

4. utils.py：工具类

from conf.report_conf import report_config


def change_report_format(report):
	# 报告优化，修改CDN地址加速报告加载
    result = ""
    with open(report, 'r', encoding='utf8') as f:
        content = f.read()
        for k, v in report_config.items():
            content = content.replace(k, v)
        result = content

    with open(report, 'w', encoding='utf8') as f:
        f.write(result)

六、业务逻辑：logic

1. 基础逻辑：Common.py

# encoding=utf-8
import os
import time
from appium.webdriver.common.touch_action import TouchAction
from selenium.webdriver.support.ui import WebDriverWait


class Common:
    def __init__(self, driver):
        self.driver = driver
        self.img_path = os.path.join(os.getenv('U_APPMBT_ROOT'), 'img')

    # 查找元素直到成功
    def find_element(self, id_or_xpath, n=5):
        time.sleep(1)
        for i in range(n):
            try:
                if id_or_xpath == 'goback':
                    self.driver.press_keycode('4')
                    return None
                elif id_or_xpath.startswith('/'):
                    ret = self.driver.find_element_by_xpath(id_or_xpath)
                else:
                    ret = self.driver.find_element_by_id(id_or_xpath)
                return ret
            except Exception as e:
                # print(str(e), id_or_xpath)
                time.sleep(1)
        return False

    def click_element(self, id_or_xpath):
        el = self.find_element(id_or_xpath)
        if el:
            el.click()
            time.sleep(3)

    def input_text(self, id_or_xpath, content):
        self.find_element(id_or_xpath).send_keys(content)
        time.sleep(3)

    def tap_a_point(self, position):
        time.sleep(3)
        width, height = self.get_deivce_size()
        x = int(width*int(position['x'])/100)
        y = int(height*int(position['y'])/100)
        TouchAction(self.driver).press(x=x, y=y).release().perform()

    def get_deivce_size(self):
        width = self.driver.get_window_size()['width']
        height = self.driver.get_window_size()['height']
        return width, height

    def find_toast(self, toast_message):
        message = '//*[@text=\'{}\']'.format(toast_message)
        try:
            toast_element = WebDriverWait(self.driver, 5).until(lambda x: x.find_element_by_xpath(message))
            if toast_message:
                return toast_element.text
        except Exception as e:
            print(str(e), "没有找到 {}".format(toast_message))
            return False

    def screenshot(self, filename):
        if not os.path.exists(self.img_path):
            os.makedirs(self.img_path, exist_ok=True)
        filename = os.path.join(self.img_path, str(filename) + '.png')
        self.driver.save_screenshot(filename)

    def make_xpath(self, content, contains=False):
        if contains:
            return '//*[contains(@text, \'{}\')]'.format(content)
        else:
            return '//*[@text=\'{}\']'.format(content)

    def clear_images(self):
        for root, dirs, files in os.walk(self.img_path):
            for name in files:
                if name.endswith('png'):
                    os.remove(os.path.join(root, name))

2. 登录类：Login.py

# encoding=utf-8
"""
登录页
"""


class Login:
    def __init__(self, comm, conf):
        self.comm = comm
        self.conf = conf.globals()

    # 未登录情况
    def before_login(self):
        res = self.comm.find_element(self.conf['my_game_title_id'])
        if res.text == "我的游戏":
            return True

    # 登录
    def login(self):
        mail = "jp.fake@xxx.com"
        code = "111111"
        xpath = self.comm.make_xpath('尚未登录')
        if not self.comm.find_element(xpath):
            print("已登录!")
            return True
        self.comm.click_element(self.conf['user_avatar_id'])
        self.comm.find_element(self.conf['witch_mode_id'])
        self.comm.click_element(self.conf['witch_mode_id'])
        # 输入邮箱
        self.comm.input_text(self.conf['email_name_id'], mail)
        # 登录
        self.comm.click_element(self.conf['email_login_id'])
        # 输入验证码
        self.comm.input_text(self.conf['item_edittext_id'], code)
        xpath = self.comm.make_xpath('尚未登录')
        if not self.comm.find_element(xpath):
            return True

    # 退出登录
    def logout(self):
        xpath = self.comm.make_xpath('尚未登录')
        if self.comm.find_element(xpath):
            return True
        self.comm.click_element(self.conf['setting_id'])
        self.comm.click_element(self.conf['logout_id'])
        self.comm.click_element(self.conf['logout_confirm_id'])
        if self.comm.find_element(xpath):
            return True

    # 获取用户UID
    def get_uid(self):
        self.comm.click_element(self.conf['login_tips_id'])
        res = self.comm.find_element(self.conf['login_tips_id'])
        if 'ID.' in res.text and res.text[-5].isdigit():
            return True

    # 同意协议
    def authorized(self):
        confirm = self.comm.find_element(self.conf['confirm_id'])
        if confirm:
            self.comm.click_element(self.conf['confirm_id'])
        request = self.comm.find_element(self.conf['authorization_request_id'])
        if request:
            self.comm.click_element(self.conf['authorization_request_id'])
        accept = self.comm.find_element(self.conf['accept_id'])
        if accept:
            self.comm.click_element(self.conf['accept_id'])
            self.comm.click_element(self.conf['notice_id'])
            self.comm.click_element('goback')
        print("授权：用户已经同意协议！")
        return True

    # 更新弹窗
    def update(self):
        ignore = self.comm.find_element(self.conf['ignore_id'])
        if ignore:
            self.comm.click_element(self.conf['ignore_id'])
            print("后台开启更新")
        else:
            print("忽略更新 或 未开启更新")
        return True

…

七、脚本：scripts

# encoding=utf-8

import unittest
from BeautifulReport import BeautifulReport
from appium import webdriver
from logic.Common import Common
from logic.User import User
from conf.config import *


class TapRunner(unittest.TestCase):

    def setUp(self):

        desired_caps = {
            'platformName': os.getenv('U_APPIUM_PLATFORM_NAME'),
            # 此处填写机器的android版本
            'platformVersion': os.getenv('U_APPIUM_PLATFORM_VERSION'),
            # 填写adb devices -l 中显示的第一列的设备号
            'deviceName': os.getenv('U_APPIUM_DEVICE_NAME'),
            'appPackage': os.getenv('U_APP_PACKAGE_NAME'),
            'appActivity': os.getenv('U_APP_MAIN_ACTIVITY'),
            'automationName': 'uiautomator2',
            'app': os.path.join(os.getenv('APK_DIR'), os.getenv('APK_VERSION')),
            'noReset': True,
            'unicodeKeyboard': True,
            'resetKeyboard': True
        }

        server_info = '{}:{}'.format(os.getenv('U_APPIUM_SERVER_IP'), os.getenv('U_APPIUM_SERVER_PORT'))
        self.driver = webdriver.Remote("http://{}/wd/hub".format(server_info), desired_caps)
        self.verificationErrors = []
        self.comm = Common(self.driver)
        self.user = User(self.comm, self.driver)
        self.user.u_clear_images()
        self.user.login.authorized()
        self.user.login.update()

    @BeautifulReport.add_test_img("未登陆打开加速器")
    def test_01_before_login(self):
        """未登陆测试：
            打开App，
            显示“我的游戏”页面
        """
        res = self.user.u_before_login()
        if not res:
            self.user.u_save_image("未登陆")
        self.assertTrue(res, msg="进入我的游戏失败！")

    @BeautifulReport.add_test_img("登陆")
    def test_02_login(self):
        """账号登陆：
            进入“我的”页面，
            邮箱登录加速器
             **目前登录后Toast：无法获取
        """
        res = self.user.u_login()
        print(res)
        if not res:
            self.user.u_save_image("登陆")
        self.assertTrue(res, msg="登陆失败！")

    @BeautifulReport.add_test_img("登出")
    def test_03_logout(self):
        """账号登出：
            进入“我的”页面，
            点击【设置】按钮
            退出账号
        """
        res = self.user.u_logout()
        if not res:
            self.user.u_save_image("登出")
        self.assertTrue(res, msg="登出失败！")

    @BeautifulReport.add_test_img("搜索中文全称")
    def test_05_search(self):
        """搜索游戏：
            进入“搜索”页面，
            游戏名称，
            点击搜索，显示结果
        """
        content = "王者荣耀"
        res = self.user.u_search(content, content)
        if not res:
            self.user.u_save_image("搜索中文全称")
        self.assertTrue(res, msg="搜索失败！")

    @BeautifulReport.add_test_img("列表切换")
    def test_06_discover(self):
        """列表切换：
            进入“发现”页面，
            列表切换，
            正常切换
        """
        content = ["港台", "日本", "韩国", "欧美", "东南亚", "大陆"]
        result = ["台服", "日服", "韩服", "欧美服", "东南亚服", "王者荣耀"]
        res = self.user.u_switch_discovery_list(content, result)
        if not res:
            self.user.u_save_image("列表切换")
        self.assertTrue(res, msg="列表切换失败！")

    def tearDown(self):
        self.driver.quit()


if __name__ == '__main__':
    unittest.main()

八、报告