測試平台系列(73) 設計測試計劃功能
大家好~我是
米洛！我正在從0到1打造一個開源的介面測試平台, 也在編寫一套與之對應的完整
教程，希望大家多多支援。歡迎關注我的公眾號
測試開發坑貨，獲取最新文章教程!
回顧
上一節我們簡單介紹了下APScheduler，這一節我們來編寫測試計劃相關內容。
設計測試計劃表
測試計劃，其實也可以叫測試集合，它是我們一組用例的集合。並且有著對應的特徵:
-
定時執行
-
執行完畢後通知方式
-
通過率低於多少發送郵件/釘釘等通知
-
優先順序
-
涵蓋多少case
-
case失敗重試間隔 等等
這裡可能比較怪的就是把測試計劃+測試集合給耦合到一起了。
基於上面的思路，我們就可以來設計測試計劃表了, 具體欄位的含義可以參照注釋。
（數據表的設計不一定完美，後續一般會根據業務需求擴展）
from sqlalchemy import Column, String, TEXT, UniqueConstraint, BOOLEAN, SMALLINT, INT
from sqlalchemy.dialects.mysql import TINYTEXT
from app.models.basic import PityBase
class PityTestPlan(PityBase):
project_id = Column(INT, nullable=False)
# 測試計劃執行環境, 可以多選
env = Column(String(64), nullable=False)
# 測試計劃名稱
name = Column(String(32), nullable=False)
# 測試計劃優先順序
priority = Column(String(3), nullable=False)
# cron表達式
cron = Column(String(12), nullable=False)
# 用例列表
case_list = Column(TEXT, nullable=False)
# 並行/串列(是否順序執行)
ordered = Column(BOOLEAN, default=False)
# 通過率低於這個數會自動發通知
pass_rate = Column(SMALLINT, default=80)
# 通知用戶，目前只有郵箱，後續用戶表可能要完善手機號欄位，為了通知
receiver = Column(TEXT)
# 通知方式 0: 郵件 1: 釘釘 2: 企業微信 3: 飛書 支援多選
msg_type = Column(TINYTEXT)
# 單次case失敗重試間隔，默認2分鐘
retry_minutes = Column(SMALLINT, default=2)
# 測試計劃是否正在執行中
state = Column(SMALLINT, default=0, comment="0: 未開始 1: 運行中")
__table_args__ = (
UniqueConstraint('project_id', 'name', 'deleted_at'),
)
__tablename__ = "pity_test_plan"
def __init__(self, project_id, env, case_list, name, priority, cron, ordered, pass_rate, receiver, msg_type,
retry_minutes, user, state=0, id=None):
super().__init__(user, id)
self.env = ",".join(map(str, env))
self.case_list = ",".join(map(str, case_list))
self.name = name
self.project_id = project_id
self.priority = priority
self.ordered = ordered
self.cron = cron
self.pass_rate = pass_rate
self.receiver = ",".join(map(str, receiver))
self.msg_type = ",".join(map(str, msg_type))
self.retry_minutes = retry_minutes
self.state = state
這裡值得注意的是，我們定義的FORM數據，環境列表env、接收人receiver、用例列表case_list都是數組，我們需要進行一波轉換。
編寫CRUD方法
- 抽離非同步分頁方法和where方法
在DatabaseHelper類中添加pagination方法，接受page和size，session和sql參數，先讀出sql匹配到的總數，如果為0則直接return，否則通過offset和limit獲取到對應分頁的數據。
where方法是用於改進我們平時的多條件查詢，類似這種:
- 編寫新增測試計劃方法
可以看到改造之後，我們只需要調用where方法，不需要寫if name != "":這樣的語句了。
因為我們不允許同一個項目裡面出現同名的測試計劃，所以條件是項目id+name不能重複。
- 編寫增改刪方法
from sqlalchemy import select
from app.models import async_session, DatabaseHelper
from app.models.schema.test_plan import PityTestPlanForm
from app.models.test_plan import PityTestPlan
from app.utils.logger import Log
class PityTestPlanDao(object):
log = Log("PityTestPlanDao")
@staticmethod
async def list_test_plan(page: int, size: int, project_id: int = None, name: str = ''):
try:
async with async_session() as session:
conditions = [PityTestPlan.deleted_at == 0]
DatabaseHelper.where(project_id, PityTestPlan.project_id == project_id, conditions) \
.where(name, PityTestPlan.name.like(f"%{name}%"), conditions)
sql = select(PityTestPlan).where(conditions)
result, total = await DatabaseHelper.pagination(page, size, session, sql)
return result, total
except Exception as e:
PityTestPlanDao.log.error(f"獲取測試計劃失敗: {str(e)}")
raise Exception(f"獲取測試計劃失敗: {str(e)}")
@staticmethod
async def insert_test_plan(plan: PityTestPlanForm, user: int):
try:
async with async_session() as session:
async with session.begin():
query = await session.execute(select(PityTestPlan).where(PityTestPlan.project_id == plan.project_id,
PityTestPlan.name == plan.name,
PityTestPlan.deleted_at == 0))
if query.scalars().first() is not None:
raise Exception("測試計劃已存在")
plan = PityTestPlan(**plan.dict(), user=user)
await session.add(plan)
except Exception as e:
PityTestPlanDao.log.error(f"新增測試計劃失敗: {str(e)}")
raise Exception(f"添加失敗: {str(e)}")
@staticmethod
async def update_test_plan(plan: PityTestPlanForm, user: int):
try:
async with async_session() as session:
async with session.begin():
query = await session.execute(
select(PityTestPlan).where(PityTestPlan.id == plan.id, PityTestPlan.deleted_at == 0))
data = query.scalars().first()
if data is None:
raise Exception("測試計劃不存在")
DatabaseHelper.update_model(data, plan, user)
except Exception as e:
PityTestPlanDao.log.error(f"編輯測試計劃失敗: {str(e)}")
raise Exception(f"編輯失敗: {str(e)}")
@staticmethod
async def delete_test_plan(id: int, user: int):
try:
async with async_session() as session:
async with session.begin():
query = await session.execute(
select(PityTestPlan).where(PityTestPlan.id == plan.id, PityTestPlan.deleted_at == 0))
data = query.scalars().first()
if data is None:
raise Exception("測試計劃不存在")
DatabaseHelper.delete_model(data, user)
except Exception as e:
PityTestPlanDao.log.error(f"刪除測試計劃失敗: {str(e)}")
raise Exception(f"刪除失敗: {str(e)}")
@staticmethod
async def query_test_plan(id: int) -> PityTestPlan:
try:
async with async_session() as session:
sql = select(PityTestPlan).where(PityTestPlan.deleted_at == 0, PityTestPlan.id == id)
data = await session.execute(sql)
return data.scalars().first()
except Exception as e:
PityTestPlanDao.log.error(f"獲取測試計劃失敗: {str(e)}")
raise Exception(f"獲取測試計劃失敗: {str(e)}")
基本思路差不多，老CRUD了！這裡就不多說了，對sqlalchemy的async內容不了解的可以去官網看看demo。
這個query方法，是給定時任務查詢測試計劃數據使用。由於做了軟刪除，會導致經常忘記帶上deleted_at==0的條件。
編寫相關介面（app/routers/testcase/testplan.py）
from fastapi import Depends
from app.dao.test_case.TestPlan import PityTestPlanDao
from app.handler.fatcory import PityResponse
from app.models.schema.test_plan import PityTestPlanForm
from app.routers import Permission
from app.routers.testcase.testcase import router
from config import Config
@router.get("/plan/list")
async def list_test_plan(page: int, size: int, project_id: int = None, name: str = "", user_info=Depends(Permission())):
try:
data, total = await PityTestPlanDao.list_test_plan(page, size, project_id, name)
return PityResponse.success_with_size(PityResponse.model_to_list(data), total=total)
except Exception as e:
return PityResponse.failed(str(e))
@router.get("/plan/insert")
async def insert_test_plan(form: PityTestPlanForm, user_info=Depends(Permission(Config.MANAGER))):
try:
await PityTestPlanDao.insert_test_plan(form, user_info['id'])
return PityResponse.success()
except Exception as e:
return PityResponse.failed(str(e))
@router.get("/plan/update")
async def update_test_plan(form: PityTestPlanForm, user_info=Depends(Permission(Config.MANAGER))):
try:
await PityTestPlanDao.update_test_plan(form, user_info['id'])
return PityResponse.success()
except Exception as e:
return PityResponse.failed(str(e))
@router.get("/plan/delete")
async def delete_test_plan(id: int, user_info=Depends(Permission(Config.MANAGER))):
try:
await PityTestPlanDao.delete_test_plan(id, user_info['id'])
return PityResponse.success()
except Exception as e:
return PityResponse.failed(str(e))
這邊我們把測試計劃的增刪改許可權賦予MANAGER，但其實最好是能給對應的項目經理，不過那樣會稍微複雜點。我們暫時先偷個懶，或許再完善。
今天的內容就分享到這裡，下節會介紹測試計劃如何和APScheduler結合起來，之後便是編寫測試計劃的前端頁面部分，完成定時任務功能。
