測試平台系列(74) 測試計划定時執行初體驗

本文轉載自查看原文 2021-11-15 16:07 185 pity接口自動化測試平台/ 自動化測試/ FastApi/ Pity測試平台/ React/ Python

大家好~我是米洛！

我正在從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 PityTestPlanDao.query_test_plan(plan_id)
        if plan is None:
            Executor.log.info(f"測試計划: [{plan_id}]不存在")
            return
        # 設置為running
        plan.state = 1
        await PityTestPlanDao.update_test_plan(plan, plan.update_user)
        # if plan.disabled:
        #     # 說明測試計划已禁用
        #     Executor.log.info(f"測試計划: [{plan.name}]未開啟")
        #     return
        env = list(map(int, plan.env.split(",")))
        case_list = list(map(int, plan.case_list.split(",")))
        await asyncio.gather(
            *(Executor.run_multiple(0, int(e), case_list, mode=1,
                                    plan_id=plan.id, ordered=plan.ordered) for e in env))
        # TODO 后續通知部分
        plan.state = 0
        await PityTestPlanDao.update_test_plan(plan, plan.update_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目錄下新建scheduler.py文件:

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger

from app.utils.executor import Executor


class Scheduler(object):
    scheduler: AsyncIOScheduler = None

    @staticmethod
    def init(scheduler):
        Scheduler.scheduler = scheduler

    @staticmethod
    def configure(**kwargs):
        Scheduler.scheduler.configure(**kwargs)

    @staticmethod
    def start():
        Scheduler.scheduler.start()

    @staticmethod
    def add_test_plan(plan_id, plan_name, cron):
        return Scheduler.scheduler.add_job(func=Executor.run_test_plan, args=(plan_id,),
                                           name=plan_name, id=str(plan_id),
                                           trigger=CronTrigger.from_crontab(cron))

    @staticmethod
    def edit():
        pass

    @staticmethod
    def remove():
        pass

    @staticmethod
    def list():
        job_list = Scheduler.scheduler.get_jobs()
        return job_list

這里我們對scheduler進行了非常簡單的封裝，然后定義了add_test_plan方法。

這個方法可以添加Executor.run_test_plan到定時任務，他接受id，name和cron3個參數，這個數據最終會落到定時任務表里面去。