测试平台系列(74) 测试计划定时执行初体验

时间：2021-12-03 作者：we8fans

大家好~我是米洛！

我正在从0到1打造一个开源的接口测试平台, 也在编写一套与之对应的完整教程，希望大家多多支持。

欢迎关注我的公众号测试开发坑货，获取最新文章教程!

回顾

上一节我们设计好了测试计划表并编写了CRUD接口，还没来得及测试，我们就要马不停蹄地编写定时任务相关内容了。今天我们就来编写一个完善的demo，可以定期执行测试计划的用例，并生成报告。

至于后面的通知，要等到以后完善了。

调整Executor类里面的部分方法

之前我们虽然支持了多条case异步运行并写入测试报告，但为了支持测试计划，我们还需要进行一些调整。

修改run_multiple方法

我们的核心还是这个方法，因为里面已经包装好了运行多条case的方法。但我们要做一些改动:

变更参数

首先我们要调整它的参数，因为我们运行case的模式，有很多种，按照之前定义的mode字段:

这个是测试报告表中定义的

调整一下

这样，当我们调用run_mutiple方法的时候，就可以知道case的模式是什么模式了，究竟是隶属于测试计划呢，还是属于普通执行的case。我们调整一下，改成图2的数据。

plan_id可以为空，如果有plan_id的话，可以直接跳转到测试计划相关页面。方便编辑测试计划，也方便排查问题。

ordered这个方法区分同步还是异步，因为我们先前的run_multiple都是默认异步的，但我们测试计划又新增了同步模式，所以需要定义这样一个参数。

修改ReportDao

也是添加这2个参数，透传进来。方便报告里面表明，这是什么模式的执行，以及是否有测试计划id。

调整执行逻辑

根据ordered字段判断是否是同步执行，如果不是则和以前一致，是的话则用for的方式执行，保证用例执行的顺序性。

新增run_test_plan方法

    @staticmethod
    async def run_test_plan(plan_id: int):
        """
        通过测试计划id执行测试计划
        :param plan_id:
        :return:
        """
        plan = await 域名y_test_plan(plan_id)
        if plan is None:
            域名(f"测试计划: [{plan_id}]不存在")
            return
        # 设置为running
        域名e = 1
        await 域名te_test_plan(plan, 域名te_user)
        # if 域名bled:
        #     # 说明测试计划已禁用
        #     域名(f"测试计划: [{域名}]未开启")
        #     return
        env = list(map(int, 域名t(",")))
        case_list = list(map(int, 域名t(",")))
        await 域名er(
            *(域名multiple(0, int(e), case_list, mode=1,
                                    plan_id=域名, ordered=域名red) for e in env))
        # TODO 后续通知部分
        域名e = 0
        await 域名te_test_plan(plan, 域名te_user)

先通过query方法查到测试计划数据，如果plan是None，说明没有这个测试计划，可能被删掉了。

否则我们走执行逻辑: 根据plan的env字段取出环境，并转换为[1, 2]这样的数组。

这里可能有点绕，因为我们数据库存的env和case_list都是这样的形式:

1,2,3,4

通过逗号分开，我们split以后，会得到: ["1", "2", "3", "4"]这样的数组，所以我们需要批量转为int。

最后，由于一般环境之间会隔离，所以我们的ordered根据环境走就可以了，多环境执行我们可以按照异步处理。

所以最后也选用了gather。

编写Scheduler类

我们在utils目录下新建域名文件:

from 域名cio import AsyncIOScheduler
from 域名 import CronTrigger

from 域名utor import Executor


class Scheduler(object):
    scheduler: AsyncIOScheduler = None

    @staticmethod
    def init(scheduler):
        域名duler = scheduler

    @staticmethod
    def configure(**kwargs):
        域名igure(**kwargs)

    @staticmethod
    def start():
        域名t()

    @staticmethod
    def add_test_plan(plan_id, plan_name, cron):
        return 域名job(func=域名test_plan, args=(plan_id,),
                                           name=plan_name, id=str(plan_id),
                                           trigger=域名_crontab(cron))

    @staticmethod
    def edit():
        pass

    @staticmethod
    def remove():
        pass

    @staticmethod
    def list():
        job_list = 域名jobs()
        return job_list

这里我们对scheduler进行了非常简单的封装，然后定义了add_test_plan方法。

这个方法可以添加域名test_plan到定时任务，他接受id，name和cron3个参数，这个数据最终会落到定时任务表里面去。