标签:01 入门 flask db Char Column Integer id String
前言
在去年的年中,我一时冲动写了一个基于flask开发的测试平台,然后把服务托管在了腾讯云上,本来想是写文章分享的怎么开发的,但是一直没有写。如今一年已经过去了,服务器也是到期了,演示网址也是无法访问了。
正好现在是国庆假期,一天好似虚度人生,就敲起键盘写一点我的这个测试平台。
首先肯定有一个名称,因为我羡慕自由的鸟儿飞翔在天空,所以我给起的名字是flytest
。好了,名称就是这个样子了。继续看吧。
技术栈
本项目是一个前后端不分离的jinja2渲染形式的平台,所以前端页面我使用的是bootstrap4。别问我为啥不用VUE,问就是不会。
环境 |
用途 |
bootstrap4 |
页面显示 |
flask |
后端使用的框架 |
celery |
测试用例执行 |
MySQL |
测试数据的存储 |
redis |
缓存 |
APScheduler |
定时任务开发 |
nginx+gunicorn+supervisor |
环境部署 |
平台结构
测试平台大概分为这么几个模块,每一个模块都有相关的作用,但是每个模块之间又有着很强的关联性。
模块 |
作用 |
项目管理 |
对不同项目进行管理 |
环境管理 |
管理测试环境和生产环境 |
测试管理 |
添加用例并进行测试 |
任务列表 |
测试的任务和任务结果显示 |
定时任务 |
定时任务列表 |
最新报告 |
显示最近的测试结果 |
问题管理 |
显示最新的问题 |
结果趋势 |
通过折线图查看最新的测试情况 |
平台截图
简陋的登录页
注册页面
主界面
测试管理页
流程设计
以下是我对于这个测试平台的流程构想,其中人员管理部分是采用邮箱注册并激活的形式进行处理。
数据库设计
在这个模型文件中使用了两个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)
以上就是模型文件的所有数据了,感觉第一篇应该讲一下怎么搭建环境,好了,先写这么多吧。
标签:01,
入门,
flask,
db,
Char,
Column,
Integer,
id,
String
From: https://www.cnblogs.com/wxhou/p/16751339.html