分享一个我的实现：在seldom框架基础上，设计一个装饰器来执行APP自动化的同时，获取APP的性能数据、录屏以及用例执行结束后的后置操作

[986379041]

现状：
1.我很喜欢seldom框架，让我更多的时间集中于如何实现用例上，不必再去造轮子；
2.公司方面希望能够进行APP的性能测试，主要在于某些功能的耗时。
想法：
1.seldom框架能够支持完成自动化的测试工作，但目前缺乏对性能的采集和录屏分帧计算耗时；
2.让每条用例更丰富的方式考虑利用装饰器来实现；
3.UI自动化主要是耗时密集型场景，所以我选择了协程实现并发；
实现：
1.首先设计一个装饰器架子然后在里面套一个协程即可

#  这里可以加一些用例前置数据准备，比如每次用例的文件夹路径什么的
testcase_name = func.__name__  # 用例名称
testcase_desc = func.__doc__  # 用例描述
testcase_file_name = os.path.split(inspect.getsourcefile(func))[1]  # 获取被装饰函数所在的模块文件路径
testcase_class_name = args[0].__class__.__name__  # 获取被装饰函数所在的类名
testcase_folder_name = os.path.basename(os.path.dirname(os.path.abspath(func.__code__.co_filename)))
confrun.PerfConfig.threadLock = True
do_list = [gevent.spawn(run_testcase, func, *args, **kwargs),  gevent.spawn(start_record, video_path, run_path), gevent.spawn(get_perf)]
gevent.joinall(do_list)  # 这里是执行协程的入口

2.其中的三个方法设计如下：

def run_testcase(func, *args, **kwargs):
    try:
        func(*args, **kwargs)
    except Exception as e:
        common.update_cache('run_error', f"Error in run_testcase: {e}")
    if confrun.PerfConfig.record and confrun.Config.device == confrun.Scene.Android.value:
        time.sleep(1)
        my_u2.stop_recording()
    elif confrun.PerfConfig.record and confrun.Config.device == confrun.Scene.iOS.value:
        log.info('用例结束，修改record录制标识为False')
        confrun.PerfConfig.record = False
        confrun.PerfConfig.temp_iOS_perf.stop()
    confrun.PerfConfig.threadLock = False  # 这标识在控制协程所有任务结束

def get_perf():
    try:
        if confrun.Config.device == confrun.Scene.Android.value:  # 这里我进行了判断当前设备，Android用solox库
            perf = MySoloX(pkgName=confrun.package_name(), platform=confrun.Config.device)
            new_cpu = gevent.spawn(perf.get_cpu)
            new_mem = gevent.spawn(perf.get_mem)
            new_fps = gevent.spawn(perf.get_fps)
            gevent.joinall([new_cpu, new_mem, new_fps])
            cache.set({'cpu_info': new_cpu.value})
            cache.set({'mem_info': new_mem.value})
            cache.set({'fps_info': new_fps.value})
        elif confrun.Config.device == confrun.Scene.iOS.value:  # iOS用tidevice库的方法
            confrun.PerfConfig.temp_iOS_perf = TidevicePerf()
            confrun.PerfConfig.temp_iOS_perf.start()
    except Exception as e:
        confrun.PerfConfig.perf_error = True
        common.update_cache('perf_error', f"Error in get_perf: {e}")

def start_record(video_path, run_path):
    while not os.path.exists(run_path):
        time.sleep(1)
    while confrun.PerfConfig.threadLock:
        if confrun.PerfConfig.record:  # 这里的判断是我需要在用例的指定步骤后穿插开始录屏的标识，这边监听状态
            try:
                if confrun.Config.device == confrun.Scene.iOS.value:  # 这里也是双端用了不同的录屏方式
                    with make_screenrecord(output_video_path=video_path):  # iOS是上下文管理器的方式来控制录屏的启停
                        while confrun.PerfConfig.record:
                            time.sleep(1)
                elif confrun.Config.device == confrun.Scene.Android.value:
                    my_u2.start_recording(video_path)
            except Exception as e:
                common.update_cache('record_error', f"Error in recording: {e}")
            break
        time.sleep(1.5)

大致的设计思路是这样，协程是这样设计，但也有后续的数据处理操作，比如：性能数据绘制图表，录屏进行分帧和图像识别计算耗时，完整的装饰器设计如下：

import os
import time
import gevent
import inspect
import confrun
import statistics
import threading
from enum import Enum
from tools import common
from functools import wraps
from seldom.utils import cache
from seldom.logging import log
from tools.wrapper_perf import MySoloX, TidevicePerf
from tools.wrapper_excel import my_excel

if confrun.Config.device == confrun.Scene.iOS.value:
    from tools.wrapper_wda import make_screenrecord, my_wda

    driver = my_wda
elif confrun.Config.device == confrun.Scene.Android.value:
    from tools.wrapper_u2 import my_u2

    driver = my_u2


class RunType(Enum):
    """
    性能装饰器参数的枚举类型
    """
    NOTHING = 0
    DEBUG = 1
    DURATION = 2
    STRESS = 3


def start_recording():
    """
    改变标记，开始录屏
    :return:
    """
    confrun.PerfConfig.record = True
    driver.sleep(2)


def run_testcase(func, *args, **kwargs):
    try:
        func(*args, **kwargs)
    except Exception as e:
        common.update_cache('run_error', f"Error in run_testcase: {e}")
    if confrun.PerfConfig.record and confrun.Config.device == confrun.Scene.Android.value:
        time.sleep(1)
        my_u2.stop_recording()
    elif confrun.PerfConfig.record and confrun.Config.device == confrun.Scene.iOS.value:
        log.info('用例结束，修改record录制标识为False')
        confrun.PerfConfig.record = False
        confrun.PerfConfig.temp_iOS_perf.stop()
    confrun.PerfConfig.threadLock = False
    confrun.Config.log = False


def get_log(log_path, run_path):
    try:
        confrun.Config.log = True
        while not os.path.exists(run_path):
            time.sleep(1)
        if confrun.Config.device == confrun.Scene.Android.value:  # 因为iOS无法获取app日志，暂时不需要执行iOS日志
            driver.write_log(log_path)
    except Exception as e:
        cache.set({'log_error': f"Error in get_log: {e}"})


def get_perf():
    try:
        if confrun.Config.device == confrun.Scene.Android.value:
            perf = MySoloX(pkgName=confrun.package_name(), platform=confrun.Config.device)
            new_cpu = gevent.spawn(perf.get_cpu)
            new_mem = gevent.spawn(perf.get_mem)
            new_fps = gevent.spawn(perf.get_fps)
            gevent.joinall([new_cpu, new_mem, new_fps])
            cache.set({'cpu_info': new_cpu.value})
            cache.set({'mem_info': new_mem.value})
            cache.set({'fps_info': new_fps.value})
        elif confrun.Config.device == confrun.Scene.iOS.value:
            confrun.PerfConfig.temp_iOS_perf = TidevicePerf()
            confrun.PerfConfig.temp_iOS_perf.start()
    except Exception as e:
        confrun.PerfConfig.perf_error = True
        common.update_cache('perf_error', f"Error in get_perf: {e}")


def start_record(video_path, run_path):
    while not os.path.exists(run_path):
        time.sleep(1)
    while confrun.PerfConfig.threadLock:
        if confrun.PerfConfig.record:
            try:
                if confrun.Config.device == confrun.Scene.iOS.value:
                    with make_screenrecord(output_video_path=video_path):
                        while confrun.PerfConfig.record:
                            time.sleep(1)
                elif confrun.Config.device == confrun.Scene.Android.value:
                    my_u2.start_recording(video_path)
            except Exception as e:
                common.update_cache('record_error', f"Error in recording: {e}")
            break
        time.sleep(1.5)


def my_perf(run_type, duration_times=confrun.PerfConfig.duration_times, set_max_mem: int = 800,
            set_max_duration: int = 100):
    """
    统计性能数据的装饰器
    :param run_type: NOTHING：什么都不做 DEBUG：只录屏分帧 DURATION：录屏分帧计算耗时 STRESS：多次压测
    :param duration_times: 只有在录屏分帧求耗时模式下才用传次数
    :param set_max_mem: 最大内存(仅压测生效)
    :param set_max_duration: 最大耗时，如果超出则说明耗时异常
    :return:
    """

    def my_decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # --------------------------初始化--------------------------
            testcase_name = func.__name__  # 用例名称
            testcase_desc = func.__doc__  # 用例描述
            testcase_file_name = os.path.split(inspect.getsourcefile(func))[1]  # 获取被装饰函数所在的模块文件路径
            testcase_class_name = args[0].__class__.__name__  # 获取被装饰函数所在的类名
            testcase_folder_name = os.path.basename(os.path.dirname(os.path.abspath(func.__code__.co_filename)))
            cache.set({'testcase_name': testcase_name})
            cache.set({'testcase_class_name': testcase_class_name})
            run_times = 1
            if run_type == RunType.NOTHING:
                log.info(f'纯跑啥也不做模式:{testcase_name}')
            elif run_type == RunType.DEBUG:
                log.info(f'仅仅录屏分帧模式:{testcase_name}')
            elif run_type == RunType.DURATION:
                log.info(f'录屏分帧耗时模式:{testcase_name}')
                run_times = duration_times
            elif run_type == RunType.STRESS:
                log.info(f'仅压测不录屏模式:{testcase_name}')
            else:
                raise ValueError
            folder_time = common.get_current_time(time_type="s")  # 以当前时间来命名文件夹
            testcase_base_path = os.path.join(confrun.output_folder(), testcase_folder_name, testcase_class_name,
                                              testcase_name)
            start_path = os.path.join(testcase_base_path, 'start.jpg')  # 开始帧的位置
            stop_path = os.path.join(testcase_base_path, 'stop.jpg')  # 结束帧的位置
            common.create_folder(testcase_base_path)
            duration_list = []
            confrun.Config.report_photo = []
            cpu_base64_list = []
            mem_base64_list = []
            fps_base64_list = []
            bat_base64_list = []
            flo_base64_list = []
            start_frame_list = []
            stop_frame_list = []
            testcase_assert = True
            for i in range(run_times):
                run_path = os.path.join(testcase_base_path, folder_time)
                video_path = os.path.join(run_path, f'{testcase_name}_{i}.mp4')  # 录屏文件路径
                log_path = os.path.join(run_path, f'log_{testcase_name}_{i}.txt')
                cache.set({'run_error': []})
                if run_type in [RunType.DEBUG, RunType.DURATION]:
                    frame_path = os.path.join(testcase_base_path, folder_time, f'{testcase_name}_frame_{i}')  # 分帧
                    common.create_folder(frame_path)
                if run_type in [RunType.DURATION, RunType.STRESS]:
                    perf_path = os.path.join(testcase_base_path, folder_time, f'{testcase_name}_jpg_{i}')  # 性能图表
                    common.create_folder(perf_path)
                else:
                    common.create_folder(run_path)
                confrun.PerfConfig.threadLock = True
                confrun.PerfConfig.record = False
                confrun.PerfConfig.perf_error = False

                # --------------------------用例执行--------------------------
                log_thread = threading.Thread(target=get_log, args=(log_path, run_path))
                log_thread.start()
                log.success("日志线程已经启动，继续执行主线程...")
                do_list = [gevent.spawn(run_testcase, func, *args, **kwargs),
                           gevent.spawn(start_record, video_path, run_path)]
                if run_type in [RunType.DURATION, RunType.STRESS]:  # 判断是否开启性能数据获取，开启的话就在协程队列中增加get_perf执行
                    do_list.append(gevent.spawn(get_perf))
                gevent.joinall(do_list)
                run_error = cache.get('run_error')
                if run_error:
                    log.error(run_error)
                    assert False, run_error
                # --------------------------视频分帧--------------------------
                if run_type in [RunType.DEBUG, RunType.DURATION]:
                    log.info("✅正在进行视频分帧")
                    common.extract_frames(video_path, frame_path)
                    log.info("✅视频分帧结束")
                # --------------------------找关键帧--------------------------
                if run_type == RunType.DURATION:
                    log.info("🌝开始寻找最相似的开始帧")
                    start_frame_path = common.find_best_frame(start_path, frame_path)
                    start_frame = int(os.path.split(start_frame_path)[1].split('.')[0][-6:])
                    log.info(f"Start frame:[{start_frame}]")
                    log.info("🌚开始寻找最相似的结束帧")
                    stop_frame_path = common.find_best_frame(stop_path, frame_path, is_start=False)
                    stop_frame = int(os.path.split(stop_frame_path)[1].split('.')[0][-6:])
                    log.info(f"Stop frame:[{stop_frame}]")
                    # --------------------------以下是计算耗时--------------------------
                    duration = round((stop_frame - start_frame) / confrun.PerfConfig.record_fps, 2)
                    log.info("🌈[{0}]功能耗时[{1}]s".format(testcase_name, duration))
                    duration_list.append(duration)
                    start_frame_list.append(common.image_to_base64(start_frame_path))
                    stop_frame_list.append(common.image_to_base64(stop_frame_path))
                    if run_times != 1:
                        driver.app_restart()
                # --------------------------性能图片保存本地并转base64--------------------------
                if run_type in [RunType.DURATION, RunType.STRESS] and confrun.PerfConfig.perf_error is False:
                    # CPU
                    cpu_info = cache.get('cpu_info')
                    cpu_image_path = os.path.join(perf_path, f'{testcase_name}_cpu_{i}.jpg')
                    cpu_base64_list.append(
                        common.draw_chart(cpu_info[1], cpu_info[0], ['appCpuRate', 'sysCpuRate'],
                                          jpg_name=cpu_image_path,
                                          label_title='CPU'))
                    # MEM
                    mem_info = cache.get('mem_info')
                    mem_image_path = os.path.join(perf_path, f'{testcase_name}_mem_{i}.jpg')
                    mem_base64_list.append(
                        common.draw_chart(mem_info[1], mem_info[0], ['totalPass', 'nativePass', 'dalvikPass'],
                                          jpg_name=mem_image_path, label_title='Memory'))
                    # 帧率
                    fps_info = cache.get('fps_info')
                    fps_image_path = os.path.join(perf_path, f'{testcase_name}_fps_{i}.jpg')
                    if fps_info is not None:
                        fps_base64_list.append(
                            common.draw_chart(fps_info[1], fps_info[0], ['fps', 'jank'], jpg_name=fps_image_path,
                                              label_title='Fps'))
                    # 电池
                    battery_info = cache.get('battery_info')
                    bat_image_path = os.path.join(perf_path, f'{testcase_name}_battery_{i}.jpg')
                    if confrun.Config.device == confrun.Scene.Android.value and battery_info is not None:
                        bat_base64_list.append(
                            common.draw_chart(battery_info[1], battery_info[0], ['level', 'temperature'],
                                              jpg_name=bat_image_path, label_title='Battery'))
                    elif confrun.Config.device == confrun.Scene.iOS.value and battery_info is not None:
                        bat_base64_list.append(
                            common.draw_chart(battery_info[1], battery_info[0],
                                              ['temperature', 'current', 'voltage', 'power'],
                                              jpg_name=bat_image_path, label_title='Battery'))
                    # 流量
                    flow_info = cache.get('flow_info')
                    flo_image_path = os.path.join(perf_path, f'{testcase_name}_flow_{i}.jpg')
                    if flow_info is not None:
                        flo_base64_list.append(
                            common.draw_chart(flow_info[1], flow_info[0], ['sendNum', 'recNum'],
                                              jpg_name=flo_image_path,
                                              label_title='Flow'))

            # --------------------------图片回写报告--------------------------
            if run_type in [RunType.DEBUG, RunType.DURATION, RunType.STRESS]:
                photo_list = cpu_base64_list + mem_base64_list + fps_base64_list + flo_base64_list \
                             + bat_base64_list + start_frame_list + stop_frame_list
                confrun.Config.report_photo.extend(photo_list)
                # --------------------------数据回写表格--------------------------
                sheet_name = '响应耗时' if run_type == RunType.DURATION else '压测数据'
                file_class_name = f"{testcase_file_name} --> {testcase_class_name}"
                in_excel_times = confrun.PerfConfig.stress_times if run_type == RunType.STRESS else duration_times
                test_case_data = [folder_time, file_class_name, testcase_name, testcase_desc, confrun.Config.device,
                                  in_excel_times]
                # --------------------------耗时or内存阈值断言--------------------------
                if run_type == RunType.DURATION:
                    max_duration_res = round(statistics.mean(duration_list), 2)
                    log.success("🌈功能平均耗时[{:.2f}]s".format(max_duration_res))
                    if max_duration_res > set_max_duration:
                        max_duration_res = f"平均耗时{max_duration_res},超出阈值{set_max_duration}"
                        testcase_assert = False
                        log.error(max_duration_res)
                    data = test_case_data + [str(duration_list), max_duration_res, cache.get('build')]
                    my_excel.append_data(sheet_name, data)
                    if testcase_assert is False:
                        assert False, max_duration_res
                elif run_type == RunType.STRESS:
                    max_mem_res = max([tup[0] for tup in cache.get('mem_info')[1]])
                    if max_mem_res > set_max_mem:
                        max_mem_res = f"最大内存{max_mem_res}超出阈值{set_max_mem}"
                        testcase_assert = False
                    data = test_case_data + [max_mem_res, cache.get('build')]
                    my_excel.append_data(sheet_name, data)
                    if testcase_assert is False:
                        assert False, max_mem_res

        return wrapper

    return my_decorator

下面再贴一下装饰器后置处理的相关代码：

def image_to_base64(image_path):
    """
    将图片转换成base64格式，处理图片格式为报告需要的格式
    :param image_path: 图片的路径
    :return: base64格式的内容
    """
    with open(image_path, "rb") as image_file:
        image_bytes = image_file.read()
        base64_data = base64.b64encode(image_bytes)
        base64_string = base64_data.decode("utf-8")
        return base64_string


def extract_frames(video_file, output_dir, start_duration=confrun.PerfConfig.frame_second,
                   end_duration=confrun.PerfConfig.frame_second):
    """
    视频分帧
    :param video_file: 视频的路径
    :param output_dir: 输出分帧的文件夹
    :param start_duration: 指定的前5s片段的时长（以秒为单位）
    :param end_duration: 指定的末尾5s片段的时长（以秒为单位）
    :return:
    """
    create_folder(output_dir)

    # 打开视频文件
    cap = cv2.VideoCapture(video_file)
    fps = cap.get(cv2.CAP_PROP_FPS)
    log.info("视频帧率为 {:.2f} 帧/秒".format(fps))

    # 计算帧数和总时长（以秒为单位）
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    total_duration = total_frames / fps

    # 计算前5s片段的开始帧数和结束帧数
    start_frame = 0
    end_frame = int(fps * start_duration)

    # 如果末尾5s片段时长大于总时长，则设置为总时长
    if end_duration > total_duration:
        end_duration = total_duration

    # 计算末尾5s片段的开始帧数和结束帧数
    end_start_frame = total_frames - int(fps * end_duration)
    end_end_frame = total_frames

    # 初始化计数器和当前帧数
    frame_count = 0
    current_frame = 0

    # 循环读取视频帧
    while True:
        # 读取一帧
        ret, frame = cap.read()

        # 如果未读取到帧，则结束循环
        if not ret:
            break

        # 如果当前帧数在前5s或末尾5s范围内，则进行保存
        if start_frame <= current_frame < end_frame or end_start_frame <= current_frame < end_end_frame:
            # 生成输出文件名
            output_file = os.path.join(output_dir, f"frame_{frame_count:06d}.jpg")

            # 保存帧为图像文件
            cv2.imwrite(output_file, frame)

        # 更新计数器
        frame_count += 1

        # 更新当前帧数
        current_frame += 1

    # 释放视频对象
    cap.release()


def frame_num():
    """视频前后各分帧的数量，比如：5s × 45fps = 225"""
    return confrun.PerfConfig.frame_second * confrun.PerfConfig.record_fps


def calculate_hash(image, hash_size=128):
    """
    计算输入图像的哈希值
    :param image: 输入图像，BGR格式的ndarray数组
    :param hash_size: 哈希值大小，默认为128，会影响识别的准确度
    :return: 计算得到的哈希值
    """
    # 将图像调整为指定大小，并转换为灰度图像
    image = cv2.resize(image, (hash_size, hash_size))
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

    # 计算均值
    mean = np.mean(gray)

    # 生成哈希值
    hash = []
    for i in range(hash_size):
        for j in range(hash_size):
            if gray[i, j] > mean:
                hash.append(1)
            else:
                hash.append(0)

    return hash


def find_best_frame(reference_image_path, image_folder_path, is_start=True):
    """
    在给定图像文件夹中寻找与参考图像相似度最高的图像
    :param reference_image_path: 参考图像的路径，BGR格式的ndarray数组
    :param image_folder_path: 图像文件夹的路径
    :param is_start: 确认是否是寻找开始帧，有处理
    :return: 与参考图像相似度最高的图像的路径
    """
    # 加载参考图像，并计算其哈希值
    reference_image = cv2.imread(reference_image_path)
    reference_hash = calculate_hash(reference_image)

    # 遍历图像文件夹中的所有图像，并计算它们的哈希值
    frame_path_list = os.listdir(image_folder_path)
    best_match = None
    best_distance = float('inf')
    if len(frame_path_list) <= int(frame_num()) * 2:
        start_frame_num = len(frame_path_list) / 2
    else:
        start_frame_num = frame_num()
    for i, filename in enumerate(frame_path_list):
        if is_start and i < start_frame_num:
            if filename.endswith('.jpg'):
                path = os.path.join(image_folder_path, filename)
                image = cv2.imread(path)
                hash = calculate_hash(image)

                # 计算当前图像和参考图像的哈希距离
                distance = sum([a != b for a, b in zip(hash, reference_hash)])
                # 更新最相似的图像
                if distance <= best_distance:
                    best_match = path
                    best_distance = distance

        elif not is_start and i >= start_frame_num:
            if filename.endswith('.jpg'):
                path = os.path.join(image_folder_path, filename)
                image = cv2.imread(path)
                hash = calculate_hash(image)

                # 计算当前图像和参考图像的哈希距离
                distance = sum([a != b for a, b in zip(hash, reference_hash)])
                # 更新最相似的图像
                if distance < best_distance:
                    best_match = path
                    best_distance = distance

    # 输出最相似的图像路径
    return best_match


def draw_chart(data_list, x_labels, line_labels, jpg_name='my_plt.jpg',
               label_title='Performance Chart',
               x_name='Time', y_name='Value'):
    """
    绘制性能数据折线图
    :param data_list: 表格绘制数据，列表嵌套元组
    :param x_labels: X轴的坐标值
    :param line_labels: 每条折线图名称的描述
    :param jpg_name: 图片名称
    :param label_title: 表格title
    :param x_name: X轴含义
    :param y_name: Y轴含义
    :return: 图片名称
    """
    matplotlib.use('TkAgg')  # 使用 Tkinter 后端
    # 将元组列表中的数据按列提取到列表中
    data_cols = list(zip(*data_list))
    # 将时间字符串转换为 datetime 对象
    x_labels = [parser.parse(x) for x in x_labels]
    # 动态调整X轴的日期间隔和显示格式
    x_len = len(x_labels)
    x_locator = mdates.AutoDateLocator(minticks=x_len, maxticks=x_len)
    x_formatter = mdates.DateFormatter('%H:%M:%S')
    # 设置绘图窗口大小
    width = max(x_len, 25)
    plt.figure(figsize=(width, 10))
    # 设置 x 轴的日期定位器和格式化器
    plt.gca().xaxis.set_major_locator(x_locator)
    plt.gca().xaxis.set_major_formatter(x_formatter)
    # 针对每列数据绘制一条折线
    for i, data in enumerate(data_cols):
        plt.plot(x_labels, data, label=line_labels[i])
        for x, y in zip(x_labels, data):
            label = f'{y:.1f}'
            plt.annotate(label, xy=(x, y), xytext=(0, 5), textcoords='offset points')
    plt.xticks(rotation=45, ha='right')
    # 添加标题、标签和图例
    plt.title(label_title)
    plt.xlabel(x_name)
    plt.ylabel(y_name)
    plt.legend()
    # 保存为jpg格式图片
    plt.savefig(jpg_name, format='jpg')
    plt.close()  # 关闭图形窗口
    return image_to_base64(jpg_name)

最后还有一个最重要的，在seldom框架执行的最开始处需要处理一下：

from gevent import monkey  # 引入协程的monkey库

monkey.patch_all()  # 统一修改全局协程等待的耗时为协程的耗时

代码还是不全，但主要的都在这里，有需要或者也是一起做APP性能的同学可以一起交流一起学习~