使用flask开发测试平台01-模型设计

编程语言2024-04-28 18:28:54

前言#

在去年的年中，我一时冲动写了一个基于flask开发的测试平台，然后把服务托管在了腾讯云上，本来想是写文章分享的怎么开发的，但是一直没有写。如今一年已经过去了，服务器也是到期了，演示网址也是无法访问了。

首先肯定有一个名称，因为我羡慕自由的鸟儿飞翔在天空，所以我给起的名字是flytest。好了，名称就是这个样子了。继续看吧。

技术栈#

本项目是一个前后端不分离的jinja2渲染形式的平台，所以前端页面我使用的是bootstrap4。别问我为啥不用VUE，问就是不会。

环境	用途
bootstrap4	页面显示
flask	后端使用的框架
celery	测试用例执行
MySQL	测试数据的存储
redis	缓存
APScheduler	定时任务开发
nginx+gunicorn+supervisor	环境部署

平台结构#

测试平台大概分为这么几个模块，每一个模块都有相关的作用，但是每个模块之间又有着很强的关联性。

模块	作用
项目管理	对不同项目进行管理
环境管理	管理测试环境和生产环境
测试管理	添加用例并进行测试
任务列表	测试的任务和任务结果显示
定时任务	定时任务列表
最新报告	显示最近的测试结果
问题管理	显示最新的问题
结果趋势	通过折线图查看最新的测试情况

平台截图#

注册页面

使用flask开发测试平台01-模型设计,使用flask开发测试平台01-模型设计_软件测试,第1张

主界面

使用flask开发测试平台01-模型设计,使用flask开发测试平台01-模型设计_软件测试_02,第2张

测试管理页

使用flask开发测试平台01-模型设计,使用flask开发测试平台01-模型设计_软件测试_03,第3张

流程设计#

以下是我对于这个测试平台的流程构想，其中人员管理部分是采用邮箱注册并激活的形式进行处理。

数据库设计#

在这个模型文件中使用了两个flask第三方库，分别是flask-login和flask-avatars。

# 需要做这样的导入
from flask_login import UserMixin, current_user
from flask_avatars import Identicon

用户表

字段名称	类型	解释	关系
id	Integer	主键
email	Char(128)	邮箱地址
username	Char(128)	用户名称
password_hash	Char(128)	密码
avatar_s	Char(64)	头像
avatar_m	Char(64)	头像
avatar_l	Char(64)	头像
is_acticed	Boolean	是否激活
product			关联项目表
apitest			关联测试表

产品管理表

字段名称	类型	解释	关系
name	Char(128)	产品名称
desc	Char(512)	产品描述
tag	Char	产品类型
user_id	Integer	产品负责人

测试地址管理表

字段名称	类型	解释	关系
name	Char(128)	地址名称
url	Char(512)	地址
product_id	Integer	产品ID

测试表

字段名称	类型	解释	关系
name	Char(128)	地址名称
results	Integer	测试结果
task_id	Char(256)	任务ID
user_id	Integer	用户ID

测试步骤

字段名称	类型	解释	关系
name	Char(128)	请求名称
method	Char(16)	请求方法
route	Char(512)	请求路径
headers	Text	请求头
request_data	Text	请求数据
expected_result	Char(512)	预期值
expected_regular	Char(512)	预期正则
request_extract	Char(512)	请求验证
response_extract	Char(512)	返回验证
results	Text(2048)	结果

报告

字段名称	类型	解释	关系
types	Integer	1普通任务2定时任务
task_id	Char(256)	任务ID
name	Char(256)	名称
result	Char(2048)	结果

问题库

字段名称	类型	解释	关系
task_id	Char(256)	任务ID
casename	Char(256)	用例名称
stepname	Char(512)	步骤名称
request	Text	请求
detail	Text	详情
level	Char(10)	级别

任务结果

字段名称	类型	解释	关系
task_id	Char(256)	任务ID
name	Char(512)	任务名称
hostname	Char(512)	主机名称
params	Char(512)	任务参数
result	Text	结果
traceback	Text	异常详情

定时任务

字段名称	类型	解释	关系
task_id	Char(256)	任务ID
test_name	Char(128)	任务名称
func_name	Char(256)	任务执行函数名
trigger	Char(64)	执行类型
args	Char(128)	参数
kwargs	Char(128)	关键字参数
max_instances
times
misfire_grace_time
next_run_time	Datetime	下次运行时间
start_date	Date	开始日期
is_active	Boolean	是否激活
product_id	Integer	项目ID

以上我们把表结构大概设计好了，看看实际的sqlalchemy模型文件吧：

from datetime import datetime

from werkzeug.security import generate_password_hash, check_password_hash
from flask import session, current_app
from flask_login import UserMixin, current_user
from flask_avatars import Identicon
from app.extensions import db, cache


class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(128), unique=True)
    username = db.Column(db.String(128))
    password_hash = db.Column(db.String(128))
    avatar_s = db.Column(db.String(64))
    avatar_m = db.Column(db.String(64))
    avatar_l = db.Column(db.String(64))
    created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated = db.Column(db.DateTime, onupdate=datetime.utcnow,
                        default=datetime.utcnow)
    is_deleted = db.Column(db.Boolean, default=False, nullable=False)
    is_active = db.Column(db.Boolean, default=False, nullable=False)

    products = db.relationship('Product', back_populates='user')
    apitests = db.relationship('Apitest', back_populates='user')

    def __init__(self, *args, **kwargs):
        super(User, self).__init__(*args, **kwargs)
        self.generate_avatar()

    def generate_avatar(self):
        avatar = Identicon()
        filenames = avatar.generate(text=self.email)
        self.avatar_s = filenames[0]
        self.avatar_m = filenames[1]
        self.avatar_l = filenames[2]
        db.session.commit()

    @property
    def password(self):
        return AttributeError("password is only readable")

    @password.setter
    def password(self, pwd):
        self.password_hash = generate_password_hash(pwd)

    def verify_password(self, pwd):
        return check_password_hash(self.password_hash, pwd)

    def __repr__(self):
        return self.username

    def __str__(self) -> str:
        return self.username


class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128))
    desc = db.Column(db.Text)
    tag = db.Column(db.String(32))
    created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated = db.Column(db.DateTime, onupdate=datetime.utcnow,
                        default=datetime.utcnow)
    is_deleted = db.Column(db.Boolean, default=False, nullable=False)

    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    user = db.relationship('User', back_populates='products')

    apiurls = db.relationship('Apiurl', back_populates='product')
    apitests = db.relationship('Apitest', back_populates='product')
    reports = db.relationship('Report', back_populates='product')
    bugs = db.relationship('Bug', back_populates='product')
    works = db.relationship('Work', back_populates='product')

    def __repr__(self):
        return '<Product %s>' % self.name

    @staticmethod
    def get_product(pk):
        if pk is not None:
            session["current_product"] = pk
        pk = session.get("current_product", None)
        product = Product.query.filter_by(id=pk, user=current_user, is_deleted=False).one_or_none() or Product.query.filter_by(user=current_user, is_deleted=False).first()
        return product


class Apiurl(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128))
    url = db.Column(db.String(512))
    created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated = db.Column(db.DateTime, onupdate=datetime.utcnow,
                        default=datetime.utcnow)
    is_deleted = db.Column(db.Boolean, default=False, nullable=False)

    product_id = db.Column(db.Integer, db.ForeignKey('product.id'))
    product = db.relationship('Product', back_populates="apiurls")
    apisteps = db.relationship('Apistep', back_populates='apiurl')

    def __repr__(self):
        return '<Url %s>' % self.name


class Apitest(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64))
    results = db.Column(db.Integer, default=-1)
    task_id = db.Column(db.String(255), index=True)
    created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated = db.Column(db.DateTime, onupdate=datetime.utcnow,
                        default=datetime.utcnow)
    is_deleted = db.Column(db.Boolean, default=False, nullable=False)

    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    user = db.relationship('User', back_populates='apitests')
    product_id = db.Column(db.Integer, db.ForeignKey('product.id'))
    product = db.relationship('Product', back_populates='apitests')
    apisteps = db.relationship('Apistep', back_populates='apitest')

    def __repr__(self):
        return '<ApiTest %s>' % self.name


class Apistep(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128))
    method = db.Column(db.String(16))
    route = db.Column(db.String(512))
    headers = db.Column(db.Text)
    request_data = db.Column(db.Text, nullable=True)
    expected_result = db.Column(db.String(512))
    expected_regular = db.Column(db.String(512), nullable=True)
    request_extract = db.Column(db.String(512))
    response_extract = db.Column(db.String(512))
    status = db.Column(db.Integer, default=-1)
    results = db.Column(db.Text(2048), nullable=True)
    created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated = db.Column(db.DateTime, onupdate=datetime.utcnow,
                        default=datetime.utcnow)
    is_deleted = db.Column(db.Boolean, default=False, nullable=False)

    apiurl_id = db.Column(db.Integer, db.ForeignKey('apiurl.id'))
    apiurl = db.relationship('Apiurl', back_populates='apisteps')
    apitest_id = db.Column(db.Integer, db.ForeignKey('apitest.id'))
    apitest = db.relationship('Apitest', back_populates='apisteps')
    report_id = db.Column(db.Integer, db.ForeignKey("report.id"))
    report = db.relationship('Report')

    def __repr__(self):
        return '<ApiStep %s>' % self.name


class Report(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    types = db.Column(db.Integer, default=1)  # 1是普通任务 2是定时任务
    task_id = db.Column(db.String(256), index=True)
    name = db.Column(db.String(256), default="NULL")
    result = db.Column(db.String(2048))
    status = db.Column(db.Integer, default=-1)
    created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated = db.Column(db.DateTime, onupdate=datetime.utcnow,
                        default=datetime.utcnow)
    is_deleted = db.Column(db.Boolean, default=False, nullable=False)

    apistep = db.relationship('Apistep', uselist=False)
    product_id = db.Column(db.Integer, db.ForeignKey('product.id'))
    product = db.relationship('Product', back_populates='reports')


class Bug(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    task_id = db.Column(db.String(256), index=True)
    casename = db.Column(db.String(256))
    stepname = db.Column(db.String(512))
    request = db.Column(db.Text)
    detail = db.Column(db.Text)
    status = db.Column(db.Integer)
    level = db.Column(db.String(10), default='一般')
    created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated = db.Column(db.DateTime, onupdate=datetime.utcnow,
                        default=datetime.utcnow)
    is_deleted = db.Column(db.Boolean, default=False, nullable=False)

    product_id = db.Column(db.Integer, db.ForeignKey('product.id'))
    product = db.relationship('Product', back_populates='bugs')

    def __repr__(self):
        return '<BUG FOR %S>' % self.stepname


class Work(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    task_id = db.Column(db.String(256), index=True)
    name = db.Column(db.String(512))
    hostname = db.Column(db.String(512))
    params = db.Column(db.String(512))
    status = db.Column(db.Text)
    result = db.Column(db.Text)
    traceback = db.Column(db.Text)
    created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

    product_id = db.Column(db.Integer, db.ForeignKey('product.id'))
    product = db.relationship('Product', back_populates='works')

    def __repr__(self):
        return self.task_id


class CronTabTask(db.Model):
    """定时任务"""
    id = db.Column(db.Integer, primary_key=True)
    task_id = db.Column(db.String(256), index=True)
    test_name = db.Column(db.String(128))
    func_name = db.Column(db.String(256))
    trigger = db.Column(db.String(64))
    args = db.Column(db.String(128))
    kwargs = db.Column(db.String(128))
    max_instances = db.Column(db.Integer)
    times = db.Column(db.String(128))
    misfire_grace_time = db.Column(db.Integer)
    next_run_time = db.Column(db.String(256))
    start_date = db.Column(db.String(256))

    is_active = db.Column(db.Boolean, default=True, nullable=False)
    product_id = db.Column(db.Integer)

以上就是模型文件的所有数据了，感觉第一篇应该讲一下怎么搭建环境，好了，先写这么多吧。

查看全文

https://www.xamrdz.com/lan/5w31937257.html

前言#