PPT
源码地址
训练营目标
- 测试平台的价值体系
- 测试平台的学习路线
- 测试平台实战练习
问题
- 为什么要做测试平台？为什么我们要学习平台开发的技术？怎么样去做测试平台才能达到价值最大化（个人、公司）
- 平台开发技术要学到什么程度？一定要非常深入吗？
- 平台开发需要掌握什么技术栈，如何学习可以提升投入产出比？
测试平台-功能设计
| 功能点 | 描述 | 案例 |
|---|---|---|
| 项目管理 | 对于项目的分类，通常以业务线为维度 | 商城、飞书、企业微信 |
| 测试类型 | 手工测试、自动化测试 | pytest、JUnit、HttpRunner、Postman |
| 测试用例 | 测试用例的信息 | 标题、步骤、预期结果、实际结果 |
| 测试套件 | 用例的集合 | 通常是用例从不同维度组合，冒烟测试用例集，某业务线用例集 |
| 测试装置 | 用例执行前后的准备工作，处理工作 | 用例依赖环境的安装，比如 python requirements |
| 测试 Runner | 测试用例执行器调度器 | 执行环境的管理：性能测试环境、兼容性环境 |
| 测试报告 | 测试结果的体现 | allure 报告、手工报告 |
技术-前端
- 界面展示
- 良好的用户体验
- 数据展示
- 相关技术： VUE、React 等
技术-后端
- 把控业务逻辑
- 数据库交互
- 相关技术：Flask、Django、SpringBoot
启动模块server.py
app = Flask(__name__)
api = Api(app)
CORS(app, supports_credentials=True)
# 用例的命名空间
username = "root"
password = 123456
server = "42.192.73.147:3307"
database = "tpf"
app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{username}:{password}@{server}/{database}?charset=utf8mb4"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
# SQLAlchemy 绑定 app
db = SQLAlchemy(app)
# 为了有类型提示所做的声明
db_session: Session = db.session
def register_router():
# 如果出现循环导入，把导包语句放在方法内执行。并且调用此函数
from controller.testcase_controller import case_ns
from controller.plan_controller import plan_ns
from controller.build_controller import build_ns
api.add_namespace(case_ns, "/testcase")
api.add_namespace(plan_ns, "/plan")
api.add_namespace(build_ns, "/build")
if __name__ == '__main__':
register_router()
app.run(debug=True, port=5001)
实战1
- 根据要求，完成后端架构优化
Do层创建-测试用例表
# 创建用例表
class TestcaseDo(db.Model):
# 表名
__tablename__ = "testcase"
# 用例ID 用例的唯 一标识
id = db.Column(Integer, primary_key=True)
# 用例的标题, 限定 80个字符 ，不为空，并且唯一
name = db.Column(String(80), nullable=False, unique=True)
# 用例的步骤
step = db.Column(String(120))
# 用例的自动化方法
method = db.Column(String(120))
# 备注
remark = db.Column(String(120))
Do层创建-测试计划表
# 创建测试计划表
class PlanDo(db.Model):
# 表名
__tablename__ = "plan"
# 用例ID 用例的唯 一标识
id = db.Column(Integer, primary_key=True)
# 计划名称,限定 80个字符 ，不为空，并且唯一
name = db.Column(String(80), nullable=False, unique=True)
# 绑定的测试用例
# 参数1： 关联的另外一个业务表类名， 参数2： 中间表 ， 参数3： 反射别名
testcases: List[TestcaseDo] = relationship("TestcaseDo", secondary=testcase_plan_rel, backref="plans")
Do层创建-构建记录表
# 创建构建记录表
class BuildDo(db.Model):
# 表名
__tablename__ = "build"
# 用例ID 用例的唯 一标识
id = db.Column(Integer, primary_key=True)
# 计划id，不为空，并且唯一,外键写在多的一方，比如 plan-> build 是一对多，那么外键就在build这方。
plan_id = db.Column(Integer, ForeignKey('plan.id', ondelete='CASCADE'))
# 报告地址,限定 80个字符 ，不为空，并且唯一
report = db.Column(String(80), nullable=False)
# 生成时间格式, 创建时间，通常不需要手动传入，在写入记录的时候自动生成
create_time = Column(DateTime, nullable=True, default=datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
# 参数1： 关联的另外一个业务表类名， 参数2： 反射别名
plan: PlanDo = relationship("PlanDo", backref="build")
Do层创建-多对多（测试用例与测试计划）
# 中间表
# 中间表表名
# 测试用例的外键
# 计划的外键
testcase_plan_rel = db.Table(
'testcase_plan_rel',
# 参数一： 表名_id， 参数二：整型，参数3： 外键字符串('表名.id'， 参数4： 是否为主键)
Column('testcase_id', Integer, ForeignKey('testcase.id', ondelete='CASCADE'), primary_key=True),
Column('plan_id', Integer, ForeignKey('plan.id', ondelete='CASCADE'), primary_key=True))
Do层创建-初始化数据库
from do.build_do import BuildDo
from do.plan_do import PlanDo
from do.testcase_do import TestcaseDo
from do.testcase_plan_rel import testcase_plan_rel
from server import db
if __name__ == '__main__':
db.create_all()
实战2
完成测试计划、测试用例、构建记录的表创建
Dao层创建-测试用例
# Dao 负责和数据库的交互
class TestcaseDao:
"""
TestcaseDao
"""
def get(self, testcase_id) -> TestcaseDo:
"""
根据id返回数据
:return:
"""
# TestcaseDo.query.filter()
return db_session.query(TestcaseDo).filter_by(id=testcase_id).first()
def list(self) -> List[TestcaseDo]:
"""
返回所有
:return:
"""
return db_session.query(TestcaseDo).all()
# 如果涉及到数据的改动，一定要添加commit
def save(self, testcase_do: TestcaseDo):
"""
保存数据
:return:
"""
# testcase = TestCase(**case_data)
db_session.add(testcase_do)
db_session.commit()
return testcase_do.id
def update(self, testcase_do: TestcaseDo):
"""
修改数据
:return:
"""
# add 底层支持 保存 或者更新 _save_or_update_state
db_session.query(TestcaseDo).filter_by(id=testcase_do.id).update(testcase_do.as_dict())
db_session.commit()
return testcase_do.id
def delete(self, testcase_id):
"""
删除数据
"""
db_session.query(TestcaseDo).filter_by(id=testcase_id).delete()
db_session.commit()
def get_by_name(self, name) -> TestcaseDo:
"""
通过用例name查询数据
:return:
"""
return db_session.query(TestcaseDo).filter_by(name=name).first()
Dao层创建-测试计划
# Dao 负责和数据库的交互
class PlanDao:
"""
PlanDao
"""
def get(self, plan_id) -> PlanDo:
"""
根据id返回数据
:return:
"""
return db_session.query(PlanDo).filter_by(id=plan_id).first()
def list(self) -> List[PlanDo]:
"""
返回所有
:return:
"""
return db_session.query(PlanDo).all()
def save(self, plan_do: PlanDo):
"""
保存数据
:return:
"""
db_session.add(plan_do)
db_session.commit()
return plan_do.id
def delete(self, plan_id):
"""
删除数据
"""
db_session.query(PlanDo).filter_by(id=plan_id).delete()
db_session.commit()
def get_by_name(self, name) -> PlanDo:
"""
通过name查询数据
:return:
"""
return db_session.query(PlanDo).filter_by(name=name).first()
Dao层创建-构建记录
# Dao 负责和数据库的交互
class BuildDao:
"""
BuildDao
"""
def get(self, build_id) -> BuildDo:
"""
根据id返回数据
:return:
"""
return db_session.query(BuildDo).filter_by(id=build_id).first()
def list(self) -> List[BuildDo]:
"""
返回所有
:return:
"""
return db_session.query(BuildDo).all()
def save(self, build_do):
"""
保存数据
:return:
"""
db_session.add(build_do)
db_session.commit()
return build_do.id
def delete(self, build_id):
"""
删除数据
"""
db_session.query(BuildDo).filter_by(id=build_id).delete()
db_session.commit()
实战3
完成测试计划、测试用例、构建记录的Dao层代码
Server层创建-测试用例
class TestcaseService:
def get(self, testcase_id):
"""
通过用例id查询数据
:return:
"""
return testcase_dao.get(testcase_id)
def list(self):
"""
返回所有
:return:
"""
return testcase_dao.list()
def save(self, testcase_do: TestcaseDo):
"""
1. 查询数据是否存在
2. 如果存在返回失败
3. 如果不存在，新增
:param testcase_do:
:return:
"""
testcase = self.get_by_name(testcase_do.name)
if not testcase:
return testcase_dao.save(testcase_do)
else:
return False
def update(self, testcase_do: TestcaseDo):
"""
如果测试用例存在才更新，反之返回错误
:param testcase_do:
:return:
"""
testcase = self.get(testcase_do.id)
if not testcase:
return False
else:
# 传入被更新的测试用例的对象
return testcase_dao.update(testcase_do)
def delete(self, testcase_id):
"""
如果测试用例存在才删除，反之返回错误
:param testcase_id: 测试用例删除
:return:
"""
testcase = self.get(testcase_id)
if not testcase:
return False
else:
testcase_dao.delete(testcase_id)
return testcase_id
def get_by_name(self, name):
"""
通过用例name查询数据
:return:
"""
return testcase_dao.get_by_name(name)
Server层创建-测试计划
class PlanService:
def list(self):
"""
查询所有的测试任务
:return:
"""
return plan_dao.list()
def get(self, plan_id):
"""
查询单个的测试任务
:param plan_id:
:return:
"""
return plan_dao.get(plan_id)
def save(self, testcase_id_list, plan_do: PlanDo):
"""
:return: 生成的测试任务的id
"""
testcase = self.get_by_name(plan_do.name)
if not testcase:
# testcase_id_list = [1,2,3] 转换为 [TestcaseDo1， TestcaseDo2]
# plan_do.testcases 要求 的输入格式为 [TestcaseDo1， TestcaseDo2]
testcase_obj_list = [testcase_service.get(testcase_id) for testcase_id in testcase_id_list]
# 多对多情况下建立关联关系，必须使用 实例.关联表属性 = [关联表对象]
plan_do.testcases = testcase_obj_list
return plan_dao.save(plan_do)
else:
return False
def delete(self, plan_id):
"""
存在才删除，反之返回错误
:param plan_id: 计划id
:return:
"""
plan = self.get(plan_id)
if not plan:
return False
else:
plan_dao.delete(plan_id)
return plan_id
def get_by_name(self, name):
"""
通过name查询数据
:return:
"""
return plan_dao.get_by_name(name)
测试用例调度
class JenkinsUtils:
# Jenkins 服务
BASE_URL = "http://42.192.73.147:7080/"
# Jenkins 服务对应的用户名
USERNAME = "admin"
# Jenkins 服务对应的token
PASSWORD = "1141cbca1d1b9ac4781d0c255a5a478fce"
JOB = "tpf"
@classmethod
def invoke(cls, invoke_params):
"""
执行构建任务
:return:
"""
jenkins_hogwarts = Jenkins(cls.BASE_URL, cls.USERNAME, cls.PASSWORD)
# 获取Jenkins 的job 对象
job = jenkins_hogwarts.get_job(cls.JOB)
# 构建hogwarts job， 传入的值必须是字典， key 对应 jenkins 设置的参数名
# job.invoke(build_params={"task": "CalculatorProject/test/cases/test_div.py"})
job.invoke(build_params={"methods": invoke_params})
# 获取job 最后一次完成构建的编号
# http://47.92.149.0:8080/job/ck24/20/allure/
last_build_number = job.get_last_buildnumber()+1
# pytest 用例名称 指定报告生成地址
# pytest $task --alluredir=path
report = f"{cls.BASE_URL}job/{cls.JOB}/{last_build_number}/allure/"
return report
Server层创建-构建记录
class BuildService:
def list(self):
"""
查询所有的测试任务
:return:
"""
return build_dao.list()
def get(self, build_id):
"""
根据id查询构建
:return:
"""
return build_dao.get(build_id)
def save(self, plan_id):
"""
保存构建任务
:return:
"""
# 根据测试计划id获取自动化方法名
plan_detail = plan_service.get(plan_id)
method_list = [testcase.method for testcase in plan_detail.testcases]
methods = " ".join(method_list)
# 执行测试用例
report = JenkinsUtils.invoke(methods)
# 构建记录添加report信息
logger.info(f"report<====== {report}")
build_do = BuildDo(plan_id=plan_id, report=report)
# 创建构建记录
return build_dao.save(build_do)
def delete(self, build_id):
"""
存在才删除，反之返回错误
:param build_id: build_id
:return:
"""
plan = self.get(build_id)
if not plan:
return False
else:
build_dao.delete(build_id)
return build_id
实战4
完成测试计划、测试用例、构建记录的Service层代码
Controller层创建-测试用例
case_ns = Namespace("case", description="用例管理")
testcase_service = TestcaseService()
# 参数1传入model 的名称,
# 前端 -> controller -> service -> dao -> 数据库
@case_ns.route("")
class TestCaseServer(Resource):
get_paresr = api.parser()
# 查询接口， 可以传id ，也可以不传id，
# 不传id：就是返回 全部记录
# 传id：返回查询到的对应的记录，如果未查到则返回 空列表
get_paresr.add_argument("id", type=int, location="args")
@case_ns.expect(get_paresr)
def get(self):
"""
测试用例的查找
:return:
"""
case_id = request.args.get("id")
logger.info(f"type(request.args) <===== {type(request.args)}")
logger.info(f"接收到的参数 <===== {case_id}")
# Object of type TestcaseDo is not JSON serializable
# 问题：直接返回了对象，应该返回dict格式，后端才能给前端返回json
if case_id:
# 为了统一格式，当返回单个的时候，在外层嵌套一个列表
datas = [testcase_service.get(case_id).as_dict()]
else:
# 为空，返回所有记录
# testcase_service.list() -> List[TestcaseDo] -> [{"id": 1, "uid": "uid"}]
# case_list = []
# 遍历列表，获取testcase对象
# for testcase in testcase_service.list():
# 把testcase 对象转成 字典，添加到测试用例列表中
# case_list.append(testcase.as_dict())
# 以上3行，等同于下方的列表推导式
datas = [testcase.as_dict() for testcase in testcase_service.list()]
# return datas
return {"code": 0, "msg": "data success get", "data": datas}
post_model = case_ns.model("post_model", {
"name": fields.String,
"step": fields.String,
"method": fields.String,
"remark": fields.String
})
@case_ns.expect(post_model)
def post(self):
"""
测试用例的新增
:return:
"""
# 获取请求数据
case_data = request.json
logger.info(f"接收到的参数<====== {case_data}")
testcase = TestcaseDo(**request.json)
case_id = testcase_service.save(testcase)
if case_id:
return {"code": 0, "msg": f"case id {case_id} success add"}
else:
return {"code": 40001, "msg": "case is exists"}
put_model = case_ns.model("put_model", {
"id": fields.Integer,
"name": fields.String,
"step": fields.String,
"method": fields.String,
"remark": fields.String
})
@case_ns.expect(put_model)
def put(self):
"""
测试用例的修改
:return:
"""
case_data = request.json
logger.info(f"接收到的参数<====== {case_data}")
testcase = TestcaseDo(**request.json)
testcase_id = testcase_service.update(testcase)
if testcase_id:
return {"code": 0, "msg": f"case id {testcase_id} success update"}
else:
return {"code": 40002, "msg": "case is not exists"}
delete_parser = api.parser()
delete_parser.add_argument("id", type=int, location="json", required=True)
@case_ns.expect(delete_parser)
def delete(self):
"""
测试用例的删除
:return:
"""
case_data = request.json
case_id = case_data.get("id")
logger.info(f"接收到的参数id <====={case_id}")
testcase_id = testcase_service.delete(case_id)
if testcase_id:
return {"code": 0, "msg": f"case id {testcase_id} success delete"}
else:
return {"code": 40002, "msg": f"case is not exists"}
Controller层创建-测试计划
plan_ns = Namespace("plan", description="测试计划管理")
plan_service = PlanService()
# 参数1传入model 的名称,
plan_model = plan_ns.model("plan_model", {
"id": fields.Integer, "name": fields.String,
"testcase_ids": fields.List(fields.Integer)
})
# 前端 -> controller -> service -> dao -> 数据库
@plan_ns.route("")
class TestCaseServer(Resource):
get_paresr = api.parser()
# 查询接口， 可以传id ，也可以不传id，
# 不传id：就是返回 全部记录
# 传id：返回查询到的对应的记录，如果未查到则返回 空列表
get_paresr.add_argument("id", type=int, location="args")
@plan_ns.expect(get_paresr)
def get(self):
"""
测试计划的查找
:return:
"""
plan_id = request.args.get("id")
logger.info(f"type(request.args) <===== {type(request.args)}")
logger.info(f"接收到的参数 <===== {plan_id}")
# Object of type TestcaseDo is not JSON serializable
# 问题：直接返回了对象，应该返回dict格式，后端才能给前端返回json
if plan_id:
# 为了统一格式，当返回单个的时候，在外层嵌套一个列表
datas = [plan_service.get(plan_id).as_dict()]
else:
# 为空，返回所有记录
datas = [plan.as_dict() for plan in plan_service.list()]
return {"code": 0, "msg": "data success get", "data": datas}
@plan_ns.expect(plan_model)
def post(self):
"""
测试计划的新增
:return:
"""
# 获取请求数据
plan_data: dict = request.json
logger.info(f"接收到的参数<====== {plan_data}")
testcase_id_list = plan_data.pop("testcase_ids")
plan = PlanDo(**plan_data)
plan_id = plan_service.save(testcase_id_list, plan)
if plan_id:
return {"code": 0, "msg": f"plan id {plan_id} success add."}
else:
return {"code": 40001, "msg": "plan is exists"}
delete_parser = api.parser()
delete_parser.add_argument("id", type=int, location="json", required=True)
@plan_ns.expect(delete_parser)
def delete(self):
"""
测试计划的删除
:return:
"""
plan_id = request.json.get("id")
logger.info(f"接收到的参数id <====={plan_id}")
id = plan_service.delete(plan_id)
if id:
return {"code": 0, "msg": f"case id {id} success delete"}
else:
return {"code": 40002, "msg": f"case is not exists"}
Controller层创建-构建记录
build_ns = Namespace("build", description="构建记录管理")
build_service = BuildService()
build_model = build_ns.model("build_model", {
"plan_id": fields.Integer
})
# 参数1传入model 的名称,
# 前端 -> controller -> service -> dao -> 数据库
@build_ns.route("")
class TestCaseServer(Resource):
get_paresr = api.parser()
get_paresr.add_argument("plan_id", type=int, location="args")
@build_ns.expect(get_paresr)
def get(self):
"""
构建记录的查找
:return:
"""
# 查询接口， 可以传id ，也可以不传id，
# 传id：返回查询到的对应的记录，如果未查到则返回 空列表
# 不传id：就是返回 全部记录
build_id = request.args.get("id")
logger.info(f"接收到的plan_id参数 <===== {build_id}")
if build_id:
# List[BuildDo] -> List[dict]
datas = [data.as_dict() for data in build_service.get(build_id)]
else:
# 为空，返回所有记录
datas = [data.as_dict() for data in build_service.list()]
return {"code": 0, "msg": "data success get", "data": datas}
@build_ns.expect(build_model)
def post(self):
"""
构建记录的新增
:return:
"""
# 获取请求数据
plan_id = request.json.get("id")
logger.info(f"接收到的plan_id参数<====== {plan_id}")
build_id = build_service.save(plan_id)
if build_id:
return {"code": 0, "msg": f"build id {build_id} success add"}
else:
return {"code": 40001, "msg": "case is exists"}
delete_parser = api.parser()
delete_parser.add_argument("id", type=int, location="json", required=True)
@build_ns.expect(delete_parser)
def delete(self):
"""
测试报告的删除
:return:
"""
build_id = request.json.get("id")
logger.info(f"接收到的参数id <====={build_id}")
id = build_service.delete(build_id)
if id:
return {"code": 0, "msg": f"case id {id} success delete"}
else:
return {"code": 40002, "msg": f"case is not exists"}
实战5
完成测试计划、测试用例、构建记录的Controller层代码
最终实现效果
下节课要求
- 后端的所有录播
- 前端的所有录播
